server/lib/tests/flow_door.rs
//! **The artifact flow's doors, against a running server** (restored by lane HOL-SVG-RON
//! from the file the lean cut deleted, holger 9521bf06, and brought to today's doors).
//!
//! What it holds, on a REAL holger over real loopback ports:
//!
//! * the page, the bare SVG and the JSON model are served to an Admin (and an Auditor),
//! never cached; a Writer is refused (403), nobody unauthenticated gets it (401);
//! * `HEAD` answers the door, a write gets 405 — never the repository router's 404;
//! * **the console's picture (`AdminService.FlowSvg`) and the door's (`GET /-/flow.svg`)
//! are one drawing**: the same nodes, edges and declared facts, from the same function;
//! * the drawing is the RON: the repositories, the ladder's rung, the storage endpoint and
//! the listener; changing what happens moves the counters (a download moves its
//! repository's fetches and bytes sent, and the door's requests);
//! * a `curl` on loopback never becomes a node;
//! * `flow: (enabled: false)` makes every flow door 404 and `FlowSvg` NotFound.
use std::io::Write;
use traits::test_support::hold_port;
use holger_server_lib::grpc::holger_proto::admin_service_client::AdminServiceClient;
use holger_server_lib::grpc::holger_proto::FlowSvgRequest;
use holger_server_lib::{read_ron_config, wire_holger, Holger};
const PRINCIPALS: [&str; 3] = ["admin", "writer", "aud"];
fn token(who: &str) -> String {
format!("{who}-flow-token-0123456789abcdefghij")
}
struct Running {
_data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
_holger: Holger,
base: String,
grpc: String,
}
fn start(extra: &str) -> Running {
let mut last = String::new();
for _ in 0..8 {
let (grpc, http) = (hold_port(), hold_port());
match try_start(grpc, http, extra) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
/// A two-rung ladder (`dev` in `a`, `prod` in `b`) when the build has one, a store and a door.
fn try_start(grpc: u16, http: u16, extra: &str) -> Result<Running, String> {
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_string();
let methods: Vec<String> = PRINCIPALS
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
#[cfg(feature = "intake")]
let ladder = r#"promotion: Some(( ladder_name: "raw", stages: [ (name: "dev", repo: "a"), (name: "prod", repo: "b") ] )),"#;
#[cfg(not(feature = "intake"))]
let ladder = "";
let ron = format!(
r#"(
repositories: [
(ron_name: "a", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{d}/a")),
(ron_name: "b", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{d}/b")),
],
exposed_endpoints: [
(ron_name: "default", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")),
],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
auth: Some((
methods: [ {methods} ],
roles: {{"admin": Admin, "writer": Writer, "aud": Reader}},
duties: {{"aud": [Auditor]}},
)),
{ladder}
{extra}
)"#,
methods = methods.join(", "),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
let mut holger: Holger = read_ron_config(cfg.path()).map_err(|e| format!("config: {e}"))?;
holger
.instantiate_backends()
.map_err(|e| format!("backends: {e}"))?;
wire_holger(&mut holger).map_err(|e| format!("wire: {e}"))?;
holger.start().map_err(|e| format!("start: {e}"))?;
Ok(Running {
_data: data,
_cfg: cfg,
_holger: holger,
base: format!("http://127.0.0.1:{http}"),
grpc: format!("http://127.0.0.1:{grpc}"),
})
}
async fn wait_healthy(c: &reqwest::Client, base: &str) {
for _ in 0..200 {
if let Ok(r) = c.get(format!("{base}/healthz")).send().await {
if r.status() == 200 {
return;
}
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
panic!("gateway never became healthy at {base}");
}
async fn get(c: &reqwest::Client, url: String, who: Option<&str>) -> reqwest::Response {
let mut r = c.get(url);
if let Some(w) = who {
r = r.bearer_auth(token(w));
}
r.send().await.expect("the door answers")
}
async fn flow_svg(
endpoint: &str,
who: Option<&str>,
) -> Result<holger_server_lib::grpc::holger_proto::FlowSvgReply, tonic::Status> {
flow_svg_open(endpoint, who, "").await
}
/// `AdminService.FlowSvg` at one zoom (`open` as `?open=` names it).
async fn flow_svg_open(
endpoint: &str,
who: Option<&str>,
open: &str,
) -> Result<holger_server_lib::grpc::holger_proto::FlowSvgReply, tonic::Status> {
let mut last = None;
for _ in 0..100 {
match AdminServiceClient::connect(endpoint.to_string()).await {
Ok(mut client) => {
let mut req = tonic::Request::new(FlowSvgRequest {
open: open.to_string(),
});
if let Some(w) = who {
req.metadata_mut().insert(
"authorization",
format!("Bearer {}", token(w)).parse().unwrap(),
);
}
return client.flow_svg(req).await.map(|r| r.into_inner());
}
Err(e) => last = Some(e),
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
panic!("the gRPC door never answered on {endpoint}: {last:?}");
}
fn header(r: &reqwest::Response, h: &str) -> String {
r.headers()
.get(h)
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_string()
}
/// A row's value in the JSON model.
fn model_row(v: &serde_json::Value, node: &str, label: &str) -> serde_json::Value {
let n = v["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["id"] == node)
.unwrap_or_else(|| panic!("node {node}"));
n["rows"]
.as_array()
.unwrap()
.iter()
.find(|r| r["label"] == label)
.unwrap_or_else(|| panic!("row {node}/{label}: {}", n["rows"]))["cell"]
.clone()
}
fn ids(
v: &serde_json::Value,
what: &str,
key: &dyn Fn(&serde_json::Value) -> String,
) -> std::collections::BTreeSet<String> {
v[what].as_array().unwrap().iter().map(key).collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn the_flow_door_serves_the_configured_graph_to_an_admin_only() {
let r = start("");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
let page = get(&c, format!("{}/-/flow?open=all", r.base), Some("admin")).await;
assert_eq!(page.status(), 200);
assert!(header(&page, "content-type").starts_with("text/html"));
assert!(header(&page, "cache-control").contains("no-store"));
let body = page.text().await.unwrap();
assert!(
body.contains("<svg") && !body.contains("<script"),
"the page is the drawing, no script"
);
let svg = get(&c, format!("{}/-/flow.svg?open=all", r.base), Some("admin")).await;
assert_eq!(svg.status(), 200);
assert!(header(&svg, "content-type").starts_with("image/svg+xml"));
let svg = svg.text().await.unwrap();
for node in [
"a",
"b",
"storage:storage",
"door:default",
"observed pullers",
] {
assert!(
svg.contains(&format!("data-node=\"{node}\"")),
"{node} is drawn"
);
}
assert!(
svg.contains("data-edge=\"a->storage:storage\"")
&& svg.contains("data-edge=\"a->door:default\""),
"the storage and the listener are boxes with arrows"
);
#[cfg(feature = "intake")]
assert!(
svg.contains("data-edge=\"a->b\" data-kind=\"promote\""),
"the ladder's rung"
);
let json = get(&c, format!("{}/-/flow.json", r.base), Some("aud")).await;
assert_eq!(json.status(), 200, "an Auditor reads it too");
assert!(header(&json, "content-type").starts_with("application/json"));
let v: serde_json::Value = json.json().await.unwrap();
assert_eq!(v["schema"], "holger.flow/v2");
assert_eq!(
get(&c, format!("{}/-/flow.svg", r.base), Some("writer"))
.await
.status(),
403,
"a Writer is refused"
);
assert_eq!(
get(&c, format!("{}/-/flow.svg", r.base), None)
.await
.status(),
401,
"nobody unauthenticated"
);
let head = c
.head(format!("{}/-/flow.svg", r.base))
.bearer_auth(token("admin"))
.send()
.await
.unwrap();
assert_eq!(head.status(), 200, "HEAD is answered by the door");
let post = c
.post(format!("{}/-/flow", r.base))
.bearer_auth(token("admin"))
.body("x")
.send()
.await
.unwrap();
assert_eq!(
post.status(),
405,
"a write is refused by the door, not routed to a repository"
);
}
/// ★ **The console and the door draw ONE picture**: `AdminService.FlowSvg` (what the
/// console shows) and `GET /-/flow.svg` / `/-/flow.json` carry the same nodes, edges and
/// declared facts — one function draws both.
#[tokio::test(flavor = "multi_thread")]
async fn the_console_and_the_door_serve_the_same_drawing() {
let r = start("");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
let over_grpc = flow_svg(&r.grpc, Some("admin"))
.await
.expect("an Admin reads the flow over gRPC");
let over_http: serde_json::Value = get(&c, format!("{}/-/flow.json", r.base), Some("admin"))
.await
.json()
.await
.unwrap();
let http_svg = get(&c, format!("{}/-/flow.svg", r.base), Some("admin"))
.await
.text()
.await
.unwrap();
let grpc_model: serde_json::Value = serde_json::from_str(&over_grpc.graph_json).unwrap();
let node = |n: &serde_json::Value| n["id"].as_str().unwrap().to_string();
let edge = |e: &serde_json::Value| {
format!(
"{}->{} {}",
e["from"].as_str().unwrap(),
e["to"].as_str().unwrap(),
e["kind"].as_str().unwrap()
)
};
let facts = |n: &serde_json::Value| format!("{} {}", n["id"], n["facts"]);
assert_eq!(
ids(&grpc_model, "nodes", &node),
ids(&over_http, "nodes", &node),
"the same boxes"
);
assert_eq!(
ids(&grpc_model, "edges", &edge),
ids(&over_http, "edges", &edge),
"the same arrows"
);
assert_eq!(
ids(&grpc_model, "nodes", &facts),
ids(&over_http, "nodes", &facts),
"the same declared facts"
);
let tags = |s: &str| {
s.match_indices("data-node=\"")
.map(|(i, _)| s[i..].split('"').nth(1).unwrap().to_string())
.collect::<std::collections::BTreeSet<_>>()
};
assert_eq!(tags(&over_grpc.svg), tags(&http_svg), "the same drawing");
assert!(over_grpc.svg.starts_with("<svg"));
let writer = flow_svg(&r.grpc, Some("writer"))
.await
.expect_err("a Writer is refused");
assert_eq!(writer.code(), tonic::Code::PermissionDenied);
let anon = flow_svg(&r.grpc, None)
.await
.expect_err("anonymous is refused");
assert_eq!(anon.code(), tonic::Code::Unauthenticated);
}
/// ★ **The counters move when the thing happens**: an upload and a download through the
/// door move the repository's artifacts, fetches and bytes sent, and the door's requests.
/// A loopback client is never a node.
#[tokio::test(flavor = "multi_thread")]
async fn a_download_moves_the_counters_and_loopback_is_never_a_node() {
let r = start("");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
let model = |c: reqwest::Client, base: String| async move {
get(&c, format!("{base}/-/flow.json?open=main"), Some("admin"))
.await
.json::<serde_json::Value>()
.await
.unwrap()
};
let before = model(c.clone(), r.base.clone()).await;
assert_eq!(
model_row(&before, "a", "fetches total")["value"],
0,
"audit on, nothing fetched: a real zero"
);
let req_before = model_row(&before, "door:default", "requests today")["value"]
.as_u64()
.unwrap();
let put = c
.put(format!("{}/a/thing.bin", r.base))
.bearer_auth(token("admin"))
.body(vec![7u8; 1234])
.send()
.await
.unwrap();
assert!(put.status().is_success(), "upload: {}", put.status());
// The visitor tally counts a client, never the box talking to itself: this one comes
// through a proxy on loopback, which names it (X-Forwarded-For, believed from loopback only).
let got = c
.get(format!("{}/a/thing.bin", r.base))
.bearer_auth(token("admin"))
.header("x-forwarded-for", "10.9.8.7")
.send()
.await
.unwrap();
assert_eq!(got.status(), 200);
assert_eq!(got.bytes().await.unwrap().len(), 1234);
let after = model(c.clone(), r.base.clone()).await;
assert_eq!(
model_row(&after, "a", "artifacts")["value"],
1,
"the upload moved the listing"
);
assert_eq!(
model_row(&after, "a", "fetches total")["value"],
1,
"the download moved the fetches"
);
assert!(
model_row(&after, "a", "sent total")["value"]
.as_u64()
.unwrap()
>= 1234,
"and the bytes sent"
);
assert!(
model_row(&after, "door:default", "requests today")["value"]
.as_u64()
.unwrap()
> req_before,
"and the door's requests"
);
// The pullers are one box; the box talking to itself is a row of it, never a node, and
// no address is ever a row's name.
let pullers = after["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["id"] == "observed pullers")
.unwrap()
.clone();
let named: Vec<String> = pullers["rows"]
.as_array()
.unwrap()
.iter()
.map(|r| r["label"].as_str().unwrap().to_string())
.collect();
assert!(
named
.iter()
.all(|l| !l.contains("127.0.0.1") && !l.contains("10.9.8.7")),
"no address names a row: {named:?}"
);
let fetched = model_row(&after, "observed pullers", "fetches")["value"]
.as_u64()
.unwrap()
+ model_row(&after, "observed pullers", "loopback")["value"]
.as_u64()
.unwrap();
assert_eq!(
fetched, 1,
"the one download is counted once, as a puller or as loopback"
);
assert!(
after["nodes"]
.as_array()
.unwrap()
.iter()
.all(|n| !n["id"].as_str().unwrap().contains("127.0.0.1")),
"an address is never a node"
);
}
/// ★ **The flow counters are the audit aggregates** (lane HOL-AUDIT): real downloads by
/// two clients move fetches, bytes sent and the pullers; the numbers are read from the
/// audit log's Arrow index (it exists and reaches the downloads' rows), and deleting the
/// index changes nothing: it is rebuilt from the chain on the next read.
#[tokio::test(flavor = "multi_thread")]
async fn the_flow_counters_move_after_real_downloads_and_come_from_the_audit_index() {
let r = start("");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
let model = |c: reqwest::Client, base: String| async move {
get(&c, format!("{base}/-/flow.json?open=main"), Some("admin"))
.await
.json::<serde_json::Value>()
.await
.unwrap()
};
let put = c
.put(format!("{}/a/big.bin", r.base))
.bearer_auth(token("admin"))
.body(vec![1u8; 2000])
.send()
.await
.unwrap();
assert!(put.status().is_success());
for client in ["10.1.1.1", "10.1.1.1", "10.2.2.2"] {
let got = c
.get(format!("{}/a/big.bin", r.base))
.bearer_auth(token("admin"))
.header("x-forwarded-for", client)
.send()
.await
.unwrap();
assert_eq!(got.status(), 200);
assert_eq!(got.bytes().await.unwrap().len(), 2000);
}
let after = model(c.clone(), r.base.clone()).await;
assert_eq!(model_row(&after, "a", "fetches total")["value"], 3);
assert_eq!(model_row(&after, "a", "fetches 24 h")["value"], 3);
assert!(
model_row(&after, "a", "sent total")["value"]
.as_u64()
.unwrap()
>= 6000
);
let pullers = model_row(&after, "observed pullers", "pullers")["value"]
.as_u64()
.unwrap();
let loopback = model_row(&after, "observed pullers", "loopback")["value"]
.as_u64()
.unwrap();
let fetched = model_row(&after, "observed pullers", "fetches")["value"]
.as_u64()
.unwrap();
assert_eq!(fetched + loopback, 3, "each download counted once");
let index = r
._data
.path()
.join("audit")
.join(holger_server_lib::audit::arrow_ipc::INDEX_DIR);
let manifest: serde_json::Value = serde_json::from_slice(
&std::fs::read(index.join("MANIFEST.json")).expect("the flow read the audit index"),
)
.unwrap();
let downloads = holger_server_lib::audit::FileAuditLog::read_dir(r._data.path().join("audit"))
.unwrap()
.iter()
.filter(|e| e.action == holger_server_lib::audit::AuditAction::Download)
.count();
assert!(
downloads >= 3 && manifest["rows"].as_u64().unwrap() >= downloads as u64,
"the index reaches the downloads: {manifest}"
);
std::fs::remove_dir_all(&index).unwrap();
let again = model(c.clone(), r.base.clone()).await;
assert_eq!(
model_row(&again, "a", "fetches total")["value"],
3,
"rebuilt from the chain, the same counters"
);
assert_eq!(
model_row(&again, "observed pullers", "pullers")["value"]
.as_u64()
.unwrap(),
pullers
);
assert!(index.join("MANIFEST.json").is_file(), "the index is back");
}
#[tokio::test(flavor = "multi_thread")]
async fn a_disabled_flow_block_removes_every_flow_door() {
let r = start("flow: (enabled: false),");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
for door in ["/-/flow", "/-/flow.svg", "/-/flow.json"] {
assert_eq!(
get(&c, format!("{}{door}", r.base), Some("admin"))
.await
.status(),
404,
"{door}"
);
}
let off = flow_svg(&r.grpc, Some("admin"))
.await
.expect_err("the flow is off");
assert_eq!(off.code(), tonic::Code::NotFound);
}
/// ★ **The doors serve the zoom `open` names** (HOL-FLOW-SMART): no `open` is level 1 (the
/// Overview: the machines and the sources), `?open=main` unfolds main (the Flow), `?open=all`
/// is the full RON; `AdminService.FlowSvg` takes the same words and draws the same picture.
#[tokio::test(flavor = "multi_thread")]
async fn the_doors_serve_the_level_open_names() {
let r = start("");
let c = reqwest::Client::new();
wait_healthy(&c, &r.base).await;
let model = |open: &'static str| {
let (c, base) = (c.clone(), r.base.clone());
async move {
get(&c, format!("{base}/-/flow.json{open}"), Some("admin"))
.await
.json::<serde_json::Value>()
.await
.unwrap()
}
};
let node = |n: &serde_json::Value| n["id"].as_str().unwrap().to_string();
let one = model("").await;
assert_eq!(one["level"], 1, "no open is level 1");
let at_one = ids(&one, "nodes", &node);
assert!(at_one.contains("estate:main"), "{at_one:?}");
assert!(
!at_one.contains("a"),
"a repository is folded into main: {at_one:?}"
);
let two = model("?open=main").await;
assert_eq!(two["level"], 2);
let at_two = ids(&two, "nodes", &node);
for part in ["a", "b", "door:default", "storage:storage"] {
assert!(at_two.contains(part), "{part} unfolded: {at_two:?}");
}
assert_eq!(model("?open=all").await["level"], 3);
assert_eq!(
model("?open=main%2Call").await["level"],
3,
"the words may come percent-encoded"
);
let svg = get(&c, format!("{}/-/flow.svg", r.base), Some("admin"))
.await
.text()
.await
.unwrap();
assert!(svg.contains("data-level=\"1\""));
assert!(
svg.contains("<a href=\"flow?open=main\""),
"main unfolds by a link"
);
let page = get(&c, format!("{}/-/flow?open=main", r.base), Some("admin"))
.await
.text()
.await
.unwrap();
assert!(
page.contains("data-level=\"2\"") && page.contains("<a href=\"flow?open=\""),
"the page at level 2 folds main again"
);
let level_one = flow_svg_open(&r.grpc, Some("admin"), "").await.unwrap();
assert!(
level_one
.links
.iter()
.any(|l| l.href == "flow?open=main" && l.label == "unfold main"),
"the console is told where main unfolds: {:?}",
level_one.links
);
for open in ["", "main", "all"] {
let over_grpc = flow_svg_open(&r.grpc, Some("admin"), open)
.await
.expect("an Admin reads the flow at any level");
let grpc: serde_json::Value = serde_json::from_str(&over_grpc.graph_json).unwrap();
let query = if open.is_empty() {
String::new()
} else {
format!("?open={open}")
};
let http: serde_json::Value =
get(&c, format!("{}/-/flow.json{query}", r.base), Some("admin"))
.await
.json()
.await
.unwrap();
assert_eq!(grpc["level"], http["level"], "open={open}");
assert_eq!(
ids(&grpc, "nodes", &node),
ids(&http, "nodes", &node),
"the console and the door draw one picture at open={open}"
);
}
}