tests/collectors_sample.rs
//! **Real-or-fake sample contracts** — for EACH collector, drive its
//! `sample::*_try_get_real_sample_data()` (a short-timeout live fetch that FALLS
//! BACK to an embedded synthetic fixture), then assert the collector parses /
//! normalises the sample into its **target shape**. These never hang and never
//! fail purely for want of a network — set `KORP_COLLECTORS_OFFLINE=1` to force
//! the synthetic path. Each check emits a functional-status row under
//! `--features testmatrix`, tagged with whether it ran on REAL or SYNTHETIC data.
//!
//! Gated on `parsers` — NOT on `osm`, even though one check here drives the OSM
//! geometry collector: that check carries its own `#[cfg(feature = "osm")]`, so a
//! file-level `osm` gate would silently drop the other 17 tests from the default
//! (`parsers`-only) run. Without any gate the binary fails to COMPILE under
//! `--no-default-features` rather than being skipped as not-applicable.
#![cfg(feature = "parsers")]
use arrow_array::{Array, Float64Array, Int64Array, StringArray};
use korp_collectors::sample;
use korp_collectors::{
SourceEndpoint, TargetKind, emit_registry_descriptors, functional_status, registry_descriptors,
};
/// Emit a per-collector row that records BOTH the verdict and the sample origin.
fn emit_sample(collector: &str, ok: bool, origin: &sample::SampleOrigin, detail: &str) {
functional_status(
&format!("korp-collectors/{collector}"),
"real_or_fake_sample_parses",
ok,
&format!("[{}] {detail}", origin.label()),
);
}
// ── #1 DCAT-AP-SE → registry rows ───────────────────────────────────────────
#[test]
fn dcat_sample_parses_into_registry() {
let s = sample::dcat_try_get_real_sample_data();
let rows = korp_collectors::dcat::parse_dcat_rdfxml(&s.body).expect("parse DCAT sample");
assert!(!rows.is_empty(), "DCAT sample must yield ≥1 registry row");
let batch = korp_collectors::dcat::registry_batch(&rows).expect("registry batch");
assert_eq!(batch.num_rows(), rows.len());
emit_sample(
"dcat",
true,
&s.origin,
&format!("{} registry rows", rows.len()),
);
}
// ── #2 PxWeb / JSON-stat2 → Arrow cube ──────────────────────────────────────
#[test]
fn pxweb_sample_normalises_to_cube() {
let s = sample::pxweb_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("sample is JSON");
let batch = korp_collectors::pxweb::parse_jsonstat2(&v).expect("normalise cube");
assert!(batch.num_rows() > 0, "cube must have ≥1 cell");
// target shape: one code column per dim + a trailing `value` column
let schema = batch.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names.last(),
Some(&"value"),
"last column is the value: {names:?}"
);
emit_sample(
"pxweb",
true,
&s.origin,
&format!("{} cells", batch.num_rows()),
);
}
// ── Eurostat SDMX-JSON 2.1 → observation value rows ─────────────────────────
#[test]
fn eurostat_sdmx_sample_flattens_to_observations() {
let s = sample::eurostat_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("sample is JSON");
let batch = korp_collectors::sdmx::parse_sdmx_json(&v).expect("flatten SDMX");
assert!(
batch.num_rows() > 0,
"SDMX message must yield ≥1 observation"
);
// target shape: one code column per dim + a trailing `value` column
let schema = batch.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names.last(),
Some(&"value"),
"last column is the value: {names:?}"
);
// the `value` column is Float64
assert!(
batch
.column(batch.num_columns() - 1)
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"eurostat-sdmx",
true,
&s.origin,
&format!("{} observations", batch.num_rows()),
);
}
// ── #4 ENTSO-E → price points ───────────────────────────────────────────────
#[test]
fn entsoe_sample_maps_to_points() {
let s = sample::entsoe_try_get_real_sample_data();
let points = korp_collectors::entsoe::parse_entsoe_xml(&s.as_str()).expect("parse ENTSO-E");
assert!(!points.is_empty(), "ENTSO-E sample must yield ≥1 point");
let batch = korp_collectors::entsoe::points_batch(&points).expect("points batch");
assert_eq!(batch.num_rows(), points.len());
emit_sample(
"entsoe",
true,
&s.origin,
&format!("{} points", points.len()),
);
}
// ── #4c ENTSO-E day-ahead PRICES → price rows (energy market) ────────────────
#[test]
fn entsoe_day_ahead_sample_maps_to_price_rows() {
let s = sample::entsoe_day_ahead_try_get_real_sample_data();
let prices = korp_collectors::entsoe_day_ahead::parse_entsoe_day_ahead_xml(&s.as_str())
.expect("parse ENTSO-E A44");
assert!(
!prices.is_empty(),
"ENTSO-E A44 sample must yield ≥1 day-ahead price"
);
let batch = korp_collectors::entsoe_day_ahead::prices_batch(&prices).expect("prices batch");
assert_eq!(batch.num_rows(), prices.len());
// target shape: prices keyed by zone + position (hour), each carrying a price.
let schema = batch.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in ["bidding_zone", "position", "price"] {
assert!(
names.contains(&want),
"price table carries `{want}`: {names:?}"
);
}
emit_sample(
"entsoe-day-ahead",
true,
&s.origin,
&format!("{} day-ahead prices", prices.len()),
);
}
// ── OpenSky flight TRACK → ordered trajectory (flight recorder) ──────────────
#[test]
fn opensky_tracks_sample_maps_to_ordered_trajectory() {
let c = korp_collectors::rest::RestCollector::opensky_tracks();
let s = sample::opensky_tracks_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch =
korp_collectors::rest::parse_records(&v, c.records_pointer(), c.fields()).expect("map");
assert!(
batch.num_rows() > 0,
"opensky track sample must yield ≥1 waypoint"
);
// target shape: an ordered (time, lat, lon, …) trajectory — time non-decreasing.
let time = batch
.column_by_name("time")
.and_then(|a| a.as_any().downcast_ref::<Int64Array>())
.expect("time is Int64");
let times: Vec<i64> = (0..time.len()).map(|i| time.value(i)).collect();
assert!(
times.windows(2).all(|w| w[0] <= w[1]),
"waypoints ordered by time: {times:?}"
);
emit_sample(
"opensky-tracks",
true,
&s.origin,
&format!("{} track waypoints", batch.num_rows()),
);
}
// ── #5 Generic REST — Polisen / Kolada / SMHI ───────────────────────────────
#[test]
fn polisen_sample_maps_to_arrow() {
let c = korp_collectors::rest::RestCollector::polisen();
let s = sample::polisen_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch =
korp_collectors::rest::parse_records(&v, c.records_pointer(), c.fields()).expect("map");
assert!(batch.num_rows() > 0, "polisen sample must yield ≥1 event");
// target shape: the `id` column is Int64
assert!(
batch
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.is_some()
);
emit_sample(
"rest-polisen",
true,
&s.origin,
&format!("{} events", batch.num_rows()),
);
}
#[test]
fn kolada_sample_maps_to_arrow() {
let c = korp_collectors::rest::RestCollector::kolada();
let s = sample::kolada_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch =
korp_collectors::rest::parse_records(&v, c.records_pointer(), c.fields()).expect("map");
assert!(batch.num_rows() > 0, "kolada sample must yield ≥1 KPI");
let id = batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.expect("kpi id is Utf8");
assert!(!id.value(0).is_empty());
emit_sample(
"rest-kolada",
true,
&s.origin,
&format!("{} KPIs", batch.num_rows()),
);
}
#[test]
fn smhi_sample_maps_to_arrow() {
let c = korp_collectors::rest::RestCollector::smhi_metobs();
let s = sample::smhi_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch =
korp_collectors::rest::parse_records(&v, c.records_pointer(), c.fields()).expect("map");
assert!(
batch.num_rows() > 0,
"smhi sample must yield ≥1 observation"
);
// the `value` column is Float64
assert!(
batch
.column(1)
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"rest-smhi",
true,
&s.origin,
&format!("{} obs", batch.num_rows()),
);
}
// ── AIS → position table (map overlay) ──────────────────────────────────────
#[test]
fn ais_sample_maps_to_position_table() {
let c = korp_collectors::rest::RestCollector::digitraffic_ais();
let s = sample::ais_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch =
korp_collectors::rest::parse_records(&v, c.records_pointer(), c.fields()).expect("map");
assert!(batch.num_rows() > 0, "ais sample must yield ≥1 position");
let mmsi = batch
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.expect("mmsi Int64");
assert!(mmsi.value(0) > 0);
emit_sample(
"rest-ais",
true,
&s.origin,
&format!("{} positions", batch.num_rows()),
);
}
// ── USGS earthquakes (GeoJSON) → quake position table (map overlay) ─────────
#[test]
fn usgs_sample_flattens_to_quake_table() {
let s = sample::usgs_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("sample is JSON");
let quakes = korp_collectors::usgs::parse_usgs_geojson(&v).expect("flatten USGS");
assert!(
!quakes.is_empty(),
"USGS sample must yield ≥1 locatable quake"
);
let batch = korp_collectors::usgs::quakes_batch(&quakes).expect("quakes batch");
assert_eq!(batch.num_rows(), quakes.len());
// target shape: the `longitude` column is Float64 (a map X coordinate).
assert!(
batch
.column_by_name("longitude")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"usgs-earthquakes",
true,
&s.origin,
&format!("{} quakes", quakes.len()),
);
}
// ── OpenAQ air quality → measurement position table (map overlay) ───────────
#[test]
fn openaq_sample_flattens_to_measurement_table() {
let s = sample::openaq_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("sample is JSON");
let rows = korp_collectors::openaq::parse_openaq_json(&v).expect("flatten OpenAQ");
assert!(
!rows.is_empty(),
"OpenAQ sample must yield ≥1 located measurement"
);
let batch = korp_collectors::openaq::measurements_batch(&rows).expect("measurements batch");
assert_eq!(batch.num_rows(), rows.len());
// target shape: the `longitude` column is Float64 (a map X coordinate).
assert!(
batch
.column_by_name("longitude")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"openaq-air-quality",
true,
&s.origin,
&format!("{} measurements", rows.len()),
);
}
// ── Energy-Charts → generation-mix power table ──────────────────────────────
#[test]
fn energy_charts_sample_transposes_to_power_table() {
let s = sample::energy_charts_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("sample is JSON");
let readings = korp_collectors::energy_charts::parse_energy_charts_json(&v).expect("transpose");
assert!(
!readings.is_empty(),
"Energy-Charts sample must yield ≥1 (time, type) power cell"
);
let batch = korp_collectors::energy_charts::power_batch(&readings).expect("power batch");
assert_eq!(batch.num_rows(), readings.len());
// target shape: the `power_mw` column is Float64.
assert!(
batch
.column_by_name("power_mw")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"energy-charts-power",
true,
&s.origin,
&format!("{} power cells", readings.len()),
);
}
// ── GTFS-Realtime → vehicle position table (map overlay) ────────────────────
#[test]
fn gtfs_rt_sample_decodes_to_vehicle_position_table() {
let s = sample::gtfs_rt_try_get_real_sample_data();
let rows = korp_collectors::gtfs_rt::parse_gtfs_rt(&s.body).expect("decode GTFS-RT");
assert!(
!rows.is_empty(),
"GTFS-RT sample must yield ≥1 locatable vehicle"
);
let batch = korp_collectors::gtfs_rt::vehicles_batch(&rows).expect("vehicles batch");
assert_eq!(batch.num_rows(), rows.len());
// target shape: the `longitude` column is Float64 (a map X coordinate).
assert!(
batch
.column_by_name("longitude")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.is_some()
);
emit_sample(
"gtfs-rt-vehicles",
true,
&s.origin,
&format!("{} vehicles", rows.len()),
);
}
// ── Trafikverket → train position table ─────────────────────────────────────
#[test]
fn trafikverket_sample_maps_to_position_table() {
let c = korp_collectors::trafikverket::TrafikverketCollector::train_positions();
let s = sample::trafikverket_try_get_real_sample_data();
let v: serde_json::Value = serde_json::from_slice(&s.body).expect("JSON");
let batch = korp_collectors::rest::parse_records(&v, &c.records_pointer(), c.fields())
.expect("map trainpos");
assert!(
batch.num_rows() > 0,
"trafikverket sample must yield ≥1 position"
);
emit_sample(
"trafikverket",
true,
&s.origin,
&format!("{} positions", batch.num_rows()),
);
}
// ── OSM admin boundaries → admin_units + trie ───────────────────────────────
#[test]
fn osm_admin_sample_extracts_units() {
use korp_collectors::osm_admin::{WorldAdminTrie, admin_batch, extract_admin_units};
let s = sample::osm_admin_try_get_real_sample_data();
let features = sample::parse_admin_features(&s.body);
assert!(!features.is_empty(), "admin sample must carry features");
let units = extract_admin_units(features);
assert!(
!units.is_empty(),
"≥1 boundary after the road is filtered out"
);
let batch = admin_batch(&units).expect("admin batch");
assert_eq!(batch.num_rows(), units.len());
let trie = WorldAdminTrie::build(&units);
assert!(trie.countries() >= 1);
emit_sample(
"osm_admin",
true,
&s.origin,
&format!(
"{} admin units, {} countries",
units.len(),
trie.countries()
),
);
}
// ── OSM ways (basemap) — feature `osm` ──────────────────────────────────────
#[cfg(feature = "osm")]
#[test]
fn osm_ways_sample_maps_to_wkb_arrow() {
let s = sample::osm_ways_try_get_real_sample_data();
// The sample body is raw PBF bytes — osm-katana reads a path.
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("sample.osm.pbf");
std::fs::write(&path, &s.body).unwrap();
let ways = korp_collectors::osm::load_ways_pbf(path.to_str().unwrap()).expect("parse ways");
let with_geom = ways.iter().filter(|w| w.geometry.is_some()).count();
assert!(with_geom >= 1, "≥1 resolvable way");
let batch = korp_collectors::osm::ways_to_batch(&ways).expect("ways → arrow");
assert_eq!(batch.num_rows(), ways.len());
emit_sample(
"osm",
true,
&s.origin,
&format!("{with_geom} ways with geometry"),
);
}
// ── descriptor discovery: every collector declares source → target ──────────
#[test]
fn every_collector_declares_source_and_target() {
let descriptors = registry_descriptors();
assert!(
descriptors.len() >= 8,
"≥8 default collectors: {}",
descriptors.len()
);
// The map-scene collectors must name their facett scene.
for (meta, st) in &descriptors {
if st.target == TargetKind::MapScene {
assert!(
st.map_scene.is_some(),
"{} MapScene must name a scene",
meta.name
);
}
// sanity: label round-trips
assert!(st.label().contains(st.source.as_str()));
}
// AIS + Trafikverket are position overlays → map-scene targets.
let by = |name: &str| {
descriptors
.iter()
.find(|(m, _)| m.name == name)
.map(|(_, s)| s)
};
// `expect`, not `if let Some(..)`: the old form named `ais-positions`, and when that
// preset was renamed to `digitraffic-ais` on 2026-09-03 the lookup would simply have
// returned `None` and the two assertions below would have stopped running — green,
// and proving nothing. A registry that cannot produce the AIS descriptor is a defect.
let ais = by("digitraffic-ais").expect("the AIS collector is in the registry");
assert_eq!(ais.source, SourceEndpoint::Ais);
assert_eq!(ais.target, TargetKind::MapScene);
// Emit a descriptor row per collector so `nornir test` sees them all.
emit_registry_descriptors();
functional_status(
"korp-collectors",
"registry_descriptors_complete",
true,
&format!("{} collectors declare (source → target)", descriptors.len()),
);
}