tests/collectors_freshness_live.rs
//! **The live proof** — `polisen-events` against the real `https://polisen.se/api/events`,
//! run twice, with the rows, the bytes and the mark measured both times.
//!
//! `#[ignore]` because CI has no network and this crate's law is that tests never hang
//! for want of one. Run it deliberately:
//!
//! ```text
//! cargo test --test collectors_freshness_live --features daemon -- --ignored --nocapture
//! ```
//!
//! It is not a fixture dressed as a live test: it fails if the endpoint is unreachable
//! rather than falling back to a synthetic sample, because the whole point is to
//! measure what a real server does with a real conditional request.
use std::sync::Arc;
use korp_collectors::daemon::{MemWarehouse, WarehouseSink};
use korp_collectors::freshness::{FileMarkStore, MarkStore, Plan};
use korp_collectors::history::{self, BusinessKey};
use korp_collectors::rest::RestCollector;
use korp_collectors::{Collector, CollectorCfg, Ragnarok, run_incremental};
/// **polisen-events, twice.** The second run must land only new rows (or an explained
/// zero) instead of re-landing the whole feed, and must transfer fewer bytes.
#[tokio::test]
#[ignore = "live network: the real polisen open-data endpoint"]
async fn polisen_run_twice_lands_a_delta_not_the_whole_feed() {
let dir = std::env::temp_dir().join(format!("korp-fr-live-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let marks = FileMarkStore::open(dir.join("marks.json"));
let wh = Arc::new(MemWarehouse::new());
let c = RestCollector::polisen();
let cfg = CollectorCfg::new("https://polisen.se/api/events", "polisen_events");
println!("freshness: {}", c.freshness().label());
println!("business key: {}", c.business_key().label());
// ── run 1 ────────────────────────────────────────────────────────────────
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks)
.await
.expect("run 1 — the live endpoint must answer");
let m1 = marks
.mark("polisen-events", &cfg.base_url)
.expect("run 1 wrote a mark");
println!("\nRUN 1 {}", r1.line());
println!(" rows landed : {}", r1.stats.rows);
println!(" bytes : {}", m1.bytes);
println!(" mark : {}", m1.mark);
println!(" table rows : {}", wh.rows("polisen_events"));
assert!(matches!(r1.plan, Plan::Window { param_value: None, .. }), "{:?}", r1.plan);
assert!(r1.stats.rows > 0, "the live feed carried events");
assert!(m1.bytes > 10_000, "run 1 fetched the whole feed: {} bytes", m1.bytes);
assert!(!m1.mark.is_empty(), "run 1 recorded a watermark");
// ── run 2, immediately ───────────────────────────────────────────────────
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks)
.await
.expect("run 2");
let m2 = marks.mark("polisen-events", &cfg.base_url).expect("a mark");
println!("\nRUN 2 {}", r2.line());
println!(" rows landed : {}", r2.stats.rows);
println!(" bytes : {}", m2.bytes);
println!(" mark : {}", m2.mark);
println!(" table rows : {}", wh.rows("polisen_events"));
// The plan carried the window, at the parameter's DAY granularity.
match &r2.plan {
Plan::Window {
param: Some(p),
param_value: Some(v),
since: Some(s),
..
} => {
assert_eq!(p, "DateTime");
assert_eq!(v.len(), 10, "a DAY, not a timestamp: {v}");
assert!(s.starts_with(v), "the day is the mark's own: {s} / {v}");
}
other => panic!("run 2 must carry the window: {other:?}"),
}
// **The measurement that matters.** Fewer bytes, and only new rows.
assert!(
m2.bytes < m1.bytes,
"run 2 must fetch LESS than the whole feed: {} vs {}",
m2.bytes,
m1.bytes
);
assert!(
r2.stats.rows < r1.stats.rows,
"run 2 must not re-land the feed: {} vs {} rows",
r2.stats.rows,
r1.stats.rows
);
// Minutes apart, polisen almost always has nothing new — an explained zero, green.
if r2.stats.rows == 0 {
assert_eq!(r2.stats.ragnarok(), Ragnarok::ExplainedEmpty, "unchanged is not RED");
assert!(r2.stats.reason.as_deref().unwrap().contains("unchanged"), "{:?}", r2.stats.reason);
}
let total = wh.rows("polisen_events");
assert!(
total < r1.stats.rows * 2,
"the table must not have doubled: {total} after {} + {}",
r1.stats.rows,
r2.stats.rows
);
println!(
"\nDELTA: {} → {} rows, {} → {} bytes ({}× less)",
r1.stats.rows,
r2.stats.rows,
m1.bytes,
m2.bytes,
if m2.bytes > 0 { m1.bytes / m2.bytes } else { 0 }
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **A conditional fetch, proven live** — USGS serves `Last-Modified` and answers a
/// real `304` to `If-Modified-Since`, so the second run transfers **zero** body bytes.
#[tokio::test]
#[ignore = "live network: the real USGS earthquake feed"]
async fn usgs_second_run_transfers_zero_bytes_on_a_304() {
let dir = std::env::temp_dir().join(format!("korp-fr-usgs-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let marks = FileMarkStore::open(dir.join("marks.json"));
let wh = Arc::new(MemWarehouse::new());
let c = korp_collectors::usgs::UsgsQuakeCollector::default();
let cfg = CollectorCfg::new(
"https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson",
"usgs_quakes",
);
println!("freshness: {}", c.freshness().label());
let r1 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 1");
let m1 = marks.mark("usgs-earthquakes", &cfg.base_url).expect("a mark");
println!("RUN 1 {}\n bytes {} last-modified {:?}", r1.line(), m1.bytes, m1.last_modified);
assert!(m1.bytes > 0, "run 1 fetched a body");
assert!(
m1.last_modified.is_some(),
"USGS must serve Last-Modified for the conditional path to exist"
);
let r2 = run_incremental(&c, &cfg, wh.clone(), &marks).await.expect("run 2");
let m2 = marks.mark("usgs-earthquakes", &cfg.base_url).expect("a mark");
println!("RUN 2 {}\n bytes {} unchanged {}", r2.line(), m2.bytes, m2.unchanged);
assert!(matches!(r2.plan, Plan::Conditional { if_modified_since: Some(_), .. }));
// Seconds apart, the hourly summary cannot have changed.
assert_eq!(m2.bytes, 0, "a 304 transfers no body at all");
assert!(m2.unchanged, "and it is recorded as unchanged");
assert_eq!(r2.stats.rows, 0);
assert_eq!(r2.stats.ragnarok(), Ragnarok::ExplainedEmpty, "unchanged is GREEN");
assert_eq!(
wh.rows("usgs_quakes"),
r1.stats.rows,
"nothing was re-landed"
);
println!("\nDELTA: {} → 0 bytes on the second run", m1.bytes);
let _ = std::fs::remove_dir_all(&dir);
}
/// **Historization, proven on live polisen rows.** Land the real feed, re-read it with
/// one row's flag changed, and the CURRENT view holds one row per key — with the old
/// value still readable.
#[tokio::test]
#[ignore = "live network: the real polisen open-data endpoint"]
async fn polisen_rows_historize_to_one_row_per_key_with_the_old_value_reachable() {
let http = korp_collectors::http::HttpCore::default();
let body = http
.get_text("https://polisen.se/api/events", &korp_collectors::http::NoAuth)
.await
.expect("the live endpoint must answer");
let json: serde_json::Value = serde_json::from_str(&body).expect("json");
let c = RestCollector::polisen();
let key = c.business_key();
assert_eq!(key.columns, vec!["id"], "polisen is keyed on its event id");
let first = korp_collectors::rest::parse_records(&json, "", c.fields()).expect("parse");
let n = first.num_rows();
println!("live polisen rows: {n}");
assert!(n > 100, "the live feed is substantial: {n}");
let v1 = history::historize(&first, &key, "2026-09-03T10:00:00Z", "run-1").expect("historize");
let digest = history::digest_of(&v1);
assert_eq!(digest.len(), n, "every row has a distinct id — the key really identifies");
// The SAME feed again: a re-read with nothing changed appends NOTHING.
let v2 = history::historize(&first, &key, "2026-09-03T11:00:00Z", "run-2").expect("historize");
let (kept, acc) = history::suppress_unchanged(&v2, &key, &digest).expect("suppress");
println!("re-read unchanged: {}", acc.line());
assert_eq!(kept.num_rows(), 0, "an unchanged re-read appends nothing");
assert_eq!(acc.unchanged as usize, n);
// Now the case that needs history: the same key comes back CHANGED.
let mut changed = json.clone();
let target_id = changed[0]["id"].clone();
changed[0]["type"] = serde_json::Value::String("REVIDERAD".into());
let re = korp_collectors::rest::parse_records(&changed, "", c.fields()).expect("parse");
let v3 = history::historize(&re, &key, "2026-09-03T12:00:00Z", "run-3").expect("historize");
let (kept, acc) = history::suppress_unchanged(&v3, &key, &digest).expect("suppress");
println!("re-read with one changed row: {}", acc.line());
assert_eq!(kept.num_rows(), 1, "only the changed row is appended");
assert_eq!((acc.changed, acc.new, acc.unchanged as usize), (1, 0, n - 1));
// The physical table = v1 + the one appended version.
let table = arrow_select::concat::concat_batches(&v1.schema(), [&v1, &kept]).expect("concat");
assert_eq!(table.num_rows(), n + 1, "the table grew by ONE row, not by {n}");
// The CURRENT view is one row per key — exactly what an upsert would leave.
let now = history::latest_per_key(&table).expect("latest");
assert_eq!(now.num_rows(), n, "one row per key, not two for the revised event");
println!("physical rows {} → current view {}", table.num_rows(), now.num_rows());
// …and the OLD value of the revised row is still reachable through the history axis.
use arrow_array::Array;
let ids = now
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<arrow_array::Int64Array>()
.unwrap();
let types = now
.column_by_name("type")
.unwrap()
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap();
let want = target_id.as_i64().expect("an id");
let row = (0..now.num_rows()).find(|r| ids.value(*r) == want).expect("the revised row");
assert_eq!(types.value(row), "REVIDERAD", "the CURRENT value is the revision");
let old_types = table
.column_by_name("type")
.unwrap()
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap();
let stamps = table
.column_by_name(history::VALID_FROM_COL)
.unwrap()
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.unwrap();
let old_ids = table
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<arrow_array::Int64Array>()
.unwrap();
let previous: Vec<&str> = (0..table.num_rows())
.filter(|r| old_ids.value(*r) == want && stamps.value(*r) == "2026-09-03T10:00:00Z")
.map(|r| old_types.value(r))
.collect();
assert_eq!(previous.len(), 1, "the previous version is there exactly once");
assert_ne!(previous[0], "REVIDERAD", "and it holds the value from before the revision");
println!("key {want}: was `{}`, is now `REVIDERAD` — both readable", previous[0]);
// The declared SQL is the one a warehouse runs for the same answer.
println!("\ncurrent state SQL:\n{}", history::latest_per_key_sql("polisen_events"));
}
/// A guard on the guard: [`BusinessKey`] over polisen's DECLARED join keys would be the
/// wrong identity, and the override says so. Offline — this is about the declaration.
#[test]
fn polisens_join_keys_are_a_place_and_its_business_key_is_not() {
let c = RestCollector::polisen();
let declared = c.business_key();
assert_eq!(declared.columns, vec!["id"]);
let derived = BusinessKey::derived(&c.meta().join_keys);
assert_eq!(derived.columns, vec!["location_name", "gps"]);
assert_ne!(
declared.columns, derived.columns,
"the override exists because the join keys identify a PLACE, not an event"
);
assert!(declared.why.contains("place"), "{}", declared.why);
}