clustri/src-tauri/tests/cluster_flow.rs

//! Integration tests for the connect-time node discovery, same-cluster
//! merging, and the connection status reporting flow.
//!
//! `connect` now authenticates, resets failover state, and discovers the
//! cluster's nodes before either folding the connection into an existing
//! same-cluster connection or marking it connected. These tests cover the
//! discovered-node persistence, the merge of a second node's connection into
//! the first, the status reporting (including failover and disconnected), and
//! the all-endpoints-down case where the connect reports `"failed"` instead of
//! erroring.

use httpmock::prelude::*;
use httpmock::Mock;
use clustri::{ConnectionConfig, ConnectionManager, EndpointConfig, Error};

const TOKEN: &str = "root@pam!cluster-token";

/// A minimal `/nodes` entry. The `Node` struct requires every field, so all
/// are included.
fn node_json(name: &str, status: &str) -> serde_json::Value {
    serde_json::json!({
        "node": name,
        "status": status,
        "cpu": 0.0,
        "maxcpu": 8,
        "mem": 0,
        "maxmem": 34359738368u64,
        "disk": 0,
        "maxdisk": 536870912000u64,
        "uptime": 0,
        "level": "",
        "id": format!("node/{}", name),
        "type": "node"
    })
}

/// Builds the `/cluster/status` response body: one `cluster` entry followed by
/// the given node entries.
fn cluster_status_body(
    cluster_id: &str,
    cluster_name: &str,
    nodes: &[serde_json::Value],
) -> String {
    let mut entries = Vec::with_capacity(nodes.len() + 1);
    entries.push(serde_json::json!({
        "type": "cluster",
        "id": cluster_id,
        "name": cluster_name,
        "nodes": nodes.len()
    }));
    entries.extend_from_slice(nodes);
    serde_json::json!({ "data": entries }).to_string()
}

fn stub_version<'a>(server: &'a MockServer) -> Mock<'a> {
    server.mock(|when, then| {
        when.method(GET).path("/api2/json/version");
        then.status(200)
            .header("content-type", "application/json")
            .body(r#"{"data":{"version":"8.2"}}"#);
    })
}

fn stub_nodes<'a>(server: &'a MockServer, nodes: &[serde_json::Value]) -> Mock<'a> {
    server.mock(|when, then| {
        when.method(GET).path("/api2/json/nodes");
        then.status(200)
            .header("content-type", "application/json")
            .body(serde_json::json!({ "data": nodes }).to_string());
    })
}

fn stub_cluster_status<'a>(
    server: &'a MockServer,
    cluster_id: &str,
    cluster_name: &str,
    nodes: &[serde_json::Value],
) -> Mock<'a> {
    server.mock(|when, then| {
        when.method(GET).path("/api2/json/cluster/status");
        then.status(200)
            .header("content-type", "application/json")
            .body(cluster_status_body(cluster_id, cluster_name, nodes));
    })
}

/// Returns a URL pointing at a closed TCP port: a listener is bound to an
/// ephemeral port, its address is captured, and the listener is dropped so any
/// subsequent connection attempt is refused.
fn closed_port_url() -> String {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let addr = listener.local_addr().expect("address should be available");
    drop(listener);
    format!("http://{}", addr)
}

fn fallback_config(url: &str) -> EndpointConfig {
    EndpointConfig {
        url: url.to_string(),
        node: None,
        token: Some(TOKEN.to_string()),
    }
}

/// Builds a token-mode connection config that skips certificate verification
/// (the mock servers speak plain HTTP).
fn token_config(id: &str, url: &str, fallbacks: Vec<EndpointConfig>) -> ConnectionConfig {
    ConnectionConfig {
        id: id.to_string(),
        name: id.to_string(),
        primary: EndpointConfig {
            url: url.to_string(),
            node: None,
            token: Some(TOKEN.to_string()),
        },
        fallbacks,
        cert_fingerprint: None,
        trusted: false,
        accept_untrusted: true,
        status: "disconnected".to_string(),
        cluster_name: None,
        is_cluster: false,
        auth_mode: "token".to_string(),
        username: None,
        nodes: vec![],
        cluster_id: None,
        server_type: "pve".to_string(),
    }
}

async fn add_connection(
    manager: &mut ConnectionManager,
    path: &std::path::Path,
    config: ConnectionConfig,
) {
    manager
        .add_connection(config, path)
        .await
        .expect("connection should be added");
}

#[tokio::test]
async fn connect_discovers_and_persists_nodes() {
    let server = MockServer::start();
    let port = server.port();

    stub_version(&server);
    stub_nodes(
        &server,
        &[node_json("pve1", "online"), node_json("pve2", "online")],
    );
    stub_cluster_status(
        &server,
        "cluster/lab",
        "lab",
        &[
            serde_json::json!({"type": "node", "id": "node/pve1", "nodeid": 1,
                               "online": 1, "local": 1, "ip": "10.0.0.5"}),
            serde_json::json!({"type": "node", "id": "node/pve2", "nodeid": 2,
                               "online": 1, "local": 0, "ip": "10.0.0.6"}),
        ],
    );

    let dir = tempfile::tempdir().expect("temp dir should be created");
    let path = dir.path().join("connections.json");
    let mut manager = ConnectionManager::new();
    add_connection(
        &mut manager,
        &path,
        token_config("conn", &server.base_url(), vec![]),
    )
    .await;

    let result = manager
        .connect("conn", &path)
        .await
        .expect("connect should succeed");
    assert_eq!(result.status, "connected");
    assert_eq!(result.merged_into, None);
    assert_eq!(result.connection_id, "conn");

    let config = manager
        .connection_config("conn")
        .expect("config should be readable");
    assert_eq!(config.nodes.len(), 2);
    assert_eq!(config.cluster_id.as_deref(), Some("cluster/lab"));
    assert_eq!(config.primary.node.as_deref(), Some("pve1"));

    // The discovered nodes and cluster identity are persisted.
    let raw = std::fs::read_to_string(&path).expect("file should be written");
    let json: serde_json::Value = serde_json::from_str(&raw).expect("file should be valid JSON");
    let conn_json = &json["connections"][0];
    assert_eq!(conn_json["clusterId"], "cluster/lab");
    assert_eq!(
        conn_json["nodes"]
            .as_array()
            .expect("nodes should be present")
            .len(),
        2
    );
    assert_eq!(
        conn_json["nodes"][0]["url"],
        format!("http://10.0.0.5:{}", port),
        "the primary node's URL must be persisted"
    );
    assert_eq!(conn_json["status"], "connected");
}

#[tokio::test]
async fn connect_second_node_merges_into_existing() {
    let server_a = MockServer::start();
    let server_b = MockServer::start();

    // Both servers report the same cluster identity, as two nodes of one
    // cluster would.
    for (server, ip1, ip2) in [
        (&server_a, "10.0.0.5", "10.0.0.6"),
        (&server_b, "10.0.0.7", "10.0.0.8"),
    ] {
        stub_version(server);
        stub_nodes(
            server,
            &[node_json("pve1", "online"), node_json("pve2", "online")],
        );
        stub_cluster_status(
            server,
            "cluster/lab",
            "lab",
            &[
                serde_json::json!({"type": "node", "id": "node/pve1", "nodeid": 1,
                                   "online": 1, "local": 1, "ip": ip1}),
                serde_json::json!({"type": "node", "id": "node/pve2", "nodeid": 2,
                                   "online": 1, "local": 0, "ip": ip2}),
            ],
        );
    }
    let vms_mock = server_a.mock(|when, then| {
        when.method(GET)
            .path("/api2/json/cluster/resources")
            .query_param("type", "vm");
        then.status(200)
            .header("content-type", "application/json")
            .body(
                serde_json::json!({
                    "data": [{"vmid": 100, "name": "web01", "type": "qemu",
                              "status": "running", "node": "pve1"}]
                })
                .to_string(),
            );
    });

    let dir = tempfile::tempdir().expect("temp dir should be created");
    let path = dir.path().join("connections.json");
    let mut manager = ConnectionManager::new();
    add_connection(
        &mut manager,
        &path,
        token_config("conn-a", &server_a.base_url(), vec![]),
    )
    .await;
    add_connection(
        &mut manager,
        &path,
        token_config("conn-b", &server_b.base_url(), vec![]),
    )
    .await;

    let first = manager
        .connect("conn-a", &path)
        .await
        .expect("first connect should succeed");
    assert_eq!(first.status, "connected");
    assert_eq!(first.merged_into, None);

    let second = manager
        .connect("conn-b", &path)
        .await
        .expect("second connect should succeed");
    assert_eq!(second.status, "connected");
    assert_eq!(second.connection_id, "conn-a");
    assert_eq!(second.merged_into.as_deref(), Some("conn-a"));

    // The merged connection is gone and conn-a carries B's endpoint.
    assert!(
        matches!(
            manager.connection_config("conn-b"),
            Err(Error::ConnectionNotFound(ref id)) if id == "conn-b"
        ),
        "conn-b must be removed after merging"
    );
    let config_a = manager
        .connection_config("conn-a")
        .expect("config should be readable");
    assert!(
        config_a
            .fallbacks
            .iter()
            .any(|endpoint| endpoint.url == server_b.base_url()),
        "conn-a must have B's URL as a fallback"
    );
    assert_eq!(
        config_a.nodes.len(),
        4,
        "the node lists of both servers must be merged"
    );

    // Only the surviving connection is persisted.
    let raw = std::fs::read_to_string(&path).expect("file should be written");
    let json: serde_json::Value = serde_json::from_str(&raw).expect("file should be valid JSON");
    let ids: Vec<&str> = json["connections"]
        .as_array()
        .expect("connections should be an array")
        .iter()
        .filter_map(|c| c["id"].as_str())
        .collect();
    assert_eq!(ids, vec!["conn-a"]);

    assert_eq!(
        manager
            .runtime_status("conn-a")
            .expect("status should be readable"),
        "connected"
    );

    // A subsequent data request on the merged connection is served by A.
    let vms = manager
        .get_vms("conn-a")
        .await
        .expect("get_vms should succeed via A");
    assert_eq!(vms.len(), 1);
    vms_mock.assert();
}

#[tokio::test]
async fn status_info_reports_failover() {
    let fallback = MockServer::start();
    stub_version(&fallback);
    stub_nodes(&fallback, &[node_json("pve1", "online")]);
    stub_cluster_status(
        &fallback,
        "cluster/lab",
        "lab",
        &[
            serde_json::json!({"type": "node", "id": "node/pve1", "nodeid": 1,
                             "online": 1, "local": 1, "ip": "10.0.0.5"}),
        ],
    );
    let vms_mock = fallback.mock(|when, then| {
        when.method(GET)
            .path("/api2/json/cluster/resources")
            .query_param("type", "vm");
        then.status(200)
            .header("content-type", "application/json")
            .body(
                serde_json::json!({
                    "data": [{"vmid": 100, "name": "web01", "type": "qemu",
                              "status": "running", "node": "pve1"}]
                })
                .to_string(),
            );
    });

    let primary_url = closed_port_url();
    let dir = tempfile::tempdir().expect("temp dir should be created");
    let path = dir.path().join("connections.json");
    let mut manager = ConnectionManager::new();
    add_connection(
        &mut manager,
        &path,
        token_config(
            "conn",
            &primary_url,
            vec![fallback_config(&fallback.base_url())],
        ),
    )
    .await;

    let result = manager
        .connect("conn", &path)
        .await
        .expect("connect should succeed via the fallback");
    assert_eq!(result.status, "connected");

    let vms = manager
        .get_vms("conn")
        .await
        .expect("get_vms should succeed via the fallback");
    assert_eq!(vms.len(), 1);
    assert_eq!(
        manager
            .runtime_status("conn")
            .expect("status should be readable"),
        "failover"
    );

    let info = manager
        .status_info("conn")
        .await
        .expect("status info should be readable");
    assert_eq!(info.status, "failover");
    assert_eq!(info.current_endpoint_url, fallback.base_url());
    assert_eq!(info.primary_url, primary_url);
    vms_mock.assert();
}

#[tokio::test]
async fn status_info_when_disconnected() {
    let server = MockServer::start();
    let version_mock = stub_version(&server);

    let dir = tempfile::tempdir().expect("temp dir should be created");
    let path = dir.path().join("connections.json");
    let mut manager = ConnectionManager::new();
    add_connection(
        &mut manager,
        &path,
        token_config("conn", &server.base_url(), vec![]),
    )
    .await;

    let info = manager
        .status_info("conn")
        .await
        .expect("status info should be readable");
    assert_eq!(info.status, "disconnected");
    assert_eq!(info.primary_url, server.base_url());
    assert_eq!(info.current_endpoint_url, server.base_url());
    assert!(info.nodes.is_empty());
    assert_eq!(
        version_mock.calls(),
        0,
        "a disconnected connection must not hit the network"
    );
}

#[tokio::test]
async fn connect_all_endpoints_down_returns_failed_result() {
    let primary_url = closed_port_url();
    let fallback_url = closed_port_url();

    let dir = tempfile::tempdir().expect("temp dir should be created");
    let path = dir.path().join("connections.json");
    let mut manager = ConnectionManager::new();
    add_connection(
        &mut manager,
        &path,
        token_config("conn", &primary_url, vec![fallback_config(&fallback_url)]),
    )
    .await;

    let result = manager
        .connect("conn", &path)
        .await
        .expect("an unreachable cluster must be reported, not errored");
    assert_eq!(result.status, "failed");
    assert_eq!(result.merged_into, None);
    assert_eq!(result.connection_id, "conn");

    let config = manager
        .connection_config("conn")
        .expect("config should be readable");
    assert_eq!(config.status, "failed");

    // The failed status is persisted so the connection stays tracked.
    let raw = std::fs::read_to_string(&path).expect("file should be written");
    let json: serde_json::Value = serde_json::from_str(&raw).expect("file should be valid JSON");
    assert_eq!(json["connections"][0]["status"], "failed");
}