tests/collectors_registry_sweep.rs
//! **The registry sweep** — the two laws this leaf claims, asserted over EVERY
//! collector in [`registry()`] instead of over the handful someone remembered to
//! write a test for.
//!
//! Before this file the coverage was per-collector and hand-maintained: 58
//! registered collectors, an offline-safe sample for 41 of them, and a
//! `collect()`-level RAGNARÖK assertion for exactly **four** (polisen, dcat, csv,
//! mainframe). Adding a collector that forgot either could not turn anything red,
//! because nothing iterated the registry. These two sweeps do:
//!
//! * [`every_registered_collector_has_an_offline_safe_sample`] — the sample seam
//! ([`sample::for_collector`]) must answer for every registered name, and the
//! body must be non-empty. RED the moment a collector lands without one.
//! * [`no_registered_collector_is_silently_green_on_an_empty_source`] — drive
//! every collector's `collect()` against an EMPTY-but-well-formed source and
//! assert it never comes back `Ok` with 0 rows and no reason. That is RAGNARÖK,
//! swept: empty is RED, never silently green.
//! * [`every_registered_collector_parses_its_own_sample`] — drive `collect()`
//! against the collector's own offline-safe sample and assert rows land. RED
//! when a parse breaks.
//!
//! Every check emits a per-collector `functional_status` row under
//! `--features testmatrix`, so `nornir test` sees one matrix cell per
//! `(collector, law)` pair rather than one aggregate verdict.
#![cfg(feature = "parsers")]
// Used only by the full-registry sweeps and their fixtures, so they carry the
// sweeps' gate too — see the block comment below.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
use std::io::{Read, Write};
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
use std::net::TcpListener;
// What the always-on checks need: the registry, the sample seam, and RAGNARÖK.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
use korp_collectors::{
BatchSink, Bounds, Collector, CollectorCfg, CollectorMeta, SourceEndpoint, SourceKind,
SourceTarget,
};
use korp_collectors::{Ragnarok, registry, sample};
mod common;
use common::{emit, emit_for};
// ── Fixtures + drivers for the two FULL-REGISTRY sweeps below. ───────────────
// They carry the SAME gate as those sweeps: the sweeps are meaningful only when
// the whole ~58-collector set is compiled in, so under a partial build (the
// default `parsers`-only one) every item here is dead. Ungated they produced 11
// dead-code warnings that said nothing except "you are not building the full
// matrix" — noise that would hide a genuinely dead helper.
/// Serve ONE `200 OK` response with `body`, then exit. Returns the base URL.
///
/// The handle is deliberately not joined: a collector that (correctly) errors
/// before issuing its request leaves the accept blocked, and the sweep must not
/// hang on it. Each call binds its own ephemeral port, so the servers of 58
/// collectors never collide.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn spawn_once(body: Vec<u8>) -> String {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || {
if let Ok((mut stream, _)) = listener.accept() {
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.write_all(&body);
let _ = stream.flush();
}
});
format!("http://{addr}")
}
/// How a collector's bytes reach it.
#[derive(Clone, Copy, PartialEq, Eq)]
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
enum Feed {
/// Point `base_url` at an in-process HTTP server serving the payload.
Http,
/// Write the payload to a temp file and point `base_url` (and `params.path`)
/// at it — the LOCAL flat-file collectors (`csv`, `mainframe-copybook`,
/// `fixed-width`) and the OSM extract readers take a path, not a URL.
LocalFile,
}
/// The sweep's per-collector plan.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
struct Plan {
feed: Feed,
params: serde_json::Value,
/// `Some(owner)` when the collector's sample is a PARSER-LEVEL fixture rather
/// than the collector's own wire format, so `collect()` cannot be driven from
/// it. Such a collector is still swept for its sample (#1) and for RAGNARÖK
/// (#2, fed an empty source); only the positive parse sweep defers to the
/// dedicated test named here.
parse_deferred_to: Option<&'static str>,
}
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
impl Plan {
fn http() -> Self {
Self {
feed: Feed::Http,
params: serde_json::Value::Null,
parse_deferred_to: None,
}
}
fn local(params: serde_json::Value) -> Self {
Self {
feed: Feed::LocalFile,
params,
parse_deferred_to: None,
}
}
fn deferring(mut self, owner: &'static str) -> Self {
self.parse_deferred_to = Some(owner);
self
}
}
/// Classify a collector: how to feed it, the extra `params` it needs, and whether
/// the positive parse sweep can drive it.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn plan_for(meta: &CollectorMeta, st: &SourceTarget) -> Plan {
let _ = st;
match meta.name.as_str() {
// Local flat files, laid out against a copybook.
"mainframe-copybook" | "fixed-width" => {
Plan::local(serde_json::json!({ "copybook": sample_copybook() }))
}
"csv" => Plan::local(serde_json::Value::Null),
// OSM: `osm-katana` reads a container from `params.path`.
"osm-geoparquet" => Plan::local(serde_json::Value::Null),
// The admin-boundary sample is the `extract_admin_units` tag-map input,
// not a PBF — the tag-filter/trie contract lives in its own test — but the
// RAGNARÖK sweep still feeds it a real (empty) file.
"osm-admin-boundaries" => Plan::local(serde_json::Value::Null)
.deferring("osm_admin_sample_extracts_units (collectors_sample.rs)"),
// Everything else is a network collector fed over HTTP.
_ => Plan::http(),
}
}
/// The copybook the flat-file samples are laid out against.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn sample_copybook() -> String {
#[cfg(feature = "mainframe")]
{
sample::SYNTHETIC_MAINFRAME_COPYBOOK.to_string()
}
#[cfg(not(feature = "mainframe"))]
{
String::new()
}
}
/// An EMPTY-but-well-formed document in the collector's own wire format — the
/// input the RAGNARÖK sweep feeds.
///
/// The first choice is the collector's OWN offline-safe sample with every JSON
/// array emptied ([`empty_like`]): same document shape, same nesting, same keys —
/// so the collector's records pointer still RESOLVES and it reaches its 0-row
/// branch, which is the branch under test. A hand-written `{}` would instead make
/// most collectors error at pointer resolution: honest (an `Err` is not silently
/// green either) but vacuous, proving nothing about the RAGNARÖK reason.
///
/// Non-JSON wire shapes (XML, protobuf, zip, binary containers) fall back to a
/// per-[`SourceEndpoint`] empty document, so a NEW collector still gets a
/// sensible payload rather than silently skipping the sweep.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn empty_payload(meta: &CollectorMeta, st: &SourceTarget) -> Vec<u8> {
if let Some(s) = sample::for_collector(&meta.name)
&& let Some(emptied) = empty_like(&s.body)
{
return emptied;
}
let s: &str = match st.source {
SourceEndpoint::DcatCatalog => {
r#"<?xml version="1.0" encoding="utf-8"?><rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"></rdf:RDF>"#
}
SourceEndpoint::EntsoE | SourceEndpoint::EntsoeDayAhead => {
r#"<?xml version="1.0" encoding="UTF-8"?><GL_MarketDocument xmlns="urn:iec62325.351:tc57wg16:451-6:generationloaddocument:3:0"></GL_MarketDocument>"#
}
SourceEndpoint::Trafikverket => r#"{"RESPONSE":{"RESULT":[]}}"#,
SourceEndpoint::Wfs => {
r#"<?xml version="1.0" encoding="UTF-8"?><wfs:FeatureCollection xmlns:wfs="http://www.opengis.net/wfs/2.0"></wfs:FeatureCollection>"#
}
SourceEndpoint::Feed => {
r#"<?xml version="1.0" encoding="UTF-8"?><rss version="2.0"><channel><title>empty</title></channel></rss>"#
}
// A zero-length protobuf FeedMessage is a valid empty feed.
SourceEndpoint::GtfsRt => return Vec::new(),
// A feed zip with a header-only member: valid archive, zero data rows.
SourceEndpoint::Gtfs => return empty_gtfs_zip(),
// Overpass is JSON; the PBF readers take the LocalFile drive and get an
// empty file instead.
SourceEndpoint::Osm | SourceEndpoint::OsmAdmin => match meta.source_kind {
SourceKind::Rest => r#"{"elements":[]}"#,
_ => return Vec::new(),
},
SourceEndpoint::FlatFile => return Vec::new(),
// Every remaining family is JSON. An empty object resolves no records
// pointer, so the collector either reports an explained empty or errors —
// both honest, neither silently green.
_ => "{}",
};
s.as_bytes().to_vec()
}
/// `body` as JSON with **every array emptied**, recursively — the same document
/// shape carrying zero records. `None` when the body is not JSON.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn empty_like(body: &[u8]) -> Option<Vec<u8>> {
fn strip(v: &serde_json::Value) -> serde_json::Value {
match v {
serde_json::Value::Array(_) => serde_json::Value::Array(Vec::new()),
serde_json::Value::Object(m) => {
serde_json::Value::Object(m.iter().map(|(k, v)| (k.clone(), strip(v))).collect())
}
other => other.clone(),
}
}
let v: serde_json::Value = serde_json::from_slice(body).ok()?;
serde_json::to_vec(&strip(&v)).ok()
}
/// A structurally-valid GTFS feed zip whose members carry only headers.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn empty_gtfs_zip() -> Vec<u8> {
#[cfg(feature = "gtfs")]
{
korp_collectors::gtfs::build_zip(&[
(
"stops.txt",
8,
b"stop_id,stop_name,stop_lat,stop_lon\n".as_slice(),
),
("routes.txt", 8, b"route_id,route_type\n".as_slice()),
("trips.txt", 8, b"route_id,trip_id\n".as_slice()),
])
}
#[cfg(not(feature = "gtfs"))]
{
Vec::new()
}
}
/// Credentials the auth-gated collectors read straight from the ENVIRONMENT
/// (not through `cfg.auth`): without them those collectors error before they
/// ever reach their 0-row branch, which would let the RAGNARÖK sweep pass them
/// vacuously. Fake values are fine — the request goes to the in-process server.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
fn export_fixture_credentials() {
for (k, v) in [
("TRAFIKVERKET_API_KEY", "sweep-fixture-key"),
("CDSAPI_KEY", "sweep-fixture-key"),
("ADSAPI_KEY", "sweep-fixture-key"),
] {
// SAFETY: the sweep tests are the only writers and each runs before any
// collector thread is spawned.
unsafe { std::env::set_var(k, v) };
}
}
/// Build the cfg for one collector and run `collect()` over `payload`.
/// Returns `Ok(stats)` or the collector's error.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
async fn collect_over(
collector: &dyn Collector,
payload: Vec<u8>,
plan: &Plan,
dir: &std::path::Path,
) -> (anyhow::Result<korp_collectors::CollectStats>, BatchSink) {
let meta = collector.meta();
let sink = BatchSink::memory();
let params = plan.params.clone();
let payload_for_canned = payload.clone();
let (base_url, params) = match plan.feed {
Feed::Http => (spawn_once(payload), params),
Feed::LocalFile => {
let path = dir.join(format!("{}.bin", meta.name.replace(['/', ' '], "_")));
std::fs::write(&path, &payload).expect("write local fixture");
let p = path.to_string_lossy().to_string();
// The OSM readers take the path in `params.path`, not `base_url`.
let params = match params {
serde_json::Value::Object(mut m) => {
m.insert("path".into(), serde_json::json!(p));
serde_json::Value::Object(m)
}
_ => serde_json::json!({ "path": p }),
};
(p, params)
}
};
let mut cfg = CollectorCfg::new(base_url.clone(), format!("sweep_{}", meta.name.replace('-', "_")))
.with_params(params);
// **The sweep physically cannot reach the network** (the O2c canned seam).
//
// `spawn_once` serves the fixture from a local port, which is enough only while
// every url a collector asks for is that port. It stopped being enough the moment
// `collect()` learned to follow a cursor: the `gleif-lei` sample's `links.next` was
// an ABSOLUTE `api.gleif.org` url, so this sweep read the local fixture, followed
// the cursor, and crawled the LIVE API from page 2 to page 667 before a 400 stopped
// it — 720 s of wall clock and a rate-limit ban, from a test whose whole contract
// is "no network".
//
// The canned table closes the class, not the instance: any url outside the fixture
// is now a LOUD `Err` naming it, so the next absolute cursor in a fixture fails in
// milliseconds with the url printed instead of going quietly to the wire.
if matches!(plan.feed, Feed::Http) {
cfg = cfg.with_canned(korp_collectors::http::Canned::new(
[(base_url, payload_for_canned)].into_iter().collect(),
));
}
let stats = collector.collect(&cfg, Bounds::default(), &sink).await;
(stats, sink)
}
// ═══════════════════════════════════════════════════════════════════════════
// SWEEP #1 — every registered collector has an offline-safe sample
// ═══════════════════════════════════════════════════════════════════════════
/// The real-or-fake law, swept. `sample::for_collector` must resolve EVERY name
/// in `registry()` to a non-empty body. RED-when-broken: register a collector and
/// forget its sample, and this names it.
#[test]
fn every_registered_collector_has_an_offline_safe_sample() {
// Deterministic: never let the sweep depend on a live endpoint.
unsafe { std::env::set_var("KORP_COLLECTORS_OFFLINE", "1") };
let mut missing: Vec<String> = Vec::new();
let mut empty: Vec<String> = Vec::new();
let mut n = 0usize;
for c in registry() {
let meta = c.meta();
n += 1;
match sample::for_collector(&meta.name) {
None => {
missing.push(meta.name.clone());
emit_for(
&format!("korp-collectors/{}", meta.name),
"offline_safe_sample_declared",
false,
"no sample::for_collector entry",
);
}
Some(s) if s.body.is_empty() => {
empty.push(meta.name.clone());
emit_for(
&format!("korp-collectors/{}", meta.name),
"offline_safe_sample_declared",
false,
"sample body is empty",
);
}
Some(s) => emit_for(
&format!("korp-collectors/{}", meta.name),
"offline_safe_sample_declared",
true,
&format!("[{}] {} bytes", s.origin.label(), s.body.len()),
),
}
}
assert!(n >= 40, "the sweep must see the whole registry, saw {n}");
assert!(
missing.is_empty(),
"{} registered collector(s) have NO offline-safe sample \
(add a `sample::*_try_get_real_sample_data` + a `sample::for_collector` arm): {missing:?}",
missing.len()
);
assert!(
empty.is_empty(),
"collector(s) with an EMPTY sample body: {empty:?}"
);
emit(
"registry_sweep_samples_complete",
true,
&format!("{n} collectors, every one resolves an offline-safe sample"),
);
}
/// `for_collector` answers ONLY for registered names — a typo'd or retired name
/// must not silently resolve to some other collector's sample.
#[test]
fn for_collector_rejects_unregistered_names() {
assert!(sample::for_collector("no-such-collector").is_none());
assert!(sample::for_collector("").is_none());
// ...and the mapping is not accidentally case-insensitive.
assert!(sample::for_collector("DCAT-AP-SE").is_none());
assert!(sample::for_collector("dcat-ap-se").is_some());
}
// ═══════════════════════════════════════════════════════════════════════════
// SWEEP #2 — RAGNARÖK: empty is RED, never silently green
// ═══════════════════════════════════════════════════════════════════════════
/// Feed EVERY registered collector an empty-but-well-formed source and assert the
/// outcome is never a *silent* empty: either the collector errors (RED by any
/// reading), or it returns 0 rows WITH a reason. A collector that returns
/// `Ok(CollectStats::default())` — 0 rows, no reason — fails here by name.
///
/// This is the sweep the crate never had: RAGNARÖK was asserted end-to-end for
/// four collectors out of fifty-eight.
// Full-registry sweep: its absolute thresholds (`explained >= 40`) assume the WHOLE
// ~58-collector set is compiled in. The extra collectors live behind osm/csv/gtfs/
// mainframe/gribnc, so under a partial build (default `parsers` only) the registry is
// smaller and the threshold cannot be met. Gate the sweep to the full-feature config
// (`cargo test --all-features`) where it is meaningful — never weaken the threshold.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
#[tokio::test]
async fn no_registered_collector_is_silently_green_on_an_empty_source() {
unsafe { std::env::set_var("KORP_COLLECTORS_OFFLINE", "1") };
export_fixture_credentials();
let dir = tempfile::tempdir().expect("tempdir");
let mut silent: Vec<String> = Vec::new();
let mut explained = 0usize;
let mut errored = 0usize;
// A collector that DECLARES a secret and finds none in this environment must
// refuse BEFORE the wire, naming the variable. That is the correct answer, not a
// missed 0-row branch — so it is counted apart from a collector that simply blew
// up, and it still satisfies the "reached a defined outcome" floor below.
let mut credential_refusals: Vec<String> = Vec::new();
let mut n = 0usize;
for c in registry() {
let meta = c.meta();
let st = c.source_target();
n += 1;
let plan = plan_for(&meta, &st);
let payload = empty_payload(&meta, &st);
let (stats, sink) = collect_over(c.as_ref(), payload, &plan, dir.path()).await;
let (ok, detail) = match &stats {
Err(e) => {
let msg = e.to_string();
let named = c
.requirements()
.secrets
.iter()
.any(|s| !s.name.is_empty() && msg.contains(&s.name));
if named {
credential_refusals.push(format!("{}: {msg}", meta.name));
(true, format!("refused, naming its own secret: {msg}"))
} else {
errored += 1;
(true, format!("errored on an empty source: {e}"))
}
}
Ok(s) => match s.ragnarok() {
Ragnarok::Red => {
silent.push(meta.name.clone());
(
false,
"0 rows, NO reason — silently green (RAGNARÖK)".to_string(),
)
}
Ragnarok::ExplainedEmpty => {
explained += 1;
assert_eq!(
sink.batches(),
0,
"{}: no batch pushed for an empty source",
meta.name
);
(
true,
format!("0 rows, reason: {}", s.reason.as_deref().unwrap_or("")),
)
}
// A collector that finds rows in an "empty" document is not a
// RAGNARÖK failure — the payload simply was not empty for it.
Ragnarok::Rows => (
true,
format!("{} rows from the empty-shaped payload", s.rows),
),
},
};
emit_for(
&format!("korp-collectors/{}", meta.name),
"ragnarok_empty_is_not_silently_green",
ok,
&detail,
);
}
assert!(n >= 40, "the sweep must see the whole registry, saw {n}");
assert!(
silent.is_empty(),
"{} collector(s) returned 0 rows with NO reason — RAGNARÖK: empty is RED, \
never silently green. Set `CollectStats.reason` in their 0-row branch: {silent:?}",
silent.len()
);
// The breakdown is load-bearing: if a future refactor made every collector
// ERROR on an empty source the sweep would still be green while proving
// nothing about the 0-row branch. Most collectors must reach it.
// Every credential refusal must NAME the variable it wants — that is the whole
// point of `SecretReq`, and a refusal that said only "not configured" would be
// counted here while telling the operator nothing.
for r in &credential_refusals {
assert!(
r.contains("credential env var") || r.contains('_'),
"a credential refusal must name its variable: {r}"
);
}
assert!(
explained + credential_refusals.len() >= 40,
"most collectors must REACH a DEFINED outcome on an empty source — their \
explained 0-row branch, or a refusal naming their own missing secret — not a \
bare error; got {explained} explained / {} credential-refused / {errored} errored",
credential_refusals.len()
);
emit(
"registry_sweep_ragnarok_complete",
true,
&format!(
"{n} collectors: {explained} explained-empty, {} credential-refused, \
{errored} hard-error, 0 silent",
credential_refusals.len()
),
);
}
/// The sweep's own RED-when-broken proof: a collector that returns
/// `Ok(CollectStats::default())` on an empty source IS detected. Without this,
/// a green sweep could mean "the check works" or "the check checks nothing".
#[test]
fn the_ragnarok_sweep_detects_a_silently_green_collector() {
use korp_collectors::CollectStats;
// What a compliant collector returns for an empty source.
let honest = CollectStats {
reason: Some("source served an empty document".into()),
..Default::default()
};
assert_ne!(
honest.ragnarok(),
Ragnarok::Red,
"an explained empty passes the sweep"
);
// What a NON-compliant one returns — the exact shape the sweep must catch.
let silent = CollectStats::default();
assert_eq!(
silent.ragnarok(),
Ragnarok::Red,
"a 0-row, no-reason collect must be caught by the sweep"
);
assert!(silent.is_red());
emit(
"ragnarok_sweep_is_red_when_broken",
true,
"silent 0-row CollectStats classified Red",
);
}
// ═══════════════════════════════════════════════════════════════════════════
// SWEEP #3 — every collector parses its OWN offline-safe sample
// ═══════════════════════════════════════════════════════════════════════════
/// Drive every collector's `collect()` over its own `sample::for_collector`
/// body and assert rows land. This is the "RED when the parse breaks" law swept
/// across the registry: break a field map, a records pointer, an XML element
/// name or a binary decoder and the owning collector is named here.
///
/// The few collectors whose sample is a parser-level fixture rather than their
/// own wire format ([`Drive::ParserFixture`]) are exercised by the dedicated test
/// the arm names; they are still swept by #1 and #2.
// Full-registry sweep: `parsed >= 57` assumes the WHOLE ~58-collector set (the extra
// collectors are behind osm/csv/gtfs/mainframe/gribnc). Gate to the full-feature config
// (`cargo test --all-features`) so a partial build's smaller registry does not RED it;
// the threshold itself is untouched.
#[cfg(all(
feature = "osm",
feature = "csv",
feature = "gtfs",
feature = "mainframe",
feature = "gribnc"
))]
#[tokio::test]
async fn every_registered_collector_parses_its_own_sample() {
unsafe { std::env::set_var("KORP_COLLECTORS_OFFLINE", "1") };
export_fixture_credentials();
let dir = tempfile::tempdir().expect("tempdir");
let mut broken: Vec<String> = Vec::new();
let mut deferred: Vec<String> = Vec::new();
// Collectors that refused for a NAMED missing credential rather than parsing —
// see the `Err` arm below.
let mut credential_gated: Vec<String> = Vec::new();
let mut parsed = 0usize;
let mut n = 0usize;
for c in registry() {
let meta = c.meta();
let st = c.source_target();
n += 1;
let plan = plan_for(&meta, &st);
if let Some(owner) = plan.parse_deferred_to {
deferred.push(format!("{} → {owner}", meta.name));
emit_for(
&format!("korp-collectors/{}", meta.name),
"sample_parses_into_rows",
true,
&format!("parser-level fixture; parse asserted by {owner}"),
);
continue;
}
let Some(s) = sample::for_collector(&meta.name) else {
// Sweep #1 already fails loudly on this; don't double-report.
continue;
};
let origin = s.origin.label();
let (stats, sink) = collect_over(c.as_ref(), s.body, &plan, dir.path()).await;
let (ok, detail) = match &stats {
Err(e) => {
// A collector that DECLARES a secret and has none here refuses before
// it ever looks at the sample. That is the required behaviour (an
// unconfigured credentialled source must never fetch), not a broken
// parse — so it is recorded, not counted as broken. Its parse is
// asserted directly, over the same fixture, in the leaf's own unit
// test (e.g. `bolagsverket_organisationer_parse_into_a_company_registry_table`).
let msg = e.to_string();
let named = c
.requirements()
.secrets
.iter()
.any(|s| !s.name.is_empty() && msg.contains(&s.name));
if named {
credential_gated.push(format!("{}: {msg}", meta.name));
(true, format!("[{origin}] credential-gated: {msg}"))
} else {
broken.push(format!("{}: {e}", meta.name));
(false, format!("[{origin}] collect() errored: {e}"))
}
}
Ok(st) if st.rows == 0 => {
broken.push(format!("{}: 0 rows from its own sample", meta.name));
(
false,
format!(
"[{origin}] 0 rows from its own sample: {}",
st.reason.as_deref().unwrap_or("no reason")
),
)
}
Ok(st) => {
parsed += 1;
assert_eq!(
sink.rows(),
st.rows,
"{}: the sink's tally must match the reported rows",
meta.name
);
(true, format!("[{origin}] {} rows", st.rows))
}
};
emit_for(
&format!("korp-collectors/{}", meta.name),
"sample_parses_into_rows",
ok,
&detail,
);
}
assert!(n >= 40, "the sweep must see the whole registry, saw {n}");
assert!(
broken.is_empty(),
"{} collector(s) failed to parse their OWN offline-safe sample \
(the parse is broken, or the sample no longer matches the wire shape): {broken:#?}",
broken.len()
);
assert!(
parsed + credential_gated.len() >= 57,
"the whole registry bar the deferred parser-fixture collectors parses its own \
sample (or refuses for a named missing credential); got {parsed} parsed / {} \
credential-gated (a drop means a collector lost its end-to-end parse coverage)",
credential_gated.len()
);
// The credential-gated set is small and named on purpose: if it grew, coverage
// would be draining out through this door rather than through `broken`.
assert!(
credential_gated.len() <= 2,
"too many collectors are skipping their own sample for want of a credential: \
{credential_gated:#?}"
);
emit(
"registry_sweep_sample_parse_complete",
true,
&format!(
"{n} collectors: {parsed} parsed their own sample, {} deferred, {} credential-gated",
deferred.len(),
credential_gated.len()
),
);
}