server/lib/tests/the_flow_has_three_levels.rs
//! **The smart flow: three zoom levels of ONE model** (HOL-FLOW-SMART, Henrik).
//!
//! The console, Overview and Flow drew the same picture: the whole RON at once. Now the
//! server builds one model in which every part knows the machine it runs on, and zooms it:
//!
//! * **level 1** (Overview): the machines only (console, front, main, twin, each
//! downstream) and the upstream sources, joined by the lines between them;
//! * **level 2** (Flow): a machine named in `open` unfolds in place and shows its part of
//! the RON, and the lines that crossed its border end at the same ports as at level 1;
//! * **level 3**: everything unfolded, which is the whole model.
//!
//! Nothing that is not there is drawn: no console without one, no front, twin or
//! downstream that does not exist, and no "no downstream" placeholder.
#![cfg_attr(
not(all(feature = "supply-chain", feature = "sync")),
allow(dead_code, unused_imports)
)]
use std::collections::BTreeSet;
use holger_server_lib::flow::{
self, ConsoleView, DownstreamView, EstateBox, FlowGraph, IntakeTally, Measured, NodeKind, Open,
};
fn holger_of(ron: &str) -> holger_server_lib::Holger {
let mut f = tempfile::NamedTempFile::new().unwrap();
std::io::Write::write_all(&mut f, ron.as_bytes()).unwrap();
match holger_server_lib::read_ron_config(f.path()) {
Ok(h) => h,
Err(e) => panic!("the fixture declaration must parse: {e:#}"),
}
}
fn attr_values(svg: &str, key: &str) -> BTreeSet<String> {
let mut out = BTreeSet::new();
let mut rest = svg;
while let Some(i) = rest.find(key) {
rest = &rest[i + key.len()..];
let end = rest.find('"').unwrap();
out.insert(rest[..end].replace("->", "->"));
rest = &rest[end..];
}
out
}
fn set<const N: usize>(xs: [&str; N]) -> BTreeSet<String> {
xs.iter().map(|s| s.to_string()).collect()
}
const KEY: &str = "abababababababababababababababababababababababababababababababab";
/// The repositories of the intake chain and a crates proxy, all behind one door and one store.
const REPOS: &str = r#"
repositories: [
(ron_name: "incoming", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "web"))),
(ron_name: "quarantine", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "web"))),
(ron_name: "approved", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "web"))),
(ron_name: "production", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "web"))),
(ron_name: "crates", ron_repo_type: "rust", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "web"))),
],
exposed_endpoints: [
(ron_name: "web", ron_url: "127.0.0.1:50051", ron_http_url: Some("127.0.0.1:8443")),
],
storage_endpoints: [ (ron_name: "store", ron_storage_type: "znippy", ron_path: "/data/store") ],"#;
/// An upload source, the crates proxy, one scanner and the approvals, one declared downstream.
fn ron() -> String {
format!(
r#"({REPOS}
sources: [
(name: "team", kind: upload),
(name: "crates-io", kind: crates, serves: ["crates"]),
],
steps: (approvals: 2, scan: [vulnerability]),
rules: (allow_unchecked_sources: true),
sync: Some((name: "main", downstreams: [(name: "lab-1", key: "{KEY}")])),
)"#
)
}
/// The estate as the main of test3.holger.rs sees it: itself, a front, a twin, a console
/// that signed in a minute ago, the declared `lab-1` (via the front) and an enrolled `lab-2`
/// (via WireGuard). Two artifacts wait for approval and one is quarantined.
fn estate() -> Measured {
Measured {
now_unix: 1_800_000_000,
estate_name: "test3.holger.rs".into(),
estate: vec![
EstateBox {
name: "main".into(),
role: "primary".into(),
this_box: true,
state: "up".into(),
headline: "answering".into(),
..Default::default()
},
EstateBox {
name: "front".into(),
role: "front".into(),
state: "up".into(),
headline: "answering".into(),
..Default::default()
},
EstateBox {
name: "twin".into(),
role: "twin".into(),
state: "behind".into(),
headline: "behind 3 entries".into(),
twin_state: "behind".into(),
lag_entries: 3,
last_seen_unix_ms: 1_799_999_990_000,
seen_ago_ms: 10_000,
..Default::default()
},
],
console: Some(ConsoleView {
last_seen_unix_ms: 1_799_999_940_000,
seen_ago_ms: 60_000,
}),
downstreams: vec![
DownstreamView {
name: "lab-1".into(),
via: "front".into(),
state: "up".into(),
headline: "current".into(),
last_seen_unix_ms: 1_799_999_940_000,
seen_ago_ms: 60_000,
bytes_total: Some(6_000_140),
fetched_total: Some(3),
..Default::default()
},
DownstreamView {
name: "lab-2".into(),
via: "wireguard".into(),
state: "behind".into(),
headline: "behind".into(),
lag_entries: 4,
..Default::default()
},
],
intake: Some(IntakeTally {
states: [
("POLICY_CHECKED".to_string(), 2),
("QUARANTINED".to_string(), 1),
]
.into_iter()
.collect(),
..Default::default()
}),
..Default::default()
}
}
fn full() -> FlowGraph {
flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &estate())
}
/// The ports a graph's lines attach to on `machine`, as `(port, edge kind)`.
fn ports_on(g: &FlowGraph, machine: &str) -> BTreeSet<(String, String)> {
let prefix = format!("{machine}/");
g.edges
.iter()
.flat_map(|e| {
[e.from_port.clone(), e.to_port.clone()]
.into_iter()
.flatten()
.filter(|p| p.starts_with(&prefix))
.map(move |p| (p, e.kind.css().to_string()))
})
.collect()
}
/// ★ **Level 1 draws exactly the machines and the sources**, joined by the lines Topology and
/// the RON give, with the counters an operator needs first on the main.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn level_one_draws_the_machines_the_sources_and_the_lines_between_them() {
let l1 = full().zoom(&Open::none());
assert_eq!(l1.level, 1);
assert_eq!(
l1.node_ids(),
set([
"estate:console",
"estate:front",
"estate:main",
"estate:twin",
"downstream:lab-1",
"downstream:lab-2",
"source:team",
"crates-io",
]),
"the machines and the upstream sources, nothing inside a machine"
);
assert_eq!(
l1.edge_ids(),
set([
"source:team->estate:main",
"crates-io->estate:main",
"estate:console->estate:front",
"estate:front->estate:main",
"estate:main->estate:twin",
"estate:main->downstream:lab-1",
"estate:main->downstream:lab-2",
]),
"the lines between them, each once"
);
let svg = l1.render_svg();
assert_eq!(
attr_values(&svg, "data-node=\""),
l1.node_ids(),
"drawn as modelled"
);
assert_eq!(attr_values(&svg, "data-edge=\""), l1.edge_ids());
let main = l1.node("estate:main").unwrap();
let row = |label: &str| {
main.rows
.iter()
.find(|r| r.label == label)
.unwrap_or_else(|| panic!("main/{label}: {:?}", main.rows))
.cell
.render()
};
assert_eq!(row("waiting for approval"), "2");
assert_eq!(row("quarantined"), "1");
let sync = l1
.edges
.iter()
.find(|e| e.to == "downstream:lab-2")
.unwrap();
assert!(
sync.cell.render().contains("lag 4"),
"the line carries the lag: {}",
sync.cell.render()
);
}
/// ★ **Nothing that is not there is drawn**: a plain `cargo install holger` with one upload
/// source and no console, front, twin or downstream draws its main and that source, and
/// nothing else, at every level.
#[test]
fn a_plain_single_server_draws_main_and_its_sources_and_nothing_else() {
let h = holger_of(&format!(
r#"({REPOS}
sources: [ (name: "team", kind: upload) ],
)"#
));
let g = flow::measure(&FlowGraph::from_config(&h), &Measured::default());
let l1 = g.zoom(&Open::none());
assert_eq!(l1.node_ids(), set(["estate:main", "source:team"]));
assert_eq!(l1.edge_ids(), set(["source:team->estate:main"]));
for level in [Open::none(), Open::parse("main"), Open::All] {
let z = g.zoom(&level);
for n in &z.nodes {
assert!(
!matches!(
n.kind,
NodeKind::Console | NodeKind::Front | NodeKind::Twin | NodeKind::Downstream
),
"{} is drawn on an installation that has none",
n.id
);
assert_ne!(n.kind.css(), "none", "{}: no placeholder box", n.id);
}
}
}
/// ★ **A console is drawn only when this installation has one**: one that signed in or
/// called recently; a front alone draws no console.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_console_is_drawn_only_when_one_signed_in_or_called() {
let mut m = estate();
m.console = None;
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
assert!(
g.node("estate:console").is_none(),
"no console seen, no box"
);
assert!(g.node("estate:front").is_some(), "the front is still there");
}
/// ★ **`open=main` unfolds main in place** with its part of the RON, and the lines that
/// reached its border at level 1 end at the same ports, continuing inside to the part they
/// belong to. The other machines stay folded.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn opening_main_shows_its_ron_part_with_the_ports_of_level_one() {
let g = full();
let l1 = g.zoom(&Open::none());
let l2 = g.zoom(&Open::parse("main"));
assert_eq!(l2.level, 2);
assert_eq!(l2.open, vec!["estate:main".to_string()]);
for part in [
"incoming",
"quarantine",
"approved",
"production",
"crates",
"door:web",
"storage:store",
"step:scan · vulnerability",
"step:approval",
"gate:rules · the floor",
] {
assert!(l2.node(part).is_some(), "main's part {part} is unfolded");
}
assert!(
l2.node("estate:twin:mirror").is_none(),
"the twin stays folded"
);
assert!(l2.node("estate:twin").is_some());
let at_one = ports_on(&l1, "estate:main");
let at_two = ports_on(&l2, "estate:main");
assert_eq!(
at_one, at_two,
"the same ports on main's border at both levels"
);
let names: BTreeSet<String> = at_one
.iter()
.map(|(p, _)| p.trim_start_matches("estate:main/").to_string())
.collect();
assert_eq!(names, set(["intake", "doors", "export", "twin"]));
// Each line crossing the border continues inside to a part of main, except the
// twin's, which follows main as a whole.
for e in l2.edges.iter().filter(|e| {
e.to_port
.as_deref()
.is_some_and(|p| p.starts_with("estate:main/"))
|| e.from_port
.as_deref()
.is_some_and(|p| p.starts_with("estate:main/"))
}) {
let inner = if e
.to_port
.as_deref()
.is_some_and(|p| p.starts_with("estate:main/"))
{
&e.to
} else {
&e.from
};
if e.kind == flow::EdgeKind::Follows {
assert_eq!(inner, "estate:main");
} else {
assert_ne!(inner, "estate:main", "{} ends inside main", e.id());
assert_eq!(
l2.node(inner).and_then(|n| n.machine.as_deref()),
Some("estate:main"),
"{}: {inner} is main's",
e.id()
);
}
}
assert!(
l2.edge_ids().contains("source:team->incoming")
&& l2.edge_ids().contains("production->downstream:lab-1"),
"sources enter at intake, the downstreams take production through export: {:?}",
l2.edge_ids()
);
let (svg1, svg2) = (l1.render_svg(), l2.render_svg());
let drawn1 = attr_values(&svg1, "data-port=\"");
let drawn2 = attr_values(&svg2, "data-port=\"");
for p in at_one.iter().map(|(p, _)| p) {
assert!(
drawn1.contains(p) && drawn2.contains(p),
"port {p} is drawn at both levels"
);
}
assert_eq!(
attr_values(&svg2, "data-node=\""),
l2.node_ids(),
"drawn as modelled"
);
}
/// ★ **Level 3 is the whole model**: every part of every machine, every line.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn level_three_shows_everything() {
let g = full();
let l3 = g.zoom(&Open::All);
assert_eq!(l3.level, 3);
assert_eq!(l3.node_ids(), g.node_ids(), "every box");
assert_eq!(l3.edge_ids(), g.edge_ids(), "every line");
assert!(
l3.node("estate:twin:mirror").is_some(),
"the twin unfolds to its mirror"
);
for m in ["estate:main", "estate:twin"] {
assert!(l3.open.iter().any(|o| o == m), "{m} is open");
}
let svg = l3.render_svg();
assert_eq!(attr_values(&svg, "data-node=\""), g.node_ids());
}
/// ★ **Every part knows which box it runs on** — the model says it, not the renderer.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn every_part_says_which_box_it_runs_on() {
let g = full();
let on = |id: &str| g.node(id).unwrap_or_else(|| panic!("{id}")).machine.clone();
for part in [
"incoming",
"production",
"crates",
"door:web",
"storage:store",
"step:approval",
"gate:rules · the floor",
"observed pullers",
] {
assert_eq!(on(part).as_deref(), Some("estate:main"), "{part}");
}
for machine in [
"estate:main",
"estate:front",
"estate:twin",
"estate:console",
"downstream:lab-1",
] {
assert_eq!(
on(machine).as_deref(),
Some(machine),
"a machine is its own box"
);
}
assert_eq!(on("estate:twin:mirror").as_deref(), Some("estate:twin"));
for outside in ["source:team", "crates-io"] {
assert_eq!(
on(outside),
None,
"{outside} runs on no machine of the estate"
);
}
let v: serde_json::Value = serde_json::from_str(&g.to_json()).unwrap();
let incoming = v["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["id"] == "incoming")
.unwrap();
assert_eq!(incoming["box"], "estate:main", "the JSON names it `box`");
}
/// ★ **Unfolding is the server re-drawing**: each machine with parts carries the link that
/// toggles it, and the drawing offers the full RON. The page has no script.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn each_machine_carries_the_link_that_unfolds_it() {
let g = full();
let l1 = g.zoom(&Open::none()).render_svg();
assert!(l1.contains("<a href=\"flow?open=main\""), "main unfolds");
assert!(
l1.contains("<a href=\"flow?open=twin\""),
"the twin unfolds"
);
assert!(l1.contains("<a href=\"flow?open=all\""), "the full RON");
let l2 = g.zoom(&Open::parse("main")).render_svg();
assert!(
l2.contains("<a href=\"flow?open=\""),
"main folds again, to nothing open (not to the Flow page's default)"
);
assert!(
l2.contains("<a href=\"flow?open=main,twin\""),
"the twin unfolds beside main"
);
assert!(!l2.contains("<script"));
assert_eq!(Open::parse("main,lab-1"), Open::parse("lab-1, main"));
assert_eq!(Open::parse("all"), Open::All);
assert_eq!(Open::parse(""), Open::none());
}
// ── the diode: one way out, one way in ─────────────────────────────────────
/// A row's value as the model prints it.
fn row_of(g: &FlowGraph, node: &str, label: &str) -> String {
let n = g.node(node).unwrap_or_else(|| panic!("node {node}"));
n.rows
.iter()
.chain(n.facts.iter())
.find(|r| r.label == label)
.unwrap_or_else(|| panic!("{node}/{label}: {:?}", n.rows))
.cell
.render()
}
/// ★ **`export: (push_to: …)` draws the diode drop area** out of main's export port: a
/// box outside every machine, a one-way arrow, and the last bundle pushed when the push
/// state records one. Without `export:` there is no such box.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn an_export_push_draws_the_diode_drop_area_out_of_mains_export_port() {
let ron = ron().replace(
"\n)",
"\n export: Some((push_to: Some(\"/mnt/diode/out\"), every: \"5m\")),\n)",
);
let shape = FlowGraph::from_config(&holger_of(&ron));
let g = flow::measure(&shape, &estate());
let l1 = g.zoom(&Open::none());
let drop = l1
.node(flow::DIODE_ID)
.expect("the diode drop area is drawn at level 1");
assert_eq!(drop.machine, None, "it runs on no machine of the estate");
assert_eq!(drop.detail.as_deref(), Some("/mnt/diode/out"));
let out: Vec<_> = l1.edges.iter().filter(|e| e.to == flow::DIODE_ID).collect();
assert_eq!(out.len(), 1);
assert_eq!(out[0].from, "estate:main");
assert_eq!(out[0].kind, flow::EdgeKind::Export);
assert_eq!(out[0].from_port.as_deref(), Some("estate:main/export"));
assert!(
l1.edges.iter().all(|e| e.from != flow::DIODE_ID),
"nothing comes back through a diode"
);
let l2 = g.zoom(&Open::parse("main"));
assert!(
l2.edge_ids()
.contains(&format!("production->{}", flow::DIODE_ID)),
"the bundles leave the chain's production: {:?}",
l2.edge_ids()
);
assert_eq!(row_of(&l1, flow::DIODE_ID, "last bundle"), "not recorded");
let mut m = estate();
m.export = Some(flow::ExportView {
sequence: 7,
last_unix: Some(m.now_unix - 180),
});
let pushed = flow::measure(&shape, &m);
assert_eq!(row_of(&pushed, flow::DIODE_ID, "last bundle"), "7 · 3m ago");
let svg = pushed.zoom(&Open::none()).render_svg();
assert!(svg.contains(&format!("data-node=\"{}\"", flow::DIODE_ID)));
assert!(svg.contains("data-kind=\"export\""), "a one-way arrow");
assert!(
full().node(flow::DIODE_ID).is_none(),
"no export:, no diode drop area"
);
}
/// ★ **A one-way downstream draws its inbox as its ONE input port** (`boundary: OneWay`):
/// the inbox enters main at `inbox`, the bundles land in production, and no line leaves
/// main — no fetch, no receipt, nothing back.
#[cfg(feature = "supply-chain")]
#[test]
fn a_one_way_downstream_draws_its_inbox_as_its_one_input_port() {
let h = holger_of(&format!(
r#"({REPOS}
steps: (approvals: 1, scan: [vulnerability]),
rules: (allow_unchecked_sources: true),
boundary: OneWay,
)"#
));
let shape = FlowGraph::from_config(&h);
let g = flow::measure(&shape, &Measured::default());
let l1 = g.zoom(&Open::none());
assert_eq!(l1.node_ids(), set(["estate:main", flow::INBOX_ID]));
assert_eq!(
l1.edge_ids(),
set([&*format!("{}->estate:main", flow::INBOX_ID)])
);
let e = &l1.edges[0];
assert_eq!(e.kind, flow::EdgeKind::Carries);
assert_eq!(e.to_port.as_deref(), Some("estate:main/inbox"));
assert_eq!(
ports_on(&l1, "estate:main"),
[("estate:main/inbox".to_string(), "carries".to_string())]
.into_iter()
.collect(),
"ONE input port, the inbox"
);
for level in [Open::parse("main"), Open::All] {
let z = g.zoom(&level);
assert!(
z.edges
.iter()
.all(
|e| z.node(&e.from).and_then(|n| n.machine.as_deref()) != Some("estate:main")
|| z.node(&e.to).and_then(|n| n.machine.as_deref()) == Some("estate:main")
),
"no line leaves a one-way downstream: {:?}",
z.edge_ids()
);
}
assert!(
g.zoom(&Open::parse("main"))
.edge_ids()
.contains(&format!("{}->production", flow::INBOX_ID)),
"the bundles land in production"
);
assert!(
g.node(flow::NO_SOURCES_NODE).is_none(),
"the inbox is where the bytes come from"
);
assert_eq!(row_of(&g, flow::INBOX_ID, "imported"), "not recorded");
let m = Measured {
now_unix: 1_800_000_000,
inbox: Some(flow::InboxView {
imported: 3,
refused: 1,
last_name: "holger-packages-12.holger.znippy".into(),
last_sequence: None,
last_unix: Some(1_800_000_000 - 7200),
}),
..Default::default()
};
let seen = flow::measure(&shape, &m);
assert_eq!(row_of(&seen, flow::INBOX_ID, "imported"), "3");
assert_eq!(row_of(&seen, flow::INBOX_ID, "refused"), "1");
assert_eq!(row_of(&seen, flow::INBOX_ID, "last import"), "2h ago");
}
// ── what a downstream's line is known from ─────────────────────────────────
/// ★ **A downstream's line says which sources agree** — declared in the RON, paired
/// (enrolled from a signed invite, or mTLS-paired), seen (a sync report within
/// `silent_after_ms`) — and looks it: solid when declared or paired AND seen recently,
/// dashed with "not seen since …" when declared or paired but silent, grey with a "?" when
/// seen but neither declared nor paired (it should not happen; it is shown, never hidden).
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn a_downstreams_line_shows_where_it_is_known_from() {
let mut m = estate();
m.silent_after_ms = 300_000;
m.downstreams = vec![
// declared in the RON, reported a minute ago
DownstreamView {
name: "lab-1".into(),
via: "front".into(),
state: "up".into(),
declared: true,
last_seen_unix_ms: 1_799_999_940_000,
seen_ago_ms: 60_000,
..Default::default()
},
// enrolled from an invite, silent for an hour
DownstreamView {
name: "lab-2".into(),
via: "wireguard".into(),
state: "silent".into(),
paired: true,
last_seen_unix_ms: 1_799_996_400_000,
seen_ago_ms: 3_600_000,
..Default::default()
},
// enrolled and never heard from
DownstreamView {
name: "lab-3".into(),
via: "front".into(),
state: "unknown".into(),
paired: true,
..Default::default()
},
// heard from, but nothing declares or pairs it
DownstreamView {
name: "stray".into(),
state: "up".into(),
last_seen_unix_ms: 1_799_999_990_000,
seen_ago_ms: 10_000,
..Default::default()
},
];
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
let l1 = g.zoom(&Open::none());
let line = |name: &str| {
l1.edges
.iter()
.find(|e| e.to == format!("downstream:{name}"))
.unwrap_or_else(|| panic!("the line to {name}"))
.clone()
};
let words = |name: &str| line(name).cell.render();
assert_eq!(line("lab-1").look, Some("live"), "declared and seen: solid");
assert_eq!(line("lab-1").known_from, vec!["declared", "seen"]);
assert_eq!(line("lab-2").look, Some("silent"), "paired, silent: dashed");
assert_eq!(line("lab-2").known_from, vec!["paired"]);
assert!(
words("lab-2").contains("not seen since 1h ago"),
"{}",
words("lab-2")
);
assert_eq!(line("lab-3").look, Some("silent"));
assert!(words("lab-3").contains("never seen"), "{}", words("lab-3"));
assert_eq!(line("stray").look, Some("unknown"), "seen only: grey");
assert_eq!(line("stray").known_from, vec!["seen"]);
assert!(words("stray").starts_with('?'), "{}", words("stray"));
let svg = l1.render_svg();
for (name, look) in [("lab-1", "live"), ("lab-2", "silent"), ("stray", "unknown")] {
assert!(
svg.contains(&format!(
"data-edge=\"estate:main->downstream:{name}\" data-kind=\"syncs\" data-look=\"{look}\""
)),
"{name} is drawn {look}"
);
}
assert!(
l1.node("downstream:stray").is_some(),
"a downstream nothing declares or pairs is shown, never hidden"
);
}
// ── a downstream's own RON, unfolded ──────────────────────────────────────
/// What `lab-1` says of its own RON in its sync report.
fn lab_summary() -> flow::RonSummary {
flow::RonSummary {
boundary: "Fetch".into(),
repositories: vec!["pypi".into(), "oci".into()],
inbox: String::new(),
first_import: "admin".into(),
domain: "lab".into(),
downstreams: 2,
console_url: String::new(),
}
}
/// ★ **A downstream unfolds to its own RON summary** — the one its sync report carried:
/// its repositories inside its border, the fetch line ending at the same port at both
/// levels and continuing to them, and on its box its boundary, its domain and how many it
/// is upstream of. One without a summary has nothing to unfold.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn opening_a_downstream_shows_its_own_ron_summary() {
let mut m = estate();
m.downstreams[0].summary = Some(lab_summary());
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
assert!(g.openable.iter().any(|o| o == "downstream:lab-1"));
assert!(
!g.openable.iter().any(|o| o == "downstream:lab-2"),
"no summary, nothing to unfold"
);
let l1 = g.zoom(&Open::none());
assert_eq!(row_of(&l1, "downstream:lab-1", "boundary"), "Fetch");
assert_eq!(row_of(&l1, "downstream:lab-1", "domain"), "lab");
assert_eq!(row_of(&l1, "downstream:lab-1", "upstream of"), "2");
assert!(l1.render_svg().contains("<a href=\"flow?open=lab-1\""));
let l2 = g.zoom(&Open::parse("lab-1"));
for part in ["downstream:lab-1:pypi", "downstream:lab-1:oci"] {
let n = l2.node(part).unwrap_or_else(|| panic!("{part} unfolded"));
assert_eq!(n.machine.as_deref(), Some("downstream:lab-1"));
assert!(
!n.kind.is_upstream(),
"a downstream's repository is no upstream"
);
}
assert!(l2.node("incoming").is_none(), "main stays folded");
assert_eq!(
ports_on(&l1, "downstream:lab-1"),
ports_on(&l2, "downstream:lab-1"),
"the same port at both levels"
);
assert!(
l2.edge_ids().contains("estate:main->downstream:lab-1:pypi"),
"the fetch line continues to what it fills: {:?}",
l2.edge_ids()
);
}
/// ★ **What a downstream says of its own RON** is names, words and folders: its
/// repositories, its boundary, its inbox, who allows a first import, its domain, how many
/// downstreams it declares — never a key.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn a_downstream_summarises_its_own_ron_without_a_secret() {
let h = holger_of(&format!(
r#"({REPOS}
boundary: Fetch,
domain: "lab",
sync: Some((name: "lab-1", downstreams: [(name: "shelf", key: "{KEY}")])),
)"#
));
let s = flow::ron_summary(&h);
assert_eq!(s.boundary, "Fetch");
assert_eq!(
s.repositories,
vec!["incoming", "quarantine", "approved", "production", "crates"]
);
assert_eq!(s.inbox, "", "a fetching downstream has no inbox");
assert_eq!(s.first_import, "admin");
assert_eq!(s.domain, "lab");
assert_eq!(s.downstreams, 1);
assert!(!format!("{s:?}").contains(KEY), "no key in the summary");
let one_way = holger_of(&format!("({REPOS}\n boundary: OneWay,\n)"));
let s = flow::ron_summary(&one_way);
assert_eq!(s.boundary, "OneWay");
assert!(s.inbox.ends_with("bundle.inbox"), "{}", s.inbox);
}
/// ★ **Where the drawing is clickable, for a console that shows it as an image**: each
/// machine with parts and the full-RON link, as the share of the drawing it covers and the
/// link it follows. An `<img>` runs nothing; the console lays plain links over it there.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_drawing_says_where_it_is_clickable() {
let g = full();
let (svg, links) = g.zoom(&Open::none()).render_with_links();
assert!(svg.starts_with("<svg"));
let by_href = |h: &str| {
links
.iter()
.find(|l| l.href == h)
.unwrap_or_else(|| panic!("a link to {h}: {links:?}"))
.clone()
};
let main = by_href("flow?open=main");
assert_eq!(main.label, "unfold main");
assert_eq!(by_href("flow?open=twin").label, "unfold twin");
assert_eq!(by_href("flow?open=all").label, "show the full RON");
for l in &links {
for v in [l.left, l.top, l.width, l.height] {
assert!((0.0..=1.0).contains(&v), "{l:?} is a share of the drawing");
}
assert!(l.width > 0.0 && l.height > 0.0);
}
assert!(
links
.iter()
.all(|l| !l.href.contains("console") && !l.href.contains("front")),
"a machine without parts has nothing to unfold"
);
let (_, opened) = g.zoom(&Open::parse("main")).render_with_links();
assert!(
opened
.iter()
.any(|l| l.href == "flow?open=" && l.label == "fold main"),
"{opened:?}"
);
}
// ── HOL-SYNC-SKEW: the review's small points ───────────────────────────────
/// ★ **A downstream nothing declares or pairs, seen only long ago, is grey with a "?"**
/// (the spec: "grey with a '?' = seen, but neither declared nor paired"), never the dashed
/// line of a silent declared one; and the grey line is no dashed line.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn a_stray_downstream_seen_long_ago_is_grey_with_a_question_mark() {
let mut m = estate();
m.silent_after_ms = 300_000;
m.downstreams = vec![DownstreamView {
name: "old-stray".into(),
state: "silent".into(),
last_seen_unix_ms: 1_799_992_800_000,
seen_ago_ms: 7_200_000,
..Default::default()
}];
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
let l1 = g.zoom(&Open::none());
let line = l1
.edges
.iter()
.find(|e| e.to == "downstream:old-stray")
.expect("the line to old-stray")
.clone();
assert_eq!(
line.look,
Some("unknown"),
"neither declared nor paired: grey"
);
assert_eq!(
line.known_from,
Vec::<&str>::new(),
"not seen lately either"
);
assert!(
line.cell.render().starts_with('?'),
"{}",
line.cell.render()
);
let svg = l1.render_svg();
let rule = svg
.split('\n')
.find(|l| l.trim_start().starts_with(".edge.unknown{"))
.expect("the grey line's style");
assert!(!rule.contains("dasharray"), "grey, not dashed: {rule}");
}
/// ★ **A downstream that sent no summary says so** (HOL-SYNC-SKEW): an older downstream's
/// report has no RON summary, and its box reads "no summary" instead of nothing.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn a_downstream_without_a_summary_says_no_summary() {
let mut m = estate();
m.downstreams[0].summary = None;
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
let l1 = g.zoom(&Open::none());
assert_eq!(row_of(&l1, "downstream:lab-1", "summary"), "no summary");
m.downstreams[0].summary = Some(lab_summary());
let said = flow::measure(&FlowGraph::from_config(&holger_of(&ron())), &m);
assert!(
said.node("downstream:lab-1")
.unwrap()
.facts
.iter()
.all(|r| r.label != "summary"),
"one that said its RON shows it instead"
);
}
fn audit_row(
ident: &str,
status: u16,
artifact: &str,
detail: &str,
ago_secs: i64,
) -> holger_server_lib::audit::AuditEvent {
let mut ev = holger_server_lib::audit::AuditEvent::new(
ident,
holger_server_lib::audit::AuditAction::Other,
"production",
artifact,
"local:inbox",
status,
0,
)
.with_detail(detail);
ev.ts_nanos = holger_server_lib::audit::now_nanos() - ago_secs * 1_000_000_000;
ev
}
fn sources_of(
h: &holger_server_lib::Holger,
audit: std::sync::Arc<holger_server_lib::audit::MemoryAuditLog>,
) -> flow::Sources {
flow::Sources {
shape: std::sync::Arc::new(FlowGraph::from_config(h)),
repos: Vec::new(),
audit,
visitors: std::sync::Arc::new(holger_server_lib::visitors::Visitors::empty()),
console: Default::default(),
board: None,
#[cfg(feature = "intake")]
intake: None,
quota: None,
}
}
/// ★ **The inbox's counters are the import audit's** (HOL-SYNC-SKEW): imported bundles are
/// the audit's `bundle-import: bundle_id=… seq=…` rows and refused ones its `bundle import
/// refused` rows, never the files left in the inbox's folders; the last bundle names its
/// sequence and its time from the audit.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_inbox_counts_from_the_import_audit_not_from_its_folders() {
use holger_server_lib::audit::AuditLog as _;
let d = tempfile::tempdir().unwrap();
let h = holger_of(&format!(
r#"({REPOS}
steps: (approvals: 1, scan: [vulnerability]),
rules: (allow_unchecked_sources: true),
boundary: OneWay,
audit: (dir: "{}/audit", enabled: true),
)"#,
d.path().display()
));
// Files in the inbox's folders that no import wrote: they are not counted.
let archive = d
.path()
.join(holger_server_lib::import_door::BUNDLE_INBOX)
.join(holger_server_lib::import_door::ARCHIVE_DIR);
std::fs::create_dir_all(&archive).unwrap();
for i in 0..5 {
std::fs::write(archive.join(format!("stale-{i}.holger.znippy")), b"x").unwrap();
}
let audit = std::sync::Arc::new(holger_server_lib::audit::MemoryAuditLog::new());
for ev in [
audit_row("inbox", 200, "acme/pkgs/2", "bundle-import: bundle_id=acme-pkgs-2 seq=2 digest=sha256:aa outcome=new into the intake chain", 9_000),
audit_row("inbox", 200, "acme-pkgs-2/a.whl", "bundle-import: imported on the airgap side from bundle acme-pkgs-2 seq 2 (new)", 9_000),
audit_row("inbox", 200, "acme/pkgs/3", "bundle-import: bundle_id=acme-pkgs-3 seq=3 digest=sha256:bb outcome=new into the intake chain", 7_200),
audit_row("inbox", 409, "bad.holger.znippy", "bundle import refused: bad.holger.znippy: holger-airgap-3: not signed", 60),
] {
audit.record(ev).unwrap();
}
let g = flow::live(&sources_of(&h, audit));
assert_eq!(row_of(&g, flow::INBOX_ID, "imported"), "2");
assert_eq!(row_of(&g, flow::INBOX_ID, "refused"), "1");
assert_eq!(row_of(&g, flow::INBOX_ID, "last import"), "2h ago");
assert_eq!(
row_of(&g, flow::INBOX_ID, "last bundle"),
"acme-pkgs-3 · seq 3"
);
}
/// ★ **The diode drop area's last bundle is the export audit's** (HOL-SYNC-SKEW): the
/// export push's `airgap export <id> seq <n>` row names the sequence and the time, not the
/// push state file's mtime; with no such row it reads "not recorded".
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_diode_drop_areas_last_bundle_is_the_export_audits() {
use holger_server_lib::audit::AuditLog as _;
let d = tempfile::tempdir().unwrap();
let ron = ron().replace(
"\n)",
&format!(
"\n export: Some((push_to: Some(\"/mnt/diode/out\"), every: \"5m\")),\n audit: (dir: \"{}/audit\", enabled: true),\n)",
d.path().display()
),
);
let h = holger_of(&ron);
let audit = std::sync::Arc::new(holger_server_lib::audit::MemoryAuditLog::new());
let g = flow::live(&sources_of(&h, audit.clone()));
assert_eq!(row_of(&g, flow::DIODE_ID, "last bundle"), "not recorded");
audit
.record(audit_row(
"export-push",
200,
"acme-pkgs-7",
"airgap export acme-pkgs-7 seq 7: 1 artifact(s) [a], verdict pass (computed), candidate sha256:cc",
180,
))
.unwrap();
let g = flow::live(&sources_of(&h, audit));
assert_eq!(row_of(&g, flow::DIODE_ID, "last bundle"), "7 · 3m ago");
}