liminal-server 0.24.5

//! Existing topology and carrier record shapes remain pinned.

use super::*;

/// Lane brief ruling 2 (Hermes 7 Oct 03:51), on the golden bytes the console
/// reads: every line's stops hold this device's exchange exactly once, and
/// the exchange is a listed station of kind `exchange` on this device.
#[test]
fn every_golden_line_passes_the_exchange_once() -> TestResult {
    let topology: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    let this = topology["this_device"]
        .as_str()
        .ok_or("the golden names no device")?;
    let lines = topology["lines"].as_array().ok_or("no lines")?;
    assert!(!lines.is_empty());
    for line in lines {
        let stops = line["stops"].as_array().ok_or("a line with no stops")?;
        let exchanges = stops
            .iter()
            .filter(|stop| stop.as_str() == Some(this))
            .count();
        assert_eq!(exchanges, 1, "{line}");
    }
    let exchange = topology["stations"]
        .as_array()
        .and_then(|stations| stations.iter().find(|station| station["id"] == this))
        .ok_or("the exchange is not a listed station")?;
    assert_eq!(
        (&exchange["kind"], &exchange["device"]),
        (&serde_json::json!("exchange"), &topology["this_device"])
    );
    Ok(())
}

/// Lane brief rulings 3, 4 and 8 (Hermes 7 Oct 03:51 to 03:57), on the
/// golden bytes the console reads: every tap hop names a listed station; a
/// message's hops never go back in time; a durable message passes the depot
/// before the exchange and reaches its receivers after it; a message that
/// appended nothing has no exchange hop (Jules's F1, Hermes 07:22); and no
/// hop carries `dwell_ms` (dwell waits, so the console's delay reading stays
/// off).
#[test]
fn every_golden_hop_names_a_station_in_order_without_dwell() -> TestResult {
    let topology: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    let batch: serde_json::Value = serde_json::from_slice(TAP_BATCH_GOLDEN)?;
    let stations: Vec<&str> = topology["stations"]
        .as_array()
        .ok_or("no stations")?
        .iter()
        .filter_map(|station| station["id"].as_str())
        .collect();
    let this = topology["this_device"].as_str().ok_or("no device")?;
    let depot = format!("{this}.depot");
    let messages = batch["messages"].as_array().ok_or("no messages")?;
    let mut durable = 0;
    let mut unmeasured = 0;
    for message in messages {
        let hops = message["hops"].as_array().ok_or("no hops")?;
        let nodes: Vec<&str> = hops.iter().filter_map(|hop| hop["node"].as_str()).collect();
        assert_eq!(nodes.len(), hops.len(), "{message}");
        for node in &nodes {
            assert!(stations.contains(node), "{node} is not a listed station");
        }
        let times: Vec<u64> = hops
            .iter()
            .filter_map(|hop| hop["at_ms"].as_u64())
            .collect();
        assert_eq!(times.len(), hops.len(), "{message}");
        assert!(
            times.is_sorted(),
            "{message}: a hop before the one ahead of it"
        );
        assert!(
            hops.iter().all(|hop| hop.get("dwell_ms").is_none()),
            "{message}"
        );
        let exchange = nodes.iter().position(|node| *node == this);
        assert!(nodes.iter().filter(|node| **node == this).count() <= 1);
        unmeasured += usize::from(exchange.is_none());
        if let Some(at) = nodes.iter().position(|node| *node == depot) {
            durable += 1;
            assert_eq!(
                Some(at + 1),
                exchange,
                "{message}: the depot, then the exchange"
            );
        }
        let receivers = message["to_stations"].as_array().map_or(0, Vec::len);
        let publisher = usize::from(message.get("from_station").is_some());
        assert_eq!(
            exchange.map_or(nodes.len() - publisher, |at| nodes.len() - at - 1),
            receivers,
            "{message}: each receiver after the exchange, or after the publisher"
        );
    }
    assert!(durable > 0, "a durable golden message passes the depot");
    assert!(messages.len() > durable, "and another does not");
    assert!(
        unmeasured > 0,
        "a golden message whose exchange nothing measured"
    );
    Ok(())
}

/// NET-004 R1 keeps documents without links byte-stable. NET-009 amendment 2
/// adds carrier refusals to links; its unspecified empty case stays an array.
#[test]
fn existing_canonical_bytes_are_unchanged_without_links() -> TestResult {
    let mut record = topology();
    record.links.clear();
    let mut golden: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    golden["links"] = serde_json::json!([]);
    let encoded = canonical_bytes(&record, "topology")?;
    assert_eq!(encoded, liminal::channel::canonical::to_vec(golden)?);
    assert_eq!(encoded, normalized(&encoded)?);
    assert_eq!(canonical_bytes(&topology(), "topology")?, TOPOLOGY_GOLDEN);
    Ok(())
}

/// NET-004 R1 names the carrier, relay and down reason. NET-009 amendment 2
/// retains the refusal names and identifies the device carrying a relay.
/// SRV-012 R10 sorts every canonical field by its spelling.
#[test]
fn a_federated_link_serialises_its_carrier_relay_and_reason() -> TestResult {
    for (carrier, word, relayed) in [
        (Carrier::Wire, "wire", None),
        (Carrier::Direct, "direct", None),
        (Carrier::Relay("hub".to_owned()), "relay", Some("hub")),
    ] {
        let link = Link::federated(
            "local",
            "edge",
            Some(carrier),
            LinkState::Down,
            Some("peer gone".to_owned()),
            LinkCounters {
                crossed: Some(7),
                refused: Some(0),
                ..LinkCounters::default()
            },
        );
        let encoded = canonical_bytes(&link, "link")?;
        let expected = serde_json::json!({
            "id": "local~edge", "name": "edge", "kind": "federated", "state": "down",
            "rtt_ms": null, "ends": ["edge", "local"], "carrier": word,
            "reason": "peer gone", "crossed": 7, "refused": 0,
            "carrier_refusals": ["peer gone"],
        });
        let mut expected = expected;
        if let Some(relayed) = relayed {
            expected["relayed"] = serde_json::json!(relayed);
            expected["carried_by"] = serde_json::json!(relayed);
        }
        assert_eq!(encoded, liminal::channel::canonical::to_vec(expected)?);
        assert_eq!(encoded, normalized(&encoded)?);
    }
    Ok(())
}

#[test]
fn a_down_link_carries_its_reason_and_an_up_one_none() -> TestResult {
    for (state, reason) in [(LinkState::Down, Some("revoked")), (LinkState::Up, None)] {
        let link = Link::federated(
            "edge",
            "local",
            None,
            state,
            Some("revoked".to_owned()),
            LinkCounters::default(),
        );
        let value = serde_json::to_value(&link)?;
        assert_eq!(
            value["state"],
            if state == LinkState::Down {
                "down"
            } else {
                "up"
            }
        );
        assert_eq!(
            value.get("reason").and_then(serde_json::Value::as_str),
            reason
        );
        for field in ["carrier", "relayed", "crossed", "refused"] {
            assert!(value.get(field).is_none(), "{field}: unknown is omitted");
        }
    }
    Ok(())
}