server/lib/tests/the_flow_is_the_ron.rs
//! **The artifact flow is the RON** (lane HOL-SVG-RON, owner 2026-10-30).
//!
//! The estate board that replaced the flow drawing (2026-10-17) drew front | main | twin
//! and nothing of the RON. The owner's rule: the RON is the source of truth, and the
//! drawing is exactly the RON — every source, the intake chain's steps and gates, the
//! downstreams — with the estate and the counters written on it. These tests hold that,
//! on the drawing itself (`data-node` / `data-edge`) and on the JSON model:
//!
//! * a RON with 2 sources, 3 steps and 1 downstream draws exactly those nodes and edges;
//! * changing the RON changes the drawing;
//! * a RON with no source says "no sources configured", and shows no sample;
//! * a downstream is never drawn or worded as an upstream;
//! * every counter moves when its thing happens, and a counter with no source reads
//! "not recorded", never 0.
// The RONs with `steps:` and `sync:` need those features; the default build runs the rest.
#![cfg_attr(
not(all(feature = "supply-chain", feature = "sync")),
allow(dead_code, unused_imports)
)]
use std::collections::{BTreeMap, BTreeSet};
use holger_server_lib::flow::{
self, AuditTally, Cell, DownstreamView, EstateBox, FlowGraph, IntakeTally, Measured,
MirrorView, NodeKind,
};
/// A declaration's text, read the way `holger-server start --config` reads it.
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:#}"),
}
}
/// The `data-node` ids of a drawing.
fn drawn_nodes(svg: &str) -> BTreeSet<String> {
attr_values(svg, "data-node=\"")
}
/// The `data-edge` ids of a drawing (`from->to`, unescaped).
fn drawn_edges(svg: &str) -> BTreeSet<String> {
attr_values(svg, "data-edge=\"")
.into_iter()
.map(|e| e.replace("->", "->"))
.collect()
}
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].to_string());
rest = &rest[end..];
}
out
}
/// The `<g data-node="id" …>…</g>` group of one node.
fn group<'a>(svg: &'a str, id: &str) -> &'a str {
let start = svg
.find(&format!("<g data-node=\"{id}\""))
.unwrap_or_else(|| panic!("node {id} is drawn"));
let end = svg[start..].find("</g>").unwrap();
&svg[start..start + end]
}
/// A row's value as drawn (`data-row="node/label"`).
fn row(svg: &str, node: &str, label: &str) -> String {
let key = format!("data-row=\"{node}/{label}\"");
let i = svg
.find(&key)
.unwrap_or_else(|| panic!("row {node}/{label} is drawn"));
let rest = &svg[i..];
let gt = rest.find('>').unwrap();
let lt = rest[gt..].find('<').unwrap();
rest[gt + 1..gt + lt].to_string()
}
fn graph_row(g: &FlowGraph, node: &str, label: &str) -> Cell {
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!("row {node}/{label}: {:?}", n.rows))
.cell
.clone()
}
fn set(xs: &[&str]) -> BTreeSet<String> {
xs.iter().map(|s| s.to_string()).collect()
}
const KEY: &str = "abababababababababababababababababababababababababababababababab";
/// The intake repositories every chain names.
const INTAKE: &str = r#"
(ron_name: "incoming", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
(ron_name: "quarantine", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
(ron_name: "approved", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
(ron_name: "production", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
(ron_name: "crates", ron_repo_type: "rust", ron_upstreams: [], ron_in: None, ron_out: None),"#;
/// 2 sources, 3 steps (two scanners and the approvals), the floor, 1 downstream.
fn ron(extra_sources: &str) -> String {
format!(
r#"(
repositories: [{INTAKE}
],
exposed_endpoints: [],
storage_endpoints: [],
sources: [
(name: "team", kind: upload),
(name: "crates-io", kind: crates, serves: ["crates"]),{extra_sources}
],
steps: (approvals: 2, scan: [vulnerability, license]),
rules: (allow_unchecked_sources: true),
sync: Some((name: "main", downstreams: [(name: "lab-1", key: "{KEY}")])),
)"#
)
}
/// ★ **A RON with 2 sources, 3 steps and 1 downstream draws exactly those** — and the
/// repositories and the ladder the RON declares, and nothing else.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn two_sources_three_steps_and_one_downstream_draw_exactly_those_nodes_and_edges() {
let shape = FlowGraph::from_config(&holger_of(&ron("")));
let svg = shape.render_svg();
assert_eq!(
drawn_nodes(&svg),
set(&[
// this server, the machine every part of its RON runs on
"estate:main",
// the 2 sources, by their names: the upload door, and the remote `crates-io` lowers to
"source:team",
"crates-io",
// the repositories the RON declares
"incoming",
"quarantine",
"approved",
"production",
"crates",
// the 3 steps
"step:scan · vulnerability",
"step:scan · license",
"step:approval",
// the floor the RON writes (`rules:`), a gate on the scan's hop
"gate:rules · the floor",
// the 1 downstream
"downstream:lab-1",
]),
"the drawing is the RON, node for node"
);
assert_eq!(
drawn_edges(&svg),
set(&[
"source:team->incoming",
"crates-io->crates",
"incoming->quarantine",
"quarantine->approved",
"approved->production",
"step:scan · vulnerability->quarantine",
"step:scan · license->quarantine",
"step:approval->approved",
"gate:rules · the floor->quarantine",
]),
"the drawing is the RON, edge for edge"
);
// The names as written in the RON.
for name in [
"team",
"crates-io",
"lab-1",
"incoming",
"production",
"crates",
] {
assert!(
svg.contains(&format!(">{name}</text>")),
"{name} is printed as the RON writes it"
);
}
assert_eq!(
shape.node("step:approval").unwrap().facts[0].cell,
Cell::Count(2),
"approvals: 2"
);
// The JSON model carries the same graph.
let v: serde_json::Value = serde_json::from_str(&shape.to_json()).unwrap();
assert_eq!(v["schema"], "holger.flow/v2");
let ids: BTreeSet<String> = v["nodes"]
.as_array()
.unwrap()
.iter()
.map(|n| n["id"].as_str().unwrap().to_string())
.collect();
assert_eq!(
ids,
drawn_nodes(&svg),
"the JSON model and the drawing are one graph"
);
}
/// ★ **Changing the RON changes the drawing**: a third source appears, by its name, with
/// its edge; take it out and it is gone.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn changing_the_ron_changes_the_drawing() {
let before = FlowGraph::from_config(&holger_of(&ron(""))).render_svg();
let after = FlowGraph::from_config(&holger_of(&ron(
"\n (name: \"vendor\", kind: upload, importers: [\"ci\"]),",
)))
.render_svg();
assert_ne!(before, after, "another RON, another drawing");
assert!(!drawn_nodes(&before).contains("source:vendor"));
assert!(
drawn_nodes(&after).contains("source:vendor")
&& drawn_edges(&after).contains("source:vendor->incoming"),
"the new source is drawn with its edge"
);
assert!(
group(&after, "source:vendor").contains(">ci</text>"),
"with its importers, as declared"
);
assert_eq!(
drawn_nodes(&after).len(),
drawn_nodes(&before).len() + 1,
"one source more, one box more"
);
}
/// ★ **A RON with no source says so plainly** — one box, "no sources configured", grey;
/// no remote, no mirror, no sample source.
#[test]
fn a_ron_with_no_sources_says_no_sources_configured() {
let h = holger_of(
r#"(
repositories: [
(ron_name: "bundles", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
],
exposed_endpoints: [],
storage_endpoints: [],
)"#,
);
let shape = FlowGraph::from_config(&h);
let svg = shape.render_svg();
assert!(svg.contains(">no sources configured</text>"), "it says so");
assert_eq!(
drawn_nodes(&svg),
set(&["bundles", flow::NO_SOURCES_NODE, flow::MAIN_ID]),
"and draws nothing else but the server it runs on: no sample"
);
assert!(
shape.nodes.iter().all(|n| !n.kind.is_upstream()),
"no source of any kind"
);
assert_eq!(
shape.node(flow::NO_SOURCES_NODE).unwrap().state,
Some("none"),
"grey: a statement, not a source"
);
}
/// The estate as a main with a twin, a front, a declared downstream and an enrolled one sees it.
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: "unknown".into(),
headline: "not asked".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(flow::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,
lag_entries: 0,
bytes_total: Some(6_000_140),
fetched_total: Some(3),
repos: vec![],
declared: true,
paired: false,
summary: None,
},
DownstreamView {
name: "lab-2".into(),
via: "wireguard".into(),
state: "behind".into(),
headline: "behind".into(),
lag_entries: 4,
..Default::default()
},
],
..Default::default()
}
}
/// ★ **A downstream is never an upstream** (the owner, holger.rs 2026-10-30: "the
/// downstreams are labelled upstream"). Each downstream is its own kind, worded
/// "downstream holger", drawn in the estate band; the only arrow touching it runs FROM the
/// main TO it and says how it reaches this holger.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn a_downstream_is_never_drawn_or_labelled_as_an_upstream() {
let shape = FlowGraph::from_config(&holger_of(&ron("")));
let g = flow::measure(&shape, &estate());
let svg = g.render_svg();
for id in ["downstream:lab-1", "downstream:lab-2"] {
let n = g.node(id).unwrap_or_else(|| panic!("{id} is drawn"));
assert_eq!(n.kind, NodeKind::Downstream);
assert!(!n.kind.is_upstream(), "{id} is no upstream");
let drawn = group(&svg, id);
assert!(
drawn.contains(">DOWNSTREAM HOLGER</text>"),
"{id} says what it is: {drawn}"
);
assert!(
!drawn.to_lowercase().contains("upstream"),
"{id} never says upstream: {drawn}"
);
let touching: Vec<_> = g
.edges
.iter()
.filter(|e| e.from == id || e.to == id)
.collect();
assert_eq!(touching.len(), 1, "{id}: one arrow");
assert_eq!(touching[0].kind, flow::EdgeKind::Syncs);
assert_eq!(touching[0].to, id, "the arrow runs to the downstream");
assert_eq!(
g.node(&touching[0].from).unwrap().machine.as_deref(),
Some(flow::MAIN_ID),
"from the main (what it takes there: the chain's production)"
);
}
assert_eq!(row(&svg, "downstream:lab-1", "via"), "front");
assert_eq!(row(&svg, "downstream:lab-2", "via"), "wireguard");
assert!(
svg.contains(">via wireguard · lag 4"),
"the arrow says the via and the lag"
);
// Nothing that brings bytes in is a downstream's name.
for n in g.nodes.iter().filter(|n| n.kind.is_upstream()) {
assert!(
!n.label.starts_with("lab-"),
"{} is drawn as an upstream",
n.label
);
}
let v: serde_json::Value = serde_json::from_str(&g.to_json()).unwrap();
let kinds: BTreeMap<String, String> = v["nodes"]
.as_array()
.unwrap()
.iter()
.map(|n| {
(
n["id"].as_str().unwrap().to_string(),
n["kind"].as_str().unwrap().to_string(),
)
})
.collect();
assert_eq!(kinds["downstream:lab-1"], "downstream");
}
/// ★ **The estate is drawn with the RON**: the console and the front, this main, the twin
/// (following, entries behind, last sync), each downstream (via, status, last sync, lag,
/// bytes sent); what this main does not record (the front's requests) says so.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_estate_is_drawn_with_its_counters() {
let g = flow::measure(&FlowGraph::from_config(&holger_of(&ron(""))), &estate());
let svg = g.render_svg();
for id in [
"estate:main",
"estate:front",
"estate:console",
"estate:twin",
] {
assert!(drawn_nodes(&svg).contains(id), "{id} is drawn");
}
let machines = g.zoom(&flow::Open::none()).render_svg();
assert!(
drawn_edges(&machines).contains("estate:front->estate:main")
&& drawn_edges(&machines).contains("estate:main->estate:twin"),
"the front serves the main, the twin follows it"
);
assert!(
drawn_edges(&svg).contains("estate:main->estate:twin:mirror"),
"unfolded, the twin's mirror follows the main"
);
assert_eq!(row(&svg, "estate:twin", "following"), "yes");
assert_eq!(row(&svg, "estate:twin", "entries behind"), "3");
assert_eq!(row(&svg, "estate:twin", "last sync"), "10s ago");
assert_eq!(
row(&svg, "estate:front", "requests"),
"not recorded",
"the front's counters live on the front"
);
assert_eq!(row(&svg, "downstream:lab-1", "bytes sent"), "6.0 MB");
assert_eq!(row(&svg, "downstream:lab-1", "last sync"), "1m ago");
assert_eq!(row(&svg, "downstream:lab-2", "entries behind"), "4");
assert_eq!(
row(&svg, "downstream:lab-2", "bytes sent"),
"not recorded",
"no peer record counts it: never 0"
);
}
fn download(
repo: &str,
ip: &str,
bytes: u64,
ts_nanos: i64,
) -> holger_server_lib::audit::AuditEvent {
let mut e = holger_server_lib::audit::AuditEvent::new(
"anonymous",
holger_server_lib::audit::AuditAction::Download,
repo,
"a/b@1",
ip,
200,
bytes,
);
e.ts_nanos = ts_nanos;
e
}
/// ★ **The audit counters move**: a download moves its repository's fetches and bytes (24 h
/// and total), its puller's bytes; an old one moves only the total; a refused one moves
/// nothing; loopback is apart; a promotion moves its arrow.
#[test]
fn the_audit_counters_move_when_a_download_or_a_promotion_happens() {
let now = 1_800_000_000 * 1_000_000_000i64;
let empty = AuditTally::fold(&[], now);
assert!(empty.repos.is_empty());
let mut events = vec![download("prod", "10.0.0.5:4431", 100, now - 1_000_000_000)];
let one = AuditTally::fold(&events, now);
assert_eq!(
(
one.repos["prod"].fetches,
one.repos["prod"].bytes,
one.repos["prod"].fetches_24h
),
(1, 100, 1)
);
events.push(download(
"prod",
"10.0.0.5:5000",
50,
now - 2 * 86_400 * 1_000_000_000,
));
events.push(download("prod", "127.0.0.1:6000", 7, now));
let mut refused = download("prod", "10.0.0.6:1", 999, now);
refused.status = 403;
events.push(refused);
let mut promote = holger_server_lib::audit::AuditEvent::new(
"admin",
holger_server_lib::audit::AuditAction::Promote,
"b",
"a/b@1",
"",
200,
0,
);
promote.ts_nanos = now;
events.push(promote);
let t = AuditTally::fold(&events, now);
assert_eq!(
t.repos["prod"].fetches, 3,
"the old and the loopback one count; the refused one does not"
);
assert_eq!(
(t.repos["prod"].fetches_24h, t.repos["prod"].bytes_24h),
(2, 107),
"the 24 h window drops the old one"
);
assert_eq!(
t.clients["10.0.0.5"].bytes, 150,
"one puller, its port dropped"
);
assert!(
!t.clients.contains_key("10.0.0.6"),
"a refusal sent nothing"
);
assert_eq!(t.loopback.fetches, 1, "loopback is apart");
assert_eq!(t.promotions["b"].0, 1);
}
/// ★ **The counters come from the audit aggregates** (lane HOL-AUDIT): the tally the
/// flow reads from the Arrow index's aggregates is the fold of the same records, peer,
/// loopback, refused, old and promotion rows included, from a real chained log.
#[test]
fn the_tally_from_the_aggregates_is_the_fold() {
use holger_server_lib::audit::{AuditAction, AuditEvent, AuditLog, FileAuditLog};
let now = 1_800_000_000 * 1_000_000_000i64;
let dir = tempfile::tempdir().unwrap();
let log = FileAuditLog::new(dir.path())
.unwrap()
.with_checkpoint_every(3);
let mut events = vec![
download("prod", "10.0.0.5:4431", 100, now - 1_000_000_000),
download(
"prod",
"10.0.0.5:5000",
50,
now - 2 * 86_400 * 1_000_000_000,
),
download("prod", "127.0.0.1:6000", 7, now),
download("dev", "[2001:db8::7]:443", 9, now),
];
let mut refused = download("prod", "10.0.0.6:1", 999, now);
refused.status = 403;
events.push(refused);
let mut peer = AuditEvent::new(
"holger-peer:lab-1",
AuditAction::Download,
"prod",
"a/b@1",
"10.0.0.8:1",
200,
4096,
);
peer.ts_nanos = now;
events.push(peer);
let mut promote = AuditEvent::new("admin", AuditAction::Promote, "b", "a/b@1", "", 200, 0);
promote.ts_nanos = now - 5;
events.push(promote);
for e in &events {
log.record(e.clone()).unwrap();
}
let table = log.query_source().unwrap().table().unwrap();
let from_index = AuditTally::from_table(&table, now).unwrap();
assert_eq!(from_index, AuditTally::fold(&events, now));
assert_eq!(from_index.peers["lab-1"].bytes, 4096);
assert_eq!(from_index.clients["2001:db8::7"].fetches, 1);
}
/// A flow with a promotion ladder `a → b`, a proxy, a mirror, a door and a store.
const LADDER: &str = r#"(
repositories: [
(ron_name: "a", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "door"))),
(ron_name: "b", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "store", ron_exposed_endpoint: "door"))),
(ron_name: "pypi-remote", ron_repo_type: "pypi-remote", ron_upstreams: [], ron_in: None, ron_out: None,
ron_base_url: Some("https://pypi.example")),
(ron_name: "pip", ron_repo_type: "pip", ron_upstreams: ["pypi-remote"], ron_in: None, ron_out: None),
],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:1", ron_http_url: Some("127.0.0.1:2")) ],
storage_endpoints: [ (ron_name: "store", ron_storage_type: "znippy", ron_path: "/data/store") ],
mirror: Some((ron_sources: [ (ron_name: "cloud", ron_kind: Holger, ron_endpoint: "http://10.77.0.9:50051", ron_every_secs: 300,
ron_repositories: [ (ron_from: "a", ron_to: "a") ]) ])),
)"#;
fn ladder_ron() -> String {
#[cfg(feature = "intake")]
return LADDER.replace("\n)", "\n promotion: Some(( ladder_name: \"raw\", stages: [ (name: \"dev\", repo: \"a\"), (name: \"prod\", repo: \"b\") ] )),\n)");
#[cfg(not(feature = "intake"))]
return LADDER.to_string();
}
/// ★ **With no source for a counter it reads "not recorded", never 0**; with the source,
/// the number.
#[test]
fn a_counter_with_no_source_reads_not_recorded_and_with_one_its_number() {
let shape = FlowGraph::from_config(&holger_of(&ladder_ron()));
let bare = flow::measure(&shape, &Measured::default());
for label in ["fetches 24 h", "fetches total", "sent 24 h", "sent total"] {
assert_eq!(
graph_row(&bare, "a", label),
Cell::NotRecorded,
"{label} with audit off"
);
}
assert_eq!(
graph_row(&bare, "a", "artifacts"),
Cell::Unlisted,
"not listed is not zero"
);
assert_eq!(graph_row(&bare, "storage:store", "used"), Cell::NotRecorded);
assert_eq!(
graph_row(&bare, "mirror:cloud", "last run"),
Cell::NotRecorded
);
assert_eq!(
graph_row(&bare, flow::PULLERS_NODE, "pullers"),
Cell::NotRecorded
);
let svg = bare.render_svg();
assert!(
!svg.contains("data-row=\"a/fetches total\" x")
|| row(&svg, "a", "fetches total") == "not recorded"
);
let now = 1_800_000_000i64 * 1_000_000_000;
let m = Measured {
now_unix: 1_800_000_000,
listing: BTreeMap::from([("a".to_string(), Some((2, 300)))]),
audit: Some(AuditTally::fold(
&[
download("a", "10.0.0.5:1", 300, now),
download("a", "10.0.0.9:1", 10, now),
],
now,
)),
storage: BTreeMap::from([("/data/store".to_string(), (1000, 400))]),
..Default::default()
};
let g = flow::measure(&shape, &m);
assert_eq!(graph_row(&g, "a", "artifacts"), Cell::Count(2));
assert_eq!(graph_row(&g, "a", "bytes stored"), Cell::Bytes(300));
assert_eq!(graph_row(&g, "a", "fetches total"), Cell::Count(2));
assert_eq!(graph_row(&g, "a", "sent 24 h"), Cell::Bytes(310));
assert_eq!(
graph_row(&g, "b", "fetches total"),
Cell::Count(0),
"audit on and nothing fetched: a real zero"
);
assert_eq!(graph_row(&g, "storage:store", "used"), Cell::Bytes(400));
assert_eq!(graph_row(&g, "storage:store", "free"), Cell::Bytes(600));
assert_eq!(graph_row(&g, flow::PULLERS_NODE, "pullers"), Cell::Count(2));
assert!(
g.edges
.iter()
.any(|e| e.kind == flow::EdgeKind::Served && e.from == "a"),
"the repository that sent bytes points at the pullers"
);
}
/// ★ **The pullers are one box, the busiest first by bytes, named by hash, capped** — the
/// total is not capped.
#[test]
fn the_pullers_are_one_box_listing_the_busiest_by_bytes() {
let shape = FlowGraph::from_config(&holger_of(&ladder_ron()));
let now = 1_800_000_000i64 * 1_000_000_000;
let events: Vec<_> = (0..12)
.map(|i| download("a", &format!("10.0.0.{i}:1"), 100 * (i + 1) as u64, now))
.collect();
let audit = AuditTally::fold(&events, now);
let labels = audit
.clients
.keys()
.map(|k| (k.clone(), format!("client {k}")))
.collect();
let g = flow::measure(
&shape,
&Measured {
audit: Some(audit),
puller_labels: labels,
..Default::default()
},
);
let p = g.node(flow::PULLERS_NODE).unwrap();
assert_eq!(
graph_row(&g, flow::PULLERS_NODE, "pullers"),
Cell::Count(12),
"the true total"
);
let named: Vec<&str> = p
.rows
.iter()
.filter(|r| r.label.starts_with("client "))
.map(|r| r.label.as_str())
.collect();
assert_eq!(named.len(), shape.max_consumers, "capped at max_consumers");
assert_eq!(named[0], "client 10.0.0.11", "the busiest by bytes first");
assert_eq!(
g.nodes
.iter()
.filter(|n| n.kind == NodeKind::Consumer)
.count(),
1,
"one box, never one per address"
);
}
/// ★ **The proxy, the mirror, the door and the promotion counters move.**
#[test]
fn the_proxy_mirror_door_and_promotion_counters_move() {
let shape = FlowGraph::from_config(&holger_of(&ladder_ron()));
let mut m = Measured {
now_unix: 1_800_000_000,
..Default::default()
};
m.upstreams
.push(holger_server_lib::upstreams::UpstreamFact {
name: "pypi-remote".into(),
kind: "proxy".into(),
answered: 5,
fetched: 3,
failed: 1,
last_ok_ms: 1_799_999_940_000,
..Default::default()
});
m.mirrors.insert(
"cloud".into(),
MirrorView {
last_success_unix: Some(1_799_999_900),
cycles_ok: 4,
fetched: 17,
cycles_failed: 2,
..Default::default()
},
);
m.door_errors.insert("door".into(), 2);
let g = flow::measure(&shape, &m);
assert_eq!(graph_row(&g, "pip", "proxy hits"), Cell::Count(3));
assert_eq!(graph_row(&g, "pip", "proxy misses"), Cell::Count(2));
let pull = g
.edges
.iter()
.find(|e| e.kind == flow::EdgeKind::Pull)
.unwrap();
assert_eq!(
pull.cell.render(),
"on miss · 3 fetched · last 1m ago · 1 errors"
);
assert_eq!(
graph_row(&g, "mirror:cloud", "items synced"),
Cell::Count(17)
);
assert_eq!(graph_row(&g, "mirror:cloud", "failures"), Cell::Count(2));
assert_eq!(
graph_row(&g, "mirror:cloud", "last run"),
Cell::Text("1m 40s ago".into())
);
assert_eq!(
graph_row(&g, "mirror:cloud", "next run"),
Cell::Text("in 3m 20s".into())
);
assert_eq!(
graph_row(&g, "mirror:cloud", "lag"),
Cell::Text("on time".into())
);
assert_eq!(
graph_row(&g, "door:door", "errors since boot"),
Cell::Count(2)
);
#[cfg(feature = "intake")]
{
let none = g
.edges
.iter()
.find(|e| e.kind == flow::EdgeKind::Promotion)
.unwrap();
assert_eq!(
none.cell,
Cell::NotRecorded,
"no audit, no intake: no source"
);
let now = 1_800_000_000i64 * 1_000_000_000;
let mut p = holger_server_lib::audit::AuditEvent::new(
"admin",
holger_server_lib::audit::AuditAction::Promote,
"b",
"x@1",
"",
200,
0,
);
p.ts_nanos = now - 120_000_000_000;
m.audit = Some(AuditTally::fold(&[p], now));
let g = flow::measure(&shape, &m);
let e = g
.edges
.iter()
.find(|e| e.kind == flow::EdgeKind::Promotion)
.unwrap();
assert_eq!(
e.cell.render(),
"1 · last 2m ago",
"a promotion moves its arrow"
);
}
}
/// ★ **The door's 5xx answers and the scanners' findings are counted where they happen.**
#[test]
fn the_door_errors_and_the_scan_findings_move_in_the_registry() {
let m = holger_server_lib::metrics::global();
let before = m.door_errors().get("flow-test-door").copied().unwrap_or(0);
m.record_door_status("flow-test-door", 200);
m.record_door_status("flow-test-door", 503);
assert_eq!(
m.door_errors()["flow-test-door"],
before + 1,
"a 5xx moves it, a 200 does not"
);
let s0 = m
.scan_tally()
.get("flow-test-kind")
.cloned()
.unwrap_or_default();
m.record_scan("flow-test-kind", "block", ["high", "critical", "high"]);
let s1 = m.scan_tally()["flow-test-kind"].clone();
assert_eq!(s1.scanned, s0.scanned + 1);
assert_eq!(
s1.findings["high"],
s0.findings.get("high").copied().unwrap_or(0) + 2
);
assert_eq!(
s1.verdicts["block"],
s0.verdicts.get("block").copied().unwrap_or(0) + 1
);
}
/// ★ **The intake steps and gates carry the intake record's counts**: in queue, passed,
/// rejected, quarantined, waiting for approval; findings by severity; provenance
/// verified vs failed.
#[cfg(all(feature = "supply-chain", feature = "sync"))]
#[test]
fn the_intake_steps_carry_the_records_counts() {
let shape = FlowGraph::from_config(&holger_of(&ron("")));
let bare = flow::measure(&shape, &Measured::default());
assert_eq!(
graph_row(&bare, "step:scan · vulnerability", "in queue"),
Cell::NotRecorded,
"no chain record: no number"
);
let intake = IntakeTally {
states: BTreeMap::from([
("INGESTED".into(), 2),
("SCANNED".into(), 1),
("POLICY_CHECKED".into(), 4),
("BLOCKED".into(), 1),
("QUARANTINED".into(), 1),
]),
held_from: BTreeMap::from([("SCANNED".into(), 1)]),
verdicts: BTreeMap::from([
("scan".into(), (5, 1, 2)),
("license".into(), (6, 0, 1)),
("approval".into(), (3, 0, 0)),
]),
hops: BTreeMap::from([(("quarantine".into(), "approved".into()), (3, 1_799_999_040))]),
};
let mut scans = BTreeMap::new();
let vuln = flow::ScanTally {
scanned: 8,
findings: BTreeMap::from([("critical".into(), 1), ("medium".into(), 4)]),
..Default::default()
};
scans.insert("vulnerability".to_string(), vuln);
let g = flow::measure(
&shape,
&Measured {
now_unix: 1_800_000_000,
intake: Some(intake),
scans,
..Default::default()
},
);
let v = "step:scan · vulnerability";
assert_eq!(graph_row(&g, v, "in queue"), Cell::Count(3));
assert_eq!(graph_row(&g, v, "passed"), Cell::Count(6));
assert_eq!(graph_row(&g, v, "rejected"), Cell::Count(2));
assert_eq!(graph_row(&g, v, "quarantined"), Cell::Count(1));
assert_eq!(graph_row(&g, v, "findings critical"), Cell::Count(1));
assert_eq!(graph_row(&g, v, "findings medium"), Cell::Count(4));
assert_eq!(
graph_row(&g, "step:scan · license", "passed"),
Cell::Count(6)
);
assert_eq!(
graph_row(&g, "step:scan · license", "findings"),
Cell::NotRecorded,
"no license scan since boot"
);
assert_eq!(
graph_row(&g, "step:approval", "waiting for approval"),
Cell::Count(4)
);
assert_eq!(
graph_row(&g, "step:approval", "approvals cast"),
Cell::Count(3)
);
let hop = g
.edges
.iter()
.find(|e| e.from == "quarantine" && e.to == "approved")
.unwrap();
assert_eq!(
hop.cell.render(),
"3 · last 16m ago",
"the intake record's hops move the promotion arrow"
);
}
/// A repository name cannot close a tag; the page fetches nothing off the box.
#[test]
fn the_drawing_is_escaped_and_fetches_nothing() {
let h = holger_of(
r#"(
repositories: [
(ron_name: "a<b>&\"c", ron_repo_type: "raw", ron_upstreams: [], ron_in: None, ron_out: None),
],
exposed_endpoints: [],
storage_endpoints: [],
)"#,
);
let g = flow::measure(&FlowGraph::from_config(&h), &Measured::default());
let page = g.render_page(30);
assert!(!page.contains("a<b>"), "escaped");
for bad in ["<script", "<link", "src=", "@import", "http://", "https://"] {
let hits: Vec<_> = page
.match_indices(bad)
.filter(|(i, _)| !page[*i..].starts_with("http://www.w3.org/2000/svg"))
.collect();
assert!(hits.is_empty(), "the page reaches nothing: {bad}");
}
for u in page.match_indices("url(").map(|(i, _)| &page[i..i + 6]) {
assert!(u.starts_with("url(#"), "same-document fragments only: {u}");
}
}
/// A proxy arrow and a mirror arrow are different kinds with different strokes.
#[test]
fn a_mirror_edge_is_visibly_not_a_proxy_edge() {
let svg = flow::measure(
&FlowGraph::from_config(&holger_of(&ladder_ron())),
&Measured::default(),
)
.render_svg();
assert!(svg.contains("data-edge=\"mirror:cloud->a\" data-kind=\"mirror\""));
assert!(svg.contains("data-edge=\"pypi-remote->pip\" data-kind=\"pull\""));
assert!(
svg.contains(">mirror, scheduled</text>")
&& svg.contains(">proxy upstream, on miss</text>"),
"the legend names both"
);
assert!(
!svg.contains(">no sources configured</text>"),
"a remote and a mirror are sources"
);
}
/// ★ **The open build speaks no money**: a paid repository's disk cap is a row of the metered
/// build only; the open build draws no such row, whatever it is handed.
#[test]
fn the_disk_cap_row_is_the_metered_builds_only() {
let shape = FlowGraph::from_config(&holger_of(&ladder_ron()));
let m = Measured {
quota: BTreeMap::from([("a".to_string(), 5_000_000)]),
..Default::default()
};
let g = flow::measure(&shape, &m);
let has = g
.node("a")
.unwrap()
.rows
.iter()
.any(|r| r.cell == Cell::Bytes(5_000_000));
assert_eq!(
has,
cfg!(feature = "metered"),
"the cap row exists exactly in a metered build"
);
}