use bytes::Bytes;
use std::collections::BTreeMap;
use unb_core::{
Envelope, NodeCore, Resolution, RouteAdvertisement, RouteError, RouteSnapshot, RouteWithdrawal,
};
#[test]
fn route_control_serializes_destination_nodes_without_subject_keys() {
let advertisement = RouteAdvertisement {
destination: "node-c".into(),
owner: "node-c".into(),
owner_instance: "node-c-1".into(),
owner_epoch: 1,
owner_revision: 0,
distance: 1,
path: vec!["node-c".into(), "node-b".into()],
};
let withdrawal = RouteWithdrawal {
destination: "node-c".into(),
owner: "node-c".into(),
owner_instance: "node-c-1".into(),
owner_epoch: 1,
owner_revision: 0,
};
let advertisement_json = serde_json::to_value(&advertisement).unwrap();
let withdrawal_json = serde_json::to_value(&withdrawal).unwrap();
assert_eq!(advertisement_json["destination"], "node-c");
assert_eq!(withdrawal_json["destination"], "node-c");
assert!(advertisement_json.get("subject").is_none());
assert!(withdrawal_json.get("subject").is_none());
let mut old_shape = advertisement_json;
old_shape["subject"] = old_shape["destination"].take();
old_shape.as_object_mut().unwrap().remove("destination");
assert!(serde_json::from_value::<RouteAdvertisement>(old_shape).is_err());
}
fn route(destination: &str, _advertiser: &str, path: &[&str]) -> RouteAdvertisement {
RouteAdvertisement {
destination: destination.into(),
owner: destination.into(),
owner_instance: format!("{destination}-1"),
owner_epoch: 1,
owner_revision: 0,
distance: (path.len() - 1) as u32,
path: path.iter().map(|hop| (*hop).to_owned()).collect(),
}
}
#[test]
fn freshness_selection_split_horizon_and_exact_session_removal_are_preserved() {
let mut core = NodeCore::new("node-a");
let through_b = route("node-d", "node-b", &["node-d", "node-b"]);
let through_c = route("node-d", "node-c", &["node-d", "node-c"]);
core.apply_snapshot(
"session-b",
"node-b",
&RouteSnapshot::canonical(1, vec![through_b]),
)
.unwrap();
core.apply_snapshot(
"session-c",
"node-c",
&RouteSnapshot::canonical(1, vec![through_c]),
)
.unwrap();
assert_eq!(core.resolve("node-d"), Resolution::Route("node-b".into()));
assert!(core
.export_for("node-b")
.iter()
.all(|route| route.destination != "node-d"));
let mut fresher = route("node-d", "node-c", &["node-d", "node-x", "node-c"]);
fresher.owner_epoch = 2;
core.apply_snapshot(
"session-c",
"node-c",
&RouteSnapshot::canonical(2, vec![fresher]),
)
.unwrap();
assert_eq!(core.resolve("node-d"), Resolution::Route("node-c".into()));
core.leave("session-c");
assert_eq!(core.resolve("node-d"), Resolution::Route("node-b".into()));
}
#[test]
fn route_owner_and_path_must_identify_the_destination_safely() {
let mut core = NodeCore::new("node-a");
let mut mismatched = route("node-d", "node-b", &["node-d", "node-b"]);
mismatched.owner = "node-x".into();
assert!(matches!(
core.apply_snapshot(
"session-b",
"node-b",
&RouteSnapshot::canonical(1, vec![mismatched])
),
Err(RouteError::InvalidAdvertisement { .. })
));
let unsafe_path = route("node/d", "node-b", &["node/d", "node-b"]);
assert!(matches!(
core.apply_snapshot(
"session-b",
"node-b",
&RouteSnapshot::canonical(1, vec![unsafe_path])
),
Err(RouteError::InvalidAdvertisement { .. })
));
}
#[test]
fn route_table_resolves_local_and_transit_destination_nodes() {
let mut core = NodeCore::new("node-a");
assert_eq!(core.resolve("node-a"), Resolution::Local);
assert_eq!(core.resolve("weather"), Resolution::Unknown);
let snapshot =
RouteSnapshot::canonical(1, vec![route("node-c", "node-b", &["node-c", "node-b"])]);
assert_eq!(
core.apply_snapshot("session-b", "node-b", &snapshot)
.unwrap(),
vec!["node-c"]
);
assert_eq!(core.resolve("node-c"), Resolution::Route("node-b".into()));
assert!(core
.apply_snapshot("session-b", "node-b", &snapshot)
.unwrap()
.is_empty());
let conflicting =
RouteSnapshot::canonical(1, vec![route("node-d", "node-b", &["node-d", "node-b"])]);
assert!(matches!(
core.apply_snapshot("session-b", "node-b", &conflicting),
Err(RouteError::GenerationConflict { .. })
));
}
#[test]
fn node_core_exports_and_forwards_destination_nodes_not_features() {
let mut core = NodeCore::new("node-a");
core.install_local_capabilities(BTreeMap::from([
("weather".into(), serde_json::json!({})),
("payments.charge".into(), serde_json::json!({})),
]));
let local_export = core.export_for("node-b");
assert_eq!(local_export.len(), 1);
assert_eq!(local_export[0].destination, "node-a");
core.apply_snapshot(
"session-b",
"node-b",
&RouteSnapshot::canonical(1, vec![route("node-c", "node-b", &["node-c", "node-b"])]),
)
.unwrap();
assert_eq!(core.reachable_names(), ["node-a", "node-c"]);
let request = http::Request::post("/node-c/weather/current")
.body(Bytes::new())
.unwrap();
let envelope = Envelope::from_request(request).unwrap();
let (peer, forwarded) = core.forward(envelope).unwrap();
assert_eq!(peer, "node-b");
assert_eq!(forwarded.target, "node-c");
assert_eq!(forwarded.subject, "weather.current");
assert_eq!(forwarded.hops, Some(unb_core::DEFAULT_HOPS - 1));
}
#[test]
fn catalog_mutation_changes_catalog_identity_without_route_churn() {
let mut core = NodeCore::new("node-a");
let routes_before = core.export_for("node-b");
let fingerprint_before = core.fingerprint();
let revision_before = core.catalog_revision();
assert!(core.install_local_capabilities(BTreeMap::from([(
"weather.current".into(),
serde_json::json!({ "one_line": "Current weather" }),
)])));
assert_ne!(core.fingerprint(), fingerprint_before);
assert_eq!(core.catalog_revision(), revision_before + 1);
assert_eq!(core.export_for("node-b"), routes_before);
}
#[test]
fn route_state_scales_with_nodes_and_forwarding_uses_only_the_target() {
let mut core = NodeCore::new("node-a");
let capabilities = (0..4_096)
.map(|index| (format!("feature.{index}"), serde_json::json!({})))
.collect::<BTreeMap<_, _>>();
core.install_local_capabilities(capabilities);
let routes = (0..256)
.map(|index| {
let destination = format!("node-{index:04}");
route(&destination, "hub", &[&destination, "hub"])
})
.collect();
core.apply_snapshot("hub-session", "hub", &RouteSnapshot::canonical(1, routes))
.unwrap();
assert_eq!(core.reachable_names().len(), 257);
assert_eq!(core.export_for("observer").len(), 257);
assert!(!core
.reachable_names()
.iter()
.any(|name| name == "feature.0"));
let envelope = Envelope::from_request(
http::Request::post("/node-0255/feature/4095")
.body(Bytes::new())
.unwrap(),
)
.unwrap();
let (peer, forwarded) = core.forward(envelope).unwrap();
assert_eq!(peer, "hub");
assert_eq!(forwarded.target, "node-0255");
assert_eq!(forwarded.subject, "feature.4095");
let routes_before_removal = core.export_for("observer");
core.install_local_capabilities(BTreeMap::new());
assert_eq!(core.export_for("observer"), routes_before_removal);
}
#[test]
fn index_and_full_catalog_entries_expose_canonical_callable_paths() {
let mut core = NodeCore::new("node-a");
core.install_local_capabilities(BTreeMap::from([(
"weather.current".into(),
serde_json::json!({
"one_line": "Current weather",
"operations": { "unary": { "output_schema": { "type": "object" } } }
}),
)]));
let index = core.catalog_subjects(false);
assert_eq!(index[0]["subject"], "weather.current");
assert_eq!(index[0]["target_path"], "/node-a/weather/current");
assert!(index[0].get("operations").is_none());
let full = core.catalog_subjects(true);
assert_eq!(full[0]["subject"], "weather.current");
assert_eq!(full[0]["target_path"], "/node-a/weather/current");
assert!(full[0]["operations"]["unary"].is_object());
}