tests/domstol_rattspraxis.rs
//! **Domstolsverket rättspraxis — the court register, and the two ways it lies quietly.**
//!
//! Lane EC O6, half A. `rattspraxis.etjanst.domstol.se/api/v1/publiceringar` serves
//! **17 340** publications (measured live 2026-09-05, `x-total-count`) 100 at a time,
//! and two things about it will silently corrupt a warehouse:
//!
//! 1. **There is no cursor.** No `links.next`, no `next`, nothing in the body — only a
//! `page` query parameter the caller counts up itself. The crate's existing crawl
//! follows a cursor, so before [`RestCollector::with_page_walk`] this collector would
//! have landed 100 of 17 340 and reported success.
//! 2. **The default sort is not a total order.** Sorted by `avgorandedatum` (a bare
//! date) a 7-page crawl of the same 610-publication slice returned 610 rows carrying
//! **607 distinct ids** — three repeated across page boundaries and **three never
//! returned at all**. With `sortorder=publiceringstid` the identical crawl returned
//! 610 of 610, including all three the default sort dropped. Both measured
//! 2026-09-05, same slice, same page size, minutes apart.
//!
//! Everything below is driven offline through the `CollectorCfg::canned` seam (lane EC
//! O2c) and asserts the APPLIED OUTPUT — the Arrow batch — never a stats field on its
//! own. The live measurements above are what chose the shapes; they are not re-run here.
use std::collections::BTreeMap;
use arrow_array::{Array, Int64Array, StringArray};
use korp_collectors::http::Canned;
use korp_collectors::rest::{DOMSTOL_PAGE_SIZE, DOMSTOL_RATTSPRAXIS_URL, RestCollector};
use korp_collectors::sink::BatchSink;
use korp_collectors::{Bounds, CollectStats, Collector, CollectorCfg};
const BASE: &str = "https://rattspraxis.invalid/api/v1/publiceringar?sortorder=publiceringstid";
/// One publication, in the source's real shape (a bare array element, `domstol` nested,
/// every list field a list).
fn publication(id: &str, court: &str, malnr: &[&str], tid: &str, text: Option<&str>) -> serde_json::Value {
serde_json::json!({
"id": id,
"gruppKorrelationsnummer": format!("grupp-{id}"),
"domstol": { "domstolKod": court, "domstolNamn": format!("{court} domstol") },
"typ": "PREJUDIKAT",
"publiceringsform": "REFERAT",
"avgorandedatum": &tid[..10],
"publiceringstid": tid,
"malNummerLista": malnr,
"referatNummerLista": [],
"rattsomradeLista": ["Skatt"],
"nyckelordLista": [],
"forarbeteLista": [],
"bilagaLista": [],
"sammanfattning": "sammanfattning",
"innehall": text,
})
}
/// Run the real `collect()` over a canned page table and return `(stats, one batch)`.
///
/// The BATCH is the applied output: a `rows` count that did not come from real Arrow
/// rows would satisfy every count assertion below on its own.
fn collect(c: &RestCollector, cfg: &CollectorCfg) -> (CollectStats, Option<arrow_array::RecordBatch>) {
let sink = BatchSink::memory();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let stats = rt
.block_on(c.collect(cfg, Bounds::default(), &sink))
.expect("collect");
let batches = sink.take_batches();
let pushed: u64 = batches.iter().map(|b| b.num_rows() as u64).sum();
assert_eq!(pushed, stats.rows, "the Arrow rows and the stats agree");
(stats, batches.into_iter().next())
}
fn utf8(b: &arrow_array::RecordBatch, name: &str) -> Vec<Option<String>> {
let a = b
.column_by_name(name)
.unwrap_or_else(|| panic!("no `{name}` column; have {:?}", b.schema().fields().iter().map(|f| f.name().clone()).collect::<Vec<_>>()))
.as_any()
.downcast_ref::<StringArray>()
.unwrap_or_else(|| panic!("`{name}` is not Utf8"));
(0..a.len())
.map(|i| (!a.is_null(i)).then(|| a.value(i).to_string()))
.collect()
}
/// Three pages of the real shape: a bare array, 2 records each, the last one SHORT.
fn three_pages() -> Canned {
let page = |recs: Vec<serde_json::Value>| serde_json::Value::Array(recs).to_string().into_bytes();
let mut m = BTreeMap::new();
m.insert(
BASE.to_string(),
page(vec![
publication("p1", "HFD", &["4566-25"], "2026-09-03T16:24:00", None),
publication("p2", "HFD", &["6143-25"], "2026-09-03T16:13:00", None),
]),
);
m.insert(
format!("{BASE}&page=1"),
page(vec![
publication("p3", "HDO", &["B 1864-25"], "2026-09-02T11:00:00", None),
publication("p4", "MMOD", &["M 14236-25"], "2026-09-01T09:00:00", None),
]),
);
// SHORT — one record where a full page is two. The walk must stop here and never
// ask for `page=3`, which is unmapped and would fall through to the base fixture.
m.insert(
format!("{BASE}&page=2"),
page(vec![publication(
"p5",
"ADO",
&["A 12-25"],
"2026-08-31T08:00:00",
None,
)]),
);
Canned::new(m)
}
/// The shipped preset, narrowed to the page width the fixture serves.
///
/// **The override is why [`the_shipped_preset_declares_the_page_walk_itself`] exists.**
/// This helper's first version simply called `.with_page_walk(…)`, which RE-ADDS the
/// walk whatever the preset declares — so deleting `page_walk` from the preset left
/// every test in this file green over a collector that would have landed 100 of 17 340
/// publications in production. Driving that red arm (LAW 2) is what found it. The
/// override is still right — a 100-record fixture to prove paging would be absurd — but
/// it is no longer the only thing asserting the walk exists.
fn walker() -> RestCollector {
let shipped = RestCollector::domstol_rattspraxis();
let pw = shipped
.page_walk()
.expect("the shipped preset declares a page walk")
.clone();
shipped.with_page_walk(&pw.param, pw.first, 2)
}
/// **The SHIPPED preset walks pages, deduplicates, and agrees with its own url.**
///
/// The guard the helper above cannot be: every other test in this file narrows the page
/// width, and a narrowing builder cannot tell a preset that declares a walk from one
/// that does not.
#[test]
fn the_shipped_preset_declares_the_page_walk_itself() {
let c = RestCollector::domstol_rattspraxis();
let pw = c.page_walk().expect(
"domstol-rattspraxis MUST declare a page-number walk — the API carries no \
cursor of any kind, so a collector without one lands page zero (100 of 17 340 \
publications, measured 2026-09-05) and reports success",
);
assert_eq!(pw.param, "page", "the parameter the OpenAPI names");
assert_eq!(pw.first, 0, "the API's pages are 0-based");
assert_eq!(
pw.page_size, DOMSTOL_PAGE_SIZE,
"the walk's stop condition must be the width the url asks for, or the very \
first full page reads as short and the crawl ends after one request"
);
assert_eq!(
c.dedupe_pointer(),
Some("/id"),
"the pinned sort removed every observed duplicate, but `publiceringstid` is \
minute-granular — two publications in one minute across a page boundary would \
put them straight back"
);
}
/// **Every page lands, and a SHORT page ends the crawl.**
///
/// Two failures in one assertion. Landing 2 is the cursor-crawl bug: the collector read
/// the first response, found no `links.next` anywhere (there is none) and called it a
/// complete register. Landing 5 but erroring is the other one: a walk that stops only on
/// an EMPTY page asks for `page=3`, and a source that answers an over-the-end page with
/// a 500 rather than `[]` turns that extra request into a failed run — every run,
/// forever.
#[test]
fn a_page_number_walk_lands_every_page_and_stops_on_a_short_one() {
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(three_pages());
let (stats, batch) = collect(&walker(), &cfg);
assert_eq!(
stats.rows, 5,
"2 + 2 + 1 across three pages — 2 would mean it read the first response and \
stopped, which is what a cursor crawl does to a source that has no cursor"
);
let b = batch.expect("one batch");
let ids: Vec<Option<String>> = utf8(&b, "id");
assert_eq!(
ids.iter().flatten().cloned().collect::<Vec<_>>(),
vec!["p1", "p2", "p3", "p4", "p5"],
"in page order, every publication once"
);
assert!(
stats.reason.is_none(),
"a clean walk that ended on a short page has nothing to report: {:?}",
stats.reason
);
}
/// **A record read twice across a page boundary lands ONCE — and the run says so.**
///
/// This is the exact shape measured against the live API on 2026-09-05 under its default
/// sort: the last record of a page reappearing as the first record of the next. The
/// dedupe is not the interesting half. The REPORT is: a repeat is evidence that the
/// server's paging is not a stable total order, and the same instability that repeats
/// three records skips three others that nothing on this side can see. A dedupe that
/// quietly returned a tidy count would hide the only signal there is.
#[test]
fn a_boundary_duplicate_lands_once_and_the_run_reports_the_instability() {
let page = |recs: Vec<serde_json::Value>| serde_json::Value::Array(recs).to_string().into_bytes();
let mut m = BTreeMap::new();
m.insert(
BASE.to_string(),
page(vec![
publication("p1", "HFD", &["4566-25"], "2026-09-03T16:24:00", None),
publication("p2", "HFD", &["6143-25"], "2026-09-03T16:13:00", None),
]),
);
// `p2` again, first on the next page — the live defect, reproduced.
m.insert(
format!("{BASE}&page=1"),
page(vec![publication(
"p2",
"HFD",
&["6143-25"],
"2026-09-03T16:13:00",
None,
)]),
);
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(Canned::new(m));
let (stats, batch) = collect(&walker(), &cfg);
assert_eq!(stats.rows, 2, "three records read, two publications landed");
let b = batch.expect("one batch");
assert_eq!(
utf8(&b, "id").iter().flatten().cloned().collect::<Vec<_>>(),
vec!["p1", "p2"],
"the repeat is dropped, not the original"
);
let reason = stats.reason.as_deref().unwrap_or("");
assert!(
reason.contains('1') && reason.to_lowercase().contains("twice"),
"the run names how many records it read twice: {reason:?}"
);
assert!(
reason.to_lowercase().contains("skip"),
"…and says the same instability may have SKIPPED records nothing here can \
see — the half a tidy dedupe count would hide: {reason:?}"
);
}
/// **The identity is the API's uuid, and the brief's key is measurably not unique.**
///
/// The lane brief named `malNummer + domstolKod`. Over the live 610-publication 2026
/// slice that pair yields 605 distinct values for 610 publications (measured
/// 2026-09-05): a court that publishes both a decision and its referat repeats the case
/// number. Keyed on the brief's pair, an Upsert would merge two genuinely different
/// documents into one warehouse row and one of them would be gone.
///
/// Reproduced here as data, not as prose: two publications, same court, same case
/// number, different uuids.
#[test]
fn the_business_key_is_the_uuid_because_case_number_and_court_repeat() {
let c = RestCollector::domstol_rattspraxis();
assert_eq!(
c.business_key().columns,
vec!["id".to_string()],
"one row per publication, keyed on the API's own uuid"
);
let page = |recs: Vec<serde_json::Value>| serde_json::Value::Array(recs).to_string().into_bytes();
let mut m = BTreeMap::new();
m.insert(
BASE.to_string(),
page(vec![
// The decision and its referat: one court, one case number, two documents.
publication("dom-1", "MMOD", &["M 14236-25"], "2026-09-03T16:24:00", None),
publication("ref-1", "MMOD", &["M 14236-25"], "2026-09-03T16:25:00", None),
]),
);
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(Canned::new(m));
let (stats, batch) = collect(&walker(), &cfg);
assert_eq!(stats.rows, 2, "two publications, not one");
let b = batch.expect("one batch");
let ids = utf8(&b, "id");
let pairs: Vec<(Option<String>, Option<String>)> = utf8(&b, "mal_nummer")
.into_iter()
.zip(utf8(&b, "domstol_kod"))
.collect();
assert_eq!(
pairs[0], pairs[1],
"the brief's key is IDENTICAL on both rows — which is why it cannot be the key"
);
assert_ne!(ids[0], ids[1], "the uuid tells them apart");
}
/// **Repeated fields land joined; an empty list lands NULL, not `""`.**
///
/// `malNummerLista` holds one to four case numbers (24 of 610 hold more than one,
/// measured 2026-09-05). Mapped as a plain string it would land the serde spelling
/// `["M 1-25","M 2-25"]` — JSON inside a text column that every downstream query then
/// re-parses. And a publication citing NO statute must be distinguishable from one
/// citing the empty statute.
#[test]
fn repeated_fields_land_joined_and_an_empty_list_lands_null() {
let page = |recs: Vec<serde_json::Value>| serde_json::Value::Array(recs).to_string().into_bytes();
let mut m = BTreeMap::new();
m.insert(
BASE.to_string(),
page(vec![publication(
"p1",
"MMOD",
&["M 14236-25", "M 14237-25"],
"2026-09-03T16:24:00",
None,
)]),
);
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(Canned::new(m));
let (_, batch) = collect(&walker(), &cfg);
let b = batch.expect("one batch");
assert_eq!(
utf8(&b, "mal_nummer")[0].as_deref(),
Some("M 14236-25; M 14237-25"),
"both case numbers, joined — never the JSON array spelling"
);
// `nyckelordLista` is `[]` in this fixture.
assert_eq!(
utf8(&b, "nyckelord")[0], None,
"an empty list is NULL: `no keywords` and `one empty keyword` are not the same \
value, and `\"\"` would make them the same"
);
assert_eq!(
utf8(&b, "bilaga_fillagring_id")[0], None,
"a publication with no attachment has no attachment id"
);
}
/// **The full text lands as a character COUNT, and the text itself never lands.**
///
/// 258 of 610 publications carry `innehall`, median 18 944 characters — and ONE single
/// judgment carries 24 552 235, which is 27.5 MB of that slice's 34.8 MB in one cell.
/// A first full crawl would put roughly a gigabyte in one `RecordBatch`.
///
/// A record with no `innehall` lands NULL, not `0`: "we do not know whether there is
/// text" and "there are zero characters of text" are different facts, and a warehouse
/// that reported the second for the first would be wrong on 352 of 610 rows.
#[test]
fn the_full_text_lands_as_a_size_and_never_as_text() {
let page = |recs: Vec<serde_json::Value>| serde_json::Value::Array(recs).to_string().into_bytes();
let mut m = BTreeMap::new();
m.insert(
BASE.to_string(),
page(vec![
// Deliberately NON-ASCII: `chars().count()`, not `len()`. A judgment in
// Swedish is full of å ä ö, and a byte count would over-report every one.
publication("p1", "HFD", &["1-25"], "2026-09-03T16:24:00", Some("Förhandsavgörande")),
publication("p2", "HFD", &["2-25"], "2026-09-03T16:13:00", None),
]),
);
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(Canned::new(m));
let (_, batch) = collect(&walker(), &cfg);
let b = batch.expect("one batch");
assert!(
b.column_by_name("innehall").is_none(),
"the full text is NOT a column of this table — see the preset's doc comment"
);
let n = b
.column_by_name("innehall_tecken")
.expect("innehall_tecken column")
.as_any()
.downcast_ref::<Int64Array>()
.expect("innehall_tecken is Int64");
assert_eq!(
n.value(0),
"Förhandsavgörande".chars().count() as i64,
"CHARACTERS, not bytes — `Förhandsavgörande` is 17 chars and 19 bytes"
);
assert!(
n.is_null(1),
"a publication with no `innehall` lands NULL, never 0 — absent text and empty \
text are different facts"
);
}
/// **The declared url carries the sort and the page size, because both are measured.**
///
/// Not decoration: the sort is the difference between 610 and 607 distinct ids over the
/// same 610 records, and `pagesize` is clamped to 100 by the server (`pagesize=500` and
/// `pagesize=1000` returned byte-identical 100-record responses on 2026-09-05), so the
/// walk's stop condition and the url must agree or the very first page reads as short.
#[test]
fn the_declared_url_pins_the_sort_and_agrees_with_the_page_width() {
assert!(
DOMSTOL_RATTSPRAXIS_URL.contains("sortorder=publiceringstid"),
"the default sort (`avgorandedatum`, a bare date) silently drops ~0.5 % of every \
crawl — the sort is pinned in the url: {DOMSTOL_RATTSPRAXIS_URL}"
);
assert!(
DOMSTOL_RATTSPRAXIS_URL.contains(&format!("pagesize={DOMSTOL_PAGE_SIZE}")),
"the url must ask for exactly the page width the walk stops on, or the first \
full page reads as short: {DOMSTOL_RATTSPRAXIS_URL}"
);
let st = RestCollector::domstol_rattspraxis().source_target();
assert_eq!(
st.source_endpoint, DOMSTOL_RATTSPRAXIS_URL,
"one writer for the url — the descriptor and the constant cannot drift"
);
assert_eq!(st.target_table, "domstol_rattspraxis");
}
/// **The freshness declaration names the server's OWN lower bound.**
///
/// `publicerad_fran_och_med` is a real parameter (measured 2026-09-05:
/// `publicerad_fran_och_med=2026-09-01` moved `x-total-count` from 17 340 to **23**), so
/// a re-run is cheaper in BYTES and not only in rows — which for this source is the
/// whole game, since the bytes are the untransferable `innehall`.
///
/// It is DAY-granular, which matters: asking a `yyyy-MM-dd` parameter for a full
/// timestamp returns nothing at all.
#[test]
fn freshness_declares_the_servers_own_day_granular_lower_bound() {
use korp_collectors::freshness::{Freshness, WindowGranularity};
let f = RestCollector::domstol_rattspraxis().freshness();
match &f {
Freshness::QueryWindow {
param,
record_field,
granularity,
..
} => {
assert_eq!(
param.as_deref(),
Some("publicerad_fran_och_med"),
"the server's own parameter, by its real name"
);
assert_eq!(record_field, "/publiceringstid");
assert_eq!(
*granularity,
WindowGranularity::Day,
"the parameter takes `yyyy-MM-dd`; a full timestamp returns nothing"
);
}
other => panic!("expected a server-narrowed query window, got {other:?}"),
}
assert!(
f.saves_bytes(),
"a re-run really does avoid transferring what it already has — for this source \
that is the point, because the bytes are the full judgment text"
);
}
/// **The graph is the two things the source itself names, and nothing it does not.**
///
/// A court, and the case group that ties a decision to its referat (607 distinct groups
/// over 610 publications, 2026-09-05). NOT rättsområde or nyckelord: those are
/// multi-valued and land as one joined cell, and an edge from a joined cell would MERGE
/// a legal-area node named `Miljömål; Skadestånd` that no court ever used.
#[test]
fn the_graph_draws_the_court_and_the_case_group_and_not_the_multi_valued_columns() {
let g = RestCollector::domstol_rattspraxis()
.graph_mapping()
.expect("this table honestly makes a graph");
assert_eq!(g.table, "domstol_rattspraxis");
assert_eq!(g.rels(), vec!["DECIDED_BY", "IN_CASE_GROUP"]);
assert!(
g.labels().contains(&"Court".to_string()),
"the court is a node: {:?}",
g.labels()
);
for forbidden in ["rattsomrade", "nyckelord", "lagrum"] {
assert!(
!g.edges.iter().any(|e| e.src.column == forbidden || e.dst.column == forbidden),
"`{forbidden}` is multi-valued and joined into one cell — an edge from it \
would merge a node no court ever used"
);
}
assert!(
g.limitation.contains("orgnr") && g.limitation.to_lowercase().contains("free text"),
"the limitation says out loud that a company is named only in free text, so \
nothing here joins the orgnr spine: {:?}",
g.limitation
);
assert!(!g.uses_spine(), "a court register has no kommunkod anywhere");
}
/// **Hitting the page ceiling lands what it read and SAYS the table is incomplete.**
///
/// The register is 174 pages today. A ceiling has to exist — a source that grew a cycle
/// must not become an infinite crawl — but a truncated table reporting plain success is
/// a table an operator believes is the whole of Swedish case law.
#[test]
fn hitting_the_page_ceiling_says_the_court_register_is_incomplete() {
let cfg = CollectorCfg::new(BASE, "domstol_rattspraxis").with_canned(three_pages());
let (stats, _) = collect(&walker().with_max_pages(2), &cfg);
assert_eq!(stats.rows, 4, "two full pages of two");
let reason = stats.reason.as_deref().unwrap_or("");
assert!(reason.contains('2'), "the reason names the ceiling: {reason}");
assert!(
reason.contains("INCOMPLETE") || reason.contains("more"),
"…and says the source HAS more: {reason}"
);
}