liminal-server 0.24.9

//! Existing topology state and station facts remain pinned.

use super::*;

/// Lane brief ruling 5 (Hermes 7 Oct 03:51): a server with a persistent
/// store has a depot, `<device>.depot`, and a durable line passes it after
/// its publishers and before its subscribers, because the append comes
/// before the fan-out. A passing line never stops there, and a server with
/// no persistent store lists no depot.
#[test]
fn a_durable_line_passes_the_depot_before_its_subscribers() -> TestResult {
    let started = Instant::now();
    let with_depot = |depot| {
        let mut model = TopologyModel::new(
            ThisDevice {
                id: "local".to_owned(),
                liminal: "1.2.3".to_owned(),
                started,
                depot,
            },
            [],
            [
                ("orders".to_owned(), LineLife::Kept, LineState::Live),
                ("audit".to_owned(), LineLife::Passing, LineState::Live),
            ],
            [],
            [],
        );
        for (connection, identity) in [(11, "ping"), (12, "pong")] {
            model.apply(TopologyChange::Worker {
                connection,
                identity: identity.to_owned(),
                subjects: Vec::new(),
                activities: Vec::new(),
            });
        }
        for line in ["orders", "audit"] {
            for (connection, role) in [(11, FlowRole::Publishes), (12, FlowRole::Subscribes)] {
                model.apply(TopologyChange::Flow {
                    connection,
                    line: line.to_owned(),
                    role,
                    present: true,
                });
            }
        }
        model
    };
    let stops = |value: &serde_json::Value, line: &str| {
        value["lines"]
            .as_array()
            .and_then(|lines| lines.iter().find(|entry| entry["id"] == line))
            .map(|entry| entry["stops"].clone())
    };
    let kept = document(&with_depot(true), started)?;
    assert_eq!(
        stops(&kept, "orders"),
        Some(serde_json::json!(["ping", "local.depot", "local", "pong"]))
    );
    assert_eq!(
        stops(&kept, "audit"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    let depot = kept["stations"]
        .as_array()
        .and_then(|stations| {
            stations
                .iter()
                .find(|station| station["id"] == "local.depot")
        })
        .ok_or_else(|| format!("no depot station: {kept}"))?;
    assert_eq!(
        (&depot["kind"], &depot["device"]),
        (&serde_json::json!("depot"), &serde_json::json!("local"))
    );

    let memory = document(&with_depot(false), started)?;
    assert_eq!(
        stops(&memory, "orders"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    assert!(
        !memory.to_string().contains("local.depot"),
        "no persistent store, no depot: {memory}"
    );
    Ok(())
}

/// Lane brief ruling 2 (Hermes 7 Oct 03:51): this server's exchange is a
/// station whose id is the device's own, of kind `exchange`, and a stop on
/// every line exactly once: after the publishers and the depot, before the
/// subscribers, and alone on a line with no flows.
#[test]
fn every_line_passes_the_exchange_once() -> TestResult {
    let started = Instant::now();
    let mut model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: true,
        },
        [],
        [
            ("orders".to_owned(), LineLife::Kept, LineState::Live),
            ("audit".to_owned(), LineLife::Passing, LineState::Live),
            ("idle".to_owned(), LineLife::Passing, LineState::Live),
        ],
        [],
        [],
    );
    for (connection, identity) in [(11, "ping"), (12, "pong")] {
        model.apply(TopologyChange::Worker {
            connection,
            identity: identity.to_owned(),
            subjects: Vec::new(),
            activities: Vec::new(),
        });
    }
    for line in ["orders", "audit"] {
        for (connection, role) in [(11, FlowRole::Publishes), (12, FlowRole::Subscribes)] {
            model.apply(TopologyChange::Flow {
                connection,
                line: line.to_owned(),
                role,
                present: true,
            });
        }
    }
    let value = document(&model, started)?;
    let stops = |line: &str| {
        value["lines"]
            .as_array()
            .and_then(|lines| lines.iter().find(|entry| entry["id"] == line))
            .map(|entry| entry["stops"].clone())
    };
    assert_eq!(
        stops("orders"),
        Some(serde_json::json!(["ping", "local.depot", "local", "pong"]))
    );
    assert_eq!(
        stops("audit"),
        Some(serde_json::json!(["ping", "local", "pong"]))
    );
    assert_eq!(stops("idle"), Some(serde_json::json!(["local"])));
    let exchange = value["stations"]
        .as_array()
        .and_then(|stations| stations.iter().find(|station| station["id"] == "local"))
        .ok_or_else(|| format!("no exchange station: {value}"))?;
    assert_eq!(
        (&exchange["kind"], &exchange["device"]),
        (&serde_json::json!("exchange"), &serde_json::json!("local"))
    );
    Ok(())
}

/// SRV-027 R3: a worker's station carries its subjects and its activities
/// with the schema strings exactly as registered (nested objects and
/// non-ASCII descriptions untouched); a worker that declared none is raw
/// bytes, `activities: []`; and the station leaves with its connection.
#[test]
fn worker_station_carries_its_subjects_and_activities_exactly() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let input = r#"{"type":"object","properties":{"items":{"type":"array","items":{"type":"object","properties":{"text":{"type":"string","description":"Texte à plonger"}}}}}}"#;
    let output = r#"{"type":"object","properties":{"vectors":{"type":"array","description":"Vecteurs, 384 dimensions"}}}"#;
    assert!(model.apply(TopologyChange::Worker {
        connection: 21,
        identity: "tessera-embed".to_owned(),
        subjects: vec!["tessera.embed.request".to_owned()],
        activities: vec![crate::console::records::StationActivity {
            name: "tessera.embed.request".to_owned(),
            input_schema_json: input.to_owned(),
            output_schema_json: output.to_owned(),
        }],
    }));
    assert!(model.apply(TopologyChange::Worker {
        connection: 22,
        identity: "raw-worker".to_owned(),
        subjects: vec!["raw.request".to_owned()],
        activities: Vec::new(),
    }));
    let with_workers = document(&model, started)?;
    let stations = with_workers["stations"].as_array().ok_or("no stations")?;
    let typed = stations
        .iter()
        .find(|station| station["id"] == "tessera-embed")
        .ok_or("the typed worker has no station")?;
    assert_eq!(
        typed["subjects"],
        serde_json::json!(["tessera.embed.request"])
    );
    assert_eq!(typed["activities"][0]["name"], "tessera.embed.request");
    assert_eq!(typed["activities"][0]["input_schema_json"], input);
    assert_eq!(typed["activities"][0]["output_schema_json"], output);
    let raw = stations
        .iter()
        .find(|station| station["id"] == "raw-worker")
        .ok_or("the raw worker has no station")?;
    assert_eq!(raw["subjects"], serde_json::json!(["raw.request"]));
    assert_eq!(raw["activities"], serde_json::json!([]));
    let exchange = stations
        .iter()
        .find(|station| station["id"] == "local")
        .ok_or("no exchange")?;
    assert!(exchange.get("activities").is_none());
    assert!(model.apply(TopologyChange::ConnectionClosed { id: 21 }));
    let after_close = document(&model, started)?;
    assert!(
        after_close["stations"]
            .as_array()
            .ok_or("no stations")?
            .iter()
            .all(|station| station["id"] != "tessera-embed")
    );
    Ok(())
}

#[test]
fn a_district_names_its_parent_device() -> TestResult {
    let started = Instant::now();
    for parent in [None, Some("holder")] {
        let mut source = "identity_path = '/identity'\npass_path = '/pass'\n".to_owned();
        if let Some(parent) = parent {
            source.extend(["parent = '", parent, "'\n"]);
        }
        let machine = toml::from_str::<crate::config::types::MachineConfig>(&source)?;
        let mut model = model(started).with_machine(&machine);
        model.apply(TopologyChange::Peer {
            name: "edge".to_owned(),
            present: true,
        });
        let value = document(&model, started)?;
        assert_eq!(
            value["devices"][0]
                .get("parent")
                .and_then(serde_json::Value::as_str),
            parent
        );
        assert!(value["devices"][1].get("parent").is_none());
    }
    Ok(())
}

#[test]
fn a_link_change_applies_once_and_a_repeat_is_not_a_change() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    model.apply(TopologyChange::Peer {
        name: "cluster".to_owned(),
        present: true,
    });
    let before = document(&model, started)?;
    let change = |carrier, state, reason, counters| TopologyChange::Link {
        link_id: "local~edge".to_owned(),
        peer: "edge".to_owned(),
        carrier,
        state,
        reason,
        counters,
    };
    let mut rev = model.rev();
    for event in [
        change(
            Some(Carrier::Relay("hub".to_owned())),
            LinkState::Up,
            None,
            LinkCounters::default(),
        ),
        change(
            Some(Carrier::Direct),
            LinkState::Up,
            None,
            LinkCounters {
                crossed: Some(9),
                refused: Some(0),
                ..LinkCounters::default()
            },
        ),
        change(
            Some(Carrier::Direct),
            LinkState::Down,
            Some("revoked".to_owned()),
            LinkCounters {
                crossed: Some(9),
                refused: Some(1),
                ..LinkCounters::default()
            },
        ),
    ] {
        assert!(model.apply(event.clone()));
        rev += 1;
        assert_eq!(model.rev(), rev);
        let applied = document(&model, started)?;
        assert!(!model.apply(event));
        assert_eq!(model.rev(), rev);
        assert_eq!(document(&model, started)?, applied);
        assert_eq!(applied["links"].as_array().map(Vec::len), Some(2));
        assert_eq!(applied["links"][0], before["links"][0]);
        let link = &applied["links"][1];
        assert_eq!(link["id"], "local~edge");
        assert_eq!(link["ends"], serde_json::json!(["edge", "local"]));
        assert_eq!(link["kind"], "federated");
        assert!(
            applied["devices"]
                .as_array()
                .ok_or("devices")?
                .iter()
                .any(|device| device["id"] == "edge")
        );
    }
    let down = document(&model, started)?;
    assert_eq!(down["links"][1]["state"], "down");
    assert_eq!(down["links"][1]["reason"], "revoked");
    assert!(down["links"][1].get("relayed").is_none());
    assert_eq!(down["links"][1]["crossed"], 9);
    assert_eq!(down["links"][1]["refused"], 1);
    Ok(())
}