tests/rest_pagination.rs
//! **A cursor-paginated REST source must land EVERY page, not the first one.**
//!
//! Lane EC O4. `gleif-lei` is the join bridge between an orgnr and a LEI — without it
//! FI's insider register and TED's procurement notices do not join to a Swedish
//! company at all — and GLEIF serves **119 764** Swedish LEIs (measured live
//! 2026-09-04) 200 at a time behind a JSON:API `links.next`. A collector that read one
//! page would land 200 of them and report success, which is the worst possible outcome
//! for a join key: every unmatched company would look like a company with no LEI.
//!
//! Driven entirely offline through the `CollectorCfg::canned` seam (lane EC O2c), so
//! this asserts the PAGINATION and never the network.
use std::collections::BTreeMap;
use arrow_array::Array;
use korp_collectors::http::Canned;
use korp_collectors::rest::RestCollector;
use korp_collectors::sink::BatchSink;
use korp_collectors::{Bounds, Collector, CollectorCfg};
/// Three pages of a JSON:API list: `data` records plus `links.next`, the last with no
/// `next`. Keys are the exact urls, so a collector that invented a url gets a loud
/// "no fixture" rather than a quiet empty.
fn three_pages(base: &str) -> Canned {
let page = |n: usize, ids: &[&str], next: Option<String>| {
let data: Vec<serde_json::Value> = ids
.iter()
.map(|lei| {
serde_json::json!({
"attributes": {
"lei": lei,
"entity": {
"legalName": { "name": format!("Bolag {lei} AB") },
"registeredAs": "559586-2557",
"status": "ACTIVE",
"legalAddress": { "city": "STOCKHOLM", "country": "SE" },
},
"registration": { "status": "ISSUED", "lastUpdateDate": format!("2026-09-0{n}T00:00:00Z") },
}
})
})
.collect();
let mut links = serde_json::Map::new();
if let Some(n) = next {
links.insert("next".into(), serde_json::Value::String(n));
}
serde_json::json!({ "data": data, "links": links }).to_string().into_bytes()
};
let p2 = format!("{base}&page=2");
let p3 = format!("{base}&page=3");
let mut m = BTreeMap::new();
m.insert(base.to_string(), page(1, &["AAA1", "AAA2"], Some(p2.clone())));
m.insert(p2.clone(), page(2, &["BBB1", "BBB2"], Some(p3.clone())));
m.insert(p3, page(3, &["CCC1"], None));
Canned::new(m)
}
fn collect(c: &RestCollector, cfg: &CollectorCfg) -> korp_collectors::CollectStats {
let sink = BatchSink::memory();
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
let stats = rt.block_on(c.collect(cfg, Bounds::default(), &sink)).expect("collect");
// The BATCH is the applied output, not the stats struct: a `rows` count that did
// not come from real Arrow rows would pass every assertion below on its own.
// (`landed()` is the WAREHOUSE's number and is 0 for a memory sink by design.)
let pushed: u64 = sink.take_batches().iter().map(|b| b.num_rows() as u64).sum();
assert_eq!(pushed, stats.rows, "the Arrow rows and the stats agree");
stats
}
/// **Every page lands, and the run says how many pages it read.**
#[test]
fn a_paginated_source_lands_every_page_and_reports_how_many() {
let base = "https://api.gleif.invalid/v1/lei-records?filter=SE";
let c = RestCollector::gleif_lei();
let cfg = CollectorCfg::new(base, "gleif_lei").with_canned(three_pages(base));
let stats = collect(&c, &cfg);
assert_eq!(
stats.rows, 5,
"2 + 2 + 1 across three pages — 2 would mean it read page one and stopped, \
which is exactly the failure that makes a join key silently wrong"
);
}
/// **The page cap is a real bound and it is REPORTED.**
///
/// A crawl has to stop somewhere — 119 764 records at 200 a page is ~600 requests, and
/// a source that grew a cycle would otherwise be an infinite loop. But a truncated
/// crawl that reports success is a table an operator believes is complete, so hitting
/// the cap must SAY so even though the run landed rows.
#[test]
fn hitting_the_page_cap_lands_what_it_read_and_says_it_stopped_early() {
let base = "https://api.gleif.invalid/v1/lei-records?filter=SE";
let c = RestCollector::gleif_lei().with_max_pages(2);
let cfg = CollectorCfg::new(base, "gleif_lei").with_canned(three_pages(base));
let stats = collect(&c, &cfg);
assert_eq!(stats.rows, 4, "two pages of two");
let reason = stats.reason.as_deref().unwrap_or("");
assert!(reason.contains("2"), "the reason names the cap it stopped at: {reason}");
assert!(
reason.contains("more") || reason.contains("incomplete"),
"…and says the source HAS more, so nobody reads the table as complete: {reason}"
);
}
/// **A `next` that points at itself terminates.** A source bug must not become an
/// infinite loop inside korp; the run lands what it read and stops.
#[test]
fn a_self_referential_next_link_terminates_instead_of_looping() {
let base = "https://api.gleif.invalid/v1/lei-records?filter=SE";
let body = serde_json::json!({
"data": [{ "attributes": { "lei": "AAA1", "entity": { "registeredAs": "559586-2557" } } }],
"links": { "next": base },
})
.to_string()
.into_bytes();
let mut m = BTreeMap::new();
m.insert(base.to_string(), body);
let cfg = CollectorCfg::new(base, "gleif_lei").with_canned(Canned::new(m));
let stats = collect(&RestCollector::gleif_lei(), &cfg);
assert_eq!(stats.rows, 1, "the one record, once — not once per loop iteration");
}
/// **The orgnr is the JOIN KEY, so it lands normalised.**
///
/// GLEIF serves `entity.registeredAs` as `559586-2557` (measured live 2026-09-04);
/// Bolagsverket's `orgnr` column carries no hyphen. A join key that is right in two
/// spellings joins nothing, and it fails SILENTLY — every company reads as a company
/// with no LEI.
#[test]
fn the_orgnr_lands_without_the_hyphen_gleif_serves_it_with() {
let base = "https://api.gleif.invalid/v1/lei-records?filter=SE";
let c = RestCollector::gleif_lei();
let cfg = CollectorCfg::new(base, "gleif_lei").with_canned(three_pages(base));
let sink = BatchSink::memory();
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
rt.block_on(c.collect(&cfg, Bounds::default(), &sink)).expect("collect");
let batches = sink.take_batches();
let b = batches.first().expect("one batch");
let col = |name: &str| -> Vec<String> {
let a = b.column_by_name(name).unwrap_or_else(|| panic!("no `{name}` column"));
let s = a.as_any().downcast_ref::<arrow_array::StringArray>().expect("utf8");
(0..s.len()).map(|i| s.value(i).to_string()).collect()
};
assert!(col("orgnr").iter().all(|o| o == "5595862557"), "{:?}", col("orgnr"));
assert_eq!(col("registered_as")[0], "559586-2557", "the source's own spelling is KEPT too");
assert_eq!(col("lei")[0], "AAA1");
}
// ─────────────────────────────────────────────────────────────────────────────
// Riksdagen — a cursor that NEVER ends (lane EC O8)
// ─────────────────────────────────────────────────────────────────────────────
/// Riksdagen's `dokumentlista` envelope: `dokumentlista.dokument` records plus the
/// `@nasta_sida` cursor, faithful to the live shape (`@traff_till`, `@sidor`).
fn riksdagen_page(dok_ids: &[&str], next: Option<String>) -> Vec<u8> {
let docs: Vec<serde_json::Value> = dok_ids
.iter()
.map(|d| {
serde_json::json!({
"dok_id": d,
"doktyp": "prop",
"subtyp": "prop",
"titel": format!("Proposition {d}"),
"rm": "2025/26",
"beteckning": "305",
"nummer": "305",
"organ": "Justitiedepartementet",
"datum": "2026-09-01",
"publicerad": "2026-09-01",
"systemdatum": "2026-09-02 08:43:01",
"status": "",
"kalla": "prop-RIM",
"dokument_url_html": format!("//data.riksdagen.se/dokument/{d}.html"),
"dokumentstatus_url_xml": format!("//data.riksdagen.se/dokumentstatus/{d}.xml"),
"summary": "s",
})
})
.collect();
let mut list = serde_json::Map::new();
list.insert("@traffar".into(), "31812".into());
list.insert("@traff_till".into(), "10000".into());
list.insert("dokument".into(), serde_json::Value::Array(docs));
if let Some(n) = next {
// The live cursor really is served as an absolute url one page further on.
list.insert("@nasta_sida".into(), serde_json::Value::String(n));
}
serde_json::json!({ "dokumentlista": list })
.to_string()
.into_bytes()
}
/// **The cursor is followed, and it is `dokumentlista.@nasta_sida` — not `links.next`.**
///
/// **RED:** a preset whose `next_pointer` misses (a `@` dropped, the envelope segment
/// forgotten) lands page one and reports success: 2 rows instead of 5, with nothing red.
/// That is the same silent truncation lane EC O4 found on GLEIF, and here it would be a
/// document register that looks like it holds 200 propositioner.
#[test]
fn the_riksdagen_cursor_is_followed_across_pages() {
let base = "https://data.riksdagen.invalid/dokumentlista/?utformat=json&doktyp=prop";
let p2 = format!("{base}&p=2");
let p3 = format!("{base}&p=3");
let mut m = BTreeMap::new();
m.insert(base.to_string(), riksdagen_page(&["HD03309", "HD03306"], Some(p2.clone())));
m.insert(p2.clone(), riksdagen_page(&["HD03305", "HD03308"], Some(p3.clone())));
m.insert(p3, riksdagen_page(&["HD03301"], None));
let cfg = CollectorCfg::new(base, "riksdagen_dokument").with_canned(Canned::new(m));
let stats = collect(&RestCollector::riksdagen_dokument(), &cfg);
assert_eq!(
stats.rows, 5,
"2 + 2 + 1 across three pages — 2 would mean the `@nasta_sida` pointer is wrong \
and the register silently ends after page one"
);
assert_eq!(stats.reason, None, "a crawl that finished has nothing to warn about");
}
/// **THE ONE THAT MATTERS: Riksdagen's cursor never terminates, so the CAP is the only
/// thing that stops it.**
///
/// Measured live 2026-09-05 with `sz=200&sort=datum&sortorder=desc`: pages 51, 159, 160,
/// 161 and 200 all answer **HTTP 200** with the SAME 200 rows (`G60314 … G50348`,
/// 1981-10-22 … 1982-08-19 — 200/200 overlap between each consecutive pair),
/// `@traff_till` pinned at `10000`, and a fresh `@nasta_sida` pointing one page further
/// every time. GLEIF at least answers HTTP 400 past its 10 000th result; this answers
/// success, for ever.
///
/// The self-referential-`next` guard in `collect()` does NOT help here: every repeat
/// carries a DIFFERENT url (`&p=N`) with IDENTICAL content, so the `seen` set never
/// fires. **Two belts do**, and this asserts both:
///
/// 1. `RIKSDAGEN_MAX_PAGES = 50` stops the crawl at the last addressable result, and the
/// run REPORTS the table incomplete rather than reporting plain success.
/// 2. `dedupe_pointer = /dok_id` (lane EC O6's mechanism, reused) makes the repeated
/// pages land NOTHING: 50 pages of the same two documents become **2 rows**, not 100,
/// and the run says it dropped 98 records it read twice.
///
/// Belt 2 is what turns a corrupted table into a short one that says it is short — and
/// it is the reason this test asserts 2 rather than 100.
///
/// **RED, driven 2026-09-05:** with `max_pages` at the crate default (700) and before
/// the dedupe was wired, this fixture landed **122** rows — 61 pages × 2, every page the
/// fixture has, the same two documents over and over. 122 rather than 1 400 only because
/// the fixture stops at cap+10; against the LIVE endpoint the default would walk all 700
/// and land 1 400 copies of one page, and report success.
#[test]
fn riksdagen_stops_at_its_reachable_page_count_and_says_the_table_is_incomplete() {
// The live pathology, modelled exactly: page N answers the SAME two documents and a
// cursor pointing at page N+1. The urls DIFFER (`&p=N`) while the content does not,
// which is precisely why `collect()`'s `seen`-url guard cannot save this — it only
// catches a `next` that repeats a url it already read.
let base = "https://data.riksdagen.invalid/dokumentlista/?utformat=json&doktyp=prop";
let url_of = |p: usize| format!("{base}&p={p}");
let mut m = BTreeMap::new();
m.insert(base.to_string(), riksdagen_page(&["G60314", "G50348"], Some(url_of(2))));
// Well past the cap, so the ceiling — not the fixture running out — is what stops it.
for p in 2..=(korp_collectors::rest::RIKSDAGEN_MAX_PAGES + 10) {
m.insert(url_of(p), riksdagen_page(&["G60314", "G50348"], Some(url_of(p + 1))));
}
let cfg = CollectorCfg::new(base, "riksdagen_dokument").with_canned(Canned::new(m));
let stats = collect(&RestCollector::riksdagen_dokument(), &cfg);
assert_eq!(
korp_collectors::rest::RIKSDAGEN_MAX_PAGES,
50,
"50 x sz=200 is the 10 000th result, which is the last one Riksdagen can address"
);
assert_eq!(
stats.rows, 2,
"the crawl read 50 pages of the SAME two documents and landed two rows — without \
the `/dok_id` dedupe it would land 100, and without the cap as well, 1 400"
);
let reason = stats.reason.as_deref().unwrap_or("");
assert!(reason.contains("50"), "the reason names the cap it stopped at: {reason}");
assert!(
reason.contains("more") || reason.contains("INCOMPLETE"),
"…and says the source HAS more, so nobody reads 10 000 of 31 812 propositioner \
as the whole register: {reason}"
);
assert!(
reason.contains("98") && reason.contains("TWICE"),
"…and it SAYS the 98 duplicates were dropped, because a silent dedupe hides the \
fact that a source whose paging repeats may equally be SKIPPING records: {reason}"
);
}
/// **The window is the way out of the cap, so it has to be DECLARED and it has to be
/// `from`.**
///
/// `from=YYYY-MM-DD` is a real server-side lower bound on `datum`: measured 2026-09-05,
/// `doktyp=prop&from=2026-01-01` reports 219 hits against 31 812 unbounded. Without it
/// every run re-reads the same first 10 000 for ever and the collector is permanently
/// stuck at 31.4 % of the corpus.
///
/// **RED:** with `freshness` left to derive (or a `param: None` record watermark), this
/// asserts `saves_bytes() == false` — the run would fetch all 50 pages every time and
/// throw away the rows it already had, which is a delta in rows only.
#[test]
fn riksdagen_declares_the_server_side_window_that_gets_it_past_the_cap() {
use korp_collectors::freshness::{Freshness, WindowGranularity};
let f = RestCollector::riksdagen_dokument().freshness();
assert_eq!(f.key(), "query_window");
assert!(
f.saves_bytes(),
"a window with no server-side parameter saves ROWS and not BYTES, and here the \
bytes are the whole problem: 50 pages of 200 on every tick"
);
assert_eq!(
f.default_refresh_secs(),
300,
"5 minutes — the QueryWindow cadence a growing register gets"
);
match f {
Freshness::QueryWindow { param, record_field, granularity, .. } => {
assert_eq!(param.as_deref(), Some("from"), "Riksdagen's own parameter name");
assert_eq!(record_field, "/datum", "the field the mark is taken from");
assert_eq!(
granularity,
WindowGranularity::Day,
"`from` takes YYYY-MM-DD — handing it a full timestamp returns nothing"
);
}
other => panic!("expected a query window, got {other:?}"),
}
}
/// **A field the source never populates must not become a column.**
///
/// Measured on page 1 of `doktyp=prop` (200 records, 2026-09-05): `undertitel`,
/// `relaterat_id` and `tempbeteckning` are **0/200**, while every mapped field is
/// 200/200. An always-NULL column is not evidence — it is a column an operator has to
/// learn to ignore, and it is exactly the shape lane EC O3 found 29 business keys
/// pointing at.
///
/// The business key is asserted against the SCHEMA rather than against a wish, for the
/// same reason.
#[test]
fn the_riksdagen_schema_carries_only_fields_the_source_populates() {
let c = RestCollector::riksdagen_dokument();
let schema = c.schema(&CollectorCfg::new("https://x.invalid/", "t")).expect("schema");
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in ["dok_id", "doktyp", "titel", "rm", "beteckning", "organ", "datum"] {
assert!(cols.contains(&want), "the register needs `{want}`: {cols:?}");
}
for never in ["undertitel", "relaterat_id", "tempbeteckning"] {
assert!(
!cols.contains(&never),
"`{never}` is 0/200 populated on the live endpoint — mapping it would add a \
column that is NULL on every row: {cols:?}"
);
}
let key = c.business_key();
assert_eq!(key.columns, vec!["dok_id".to_string()]);
assert!(
key.missing_from(&schema).is_empty(),
"the business key names a column the collector does not emit — an Upsert on it \
matches nothing and silently degrades to an append"
);
}