tests/collectors_parsers.rs
//! Offline INJECT/ASSERT contracts for the pure-Rust parsers — feed each
//! collector's real parser canned bytes and assert the round-tripped Arrow as
//! DATA (schema + values), not "didn't panic". Every check also emits a
//! functional-status row under `--features testmatrix`.
//!
//! Every parser here lives behind `parsers`, so this target self-gates to match
//! (as the csv/osm/mainframe/znippy suites already do) — otherwise the binary
//! fails to COMPILE under `--no-default-features` instead of being skipped.
#![cfg(feature = "parsers")]
use arrow_array::{Array, Float64Array, Int64Array, StringArray};
use arrow_schema::DataType;
use korp_collectors::dcat::{
eu_discovery_query, parse_dcat_rdfxml, parse_sparql_results, registry_batch, registry_schema,
};
use korp_collectors::entsoe::{eic_of, parse_entsoe_xml, points_batch, zone_of};
use korp_collectors::entsoe_generation::{
gen_points_batch, parse_entsoe_generation_xml, psr_label,
};
use korp_collectors::osm_admin::{
WorldAdminTrie, admin_batch, admin_unit_from_tags, country_division_name, extract_admin_units,
};
use korp_collectors::pxweb::parse_jsonstat2;
use korp_collectors::rest::{
FieldMap, FieldType, RestCollector, filter_by_bounds, parse_records, parse_records_bounded,
};
use korp_collectors::trafikverket::{TrafikverketCollector, build_request};
use korp_collectors::{Bounds, Direction, by_name, registry};
mod common;
use common::emit_for;
// ── #1 DCAT-AP-SE ───────────────────────────────────────────────────────────
const DCAT: &str = r#"<?xml version="1.0" encoding="utf-8"?>
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
xmlns:dcat="http://www.w3.org/ns/dcat#"
xmlns:dct="http://purl.org/dc/terms/"
xmlns:foaf="http://xmlns.com/foaf/0.1/">
<dcat:Dataset rdf:about="https://data.example.se/dataset/befolkning">
<dct:title>Befolkning per kommun</dct:title>
<dct:publisher rdf:resource="https://data.example.se/org/scb"/>
<dct:license rdf:resource="http://creativecommons.org/publicdomain/zero/1.0/"/>
<dcat:distribution rdf:resource="https://data.example.se/dist/befolkning.csv"/>
</dcat:Dataset>
<dcat:Distribution rdf:about="https://data.example.se/dist/befolkning.csv">
<dcat:downloadURL rdf:resource="https://data.example.se/files/befolkning.csv"/>
<dct:format>CSV</dct:format>
<dcat:mediaType>text/csv</dcat:mediaType>
</dcat:Distribution>
<foaf:Agent rdf:about="https://data.example.se/org/scb">
<foaf:name>Statistiska centralbyrån</foaf:name>
</foaf:Agent>
<dcat:Dataset rdf:about="https://data.example.se/dataset/empty">
<dct:title>Tom dataset</dct:title>
</dcat:Dataset>
</rdf:RDF>"#;
#[test]
fn dcat_ap_se_parses_into_registry_rows_and_arrow() {
let rows = parse_dcat_rdfxml(DCAT.as_bytes()).expect("parse RDF/XML");
assert_eq!(
rows.len(),
2,
"one row per (dataset, distribution): {rows:#?}"
);
assert_eq!(
rows[0].dataset_uri,
"https://data.example.se/dataset/befolkning"
);
assert_eq!(
rows[0].publisher.as_deref(),
Some("Statistiska centralbyrån")
);
assert_eq!(
rows[0].download_url.as_deref(),
Some("https://data.example.se/files/befolkning.csv")
);
assert!(
rows[1].download_url.is_none(),
"empty dataset → no distribution"
);
let batch = registry_batch(&rows).expect("registry batch");
assert_eq!(batch.num_rows(), 2);
assert_eq!(&batch.schema(), &std::sync::Arc::new(registry_schema()));
let uri = batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(uri.value(0), "https://data.example.se/dataset/befolkning");
let dl = batch
.column(5)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert!(dl.is_null(1), "empty dataset's download_url cell is null");
emit_for(
"korp-collectors/dcat",
"dcat_rdfxml_registry_rows",
true,
&format!(
"{} registry rows, publisher via foaf:name resolved",
rows.len()
),
);
}
// ── #2 PxWeb / JSON-stat2 ───────────────────────────────────────────────────
const CUBE: &str = r#"{
"version": "2.0", "class": "dataset",
"id": ["region", "year", "ContentsCode"],
"size": [2, 2, 1],
"dimension": {
"region": { "category": { "index": { "0114": 0, "0115": 1 } } },
"year": { "category": { "index": { "2020": 0, "2021": 1 } } },
"ContentsCode": { "category": { "index": { "BE0101N1": 0 } } }
},
"value": [100, 110, 200, 210]
}"#;
#[test]
fn pxweb_jsonstat2_cube_normalises_to_arrow() {
let v: serde_json::Value = serde_json::from_str(CUBE).unwrap();
let batch = parse_jsonstat2(&v).expect("normalise cube");
assert_eq!(batch.num_rows(), 4, "2*2*1 cells");
let schema = batch.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(names, vec!["region", "year", "ContentsCode", "value"]);
let region = batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let year = batch
.column(1)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let value = batch
.column(3)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
// last dim varies fastest: (0114,2020),(0114,2021),(0115,2020),(0115,2021)
assert_eq!(
(region.value(0), year.value(0), value.value(0)),
("0114", "2020", 100.0)
);
assert_eq!(
(region.value(1), year.value(1), value.value(1)),
("0114", "2021", 110.0)
);
assert_eq!(
(region.value(2), year.value(2), value.value(2)),
("0115", "2020", 200.0)
);
assert_eq!(value.value(3), 210.0);
emit_for(
"korp-collectors/pxweb",
"jsonstat2_cube_to_arrow",
true,
"4 cells, row-major decode ok",
);
}
// ── #4 ENTSO-E ──────────────────────────────────────────────────────────────
const ENTSOE: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
<Publication_MarketDocument xmlns="urn:iec62325.351:tc57wg16:451-3:publicationdocument:7:0">
<TimeSeries>
<mRID>1</mRID>
<in_Domain.mRID codingScheme="A01">10Y1001A1001A46L</in_Domain.mRID>
<Period>
<timeInterval><start>2026-01-01T00:00Z</start><end>2026-01-01T02:00Z</end></timeInterval>
<resolution>PT60M</resolution>
<Point><position>1</position><quantity>42.5</quantity></Point>
<Point><position>2</position><quantity>43.0</quantity></Point>
</Period>
</TimeSeries>
</Publication_MarketDocument>"#;
#[test]
fn entsoe_market_document_maps_se3_points() {
assert_eq!(zone_of("10Y1001A1001A46L"), Some("SE3"));
assert_eq!(eic_of("se1"), Some("10Y1001A1001A44P"));
let points = parse_entsoe_xml(ENTSOE).expect("parse ENTSO-E xml");
assert_eq!(points.len(), 2);
assert_eq!(points[0].bidding_zone.as_deref(), Some("SE3"));
assert_eq!(points[0].period_start.as_deref(), Some("2026-01-01T00:00Z"));
assert_eq!(points[0].quantity, 42.5);
let batch = points_batch(&points).unwrap();
let zone = batch
.column(1)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let pos = batch
.column(4)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert_eq!(zone.value(0), "SE3");
assert_eq!(pos.value(1), 2);
emit_for(
"korp-collectors/entsoe",
"entsoe_xml_points_se3",
true,
"2 points, EIC→SE3 mapped",
);
}
// ── #4b ENTSO-E generation per type ─────────────────────────────────────────
const ENTSOE_GEN: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
<GL_MarketDocument xmlns="urn:iec62325.351:tc57wg16:451-6:generationloaddocument:3:0">
<TimeSeries>
<mRID>1</mRID>
<in_Domain.mRID codingScheme="A01">10Y1001A1001A46L</in_Domain.mRID>
<MktPSRType><psrType>B16</psrType></MktPSRType>
<Period>
<timeInterval><start>2026-01-01T00:00Z</start><end>2026-01-01T02:00Z</end></timeInterval>
<resolution>PT60M</resolution>
<Point><position>1</position><quantity>120.0</quantity></Point>
<Point><position>2</position><quantity>135.5</quantity></Point>
</Period>
</TimeSeries>
<TimeSeries>
<mRID>2</mRID>
<in_Domain.mRID codingScheme="A01">10Y1001A1001A46L</in_Domain.mRID>
<MktPSRType><psrType>B19</psrType></MktPSRType>
<Period>
<timeInterval><start>2026-01-01T00:00Z</start><end>2026-01-01T02:00Z</end></timeInterval>
<resolution>PT60M</resolution>
<Point><position>1</position><quantity>900.0</quantity></Point>
<Point><position>2</position><quantity>880.0</quantity></Point>
</Period>
</TimeSeries>
</GL_MarketDocument>"#;
#[test]
fn entsoe_generation_maps_psr_type_per_series() {
assert_eq!(psr_label("B16"), Some("Solar"));
assert_eq!(psr_label("B19"), Some("Wind Onshore"));
assert_eq!(psr_label("ZZZ"), None);
let points = parse_entsoe_generation_xml(ENTSOE_GEN).expect("parse ENTSO-E A75 xml");
assert_eq!(points.len(), 4, "2 series x 2 points");
// Series 1 = Solar (B16), SE3.
assert_eq!(points[0].bidding_zone.as_deref(), Some("SE3"));
assert_eq!(points[0].psr_type.as_deref(), Some("B16"));
assert_eq!(points[0].psr_label.as_deref(), Some("Solar"));
assert_eq!(points[0].period_start.as_deref(), Some("2026-01-01T00:00Z"));
assert_eq!(points[0].quantity, 120.0);
// Series 2 = Wind Onshore (B19) — psrType reset per TimeSeries, not leaked.
assert_eq!(points[2].psr_type.as_deref(), Some("B19"));
assert_eq!(points[2].psr_label.as_deref(), Some("Wind Onshore"));
assert_eq!(points[2].quantity, 900.0);
let batch = gen_points_batch(&points).unwrap();
assert_eq!(batch.num_rows(), 4);
let schema = batch.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names,
vec![
"in_domain",
"bidding_zone",
"psr_type",
"psr_label",
"resolution",
"period_start",
"position",
"quantity"
]
);
let label = batch
.column(3)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let pos = batch
.column(6)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let qty = batch
.column(7)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(label.value(0), "Solar");
assert_eq!(label.value(2), "Wind Onshore");
assert_eq!(pos.value(3), 2);
assert_eq!(qty.value(2), 900.0);
emit_for(
"korp-collectors/entsoe_generation",
"entsoe_a75_psr_type_per_series",
true,
"4 points across 2 series, psrType B16/B19 labelled, SE3 mapped",
);
}
// ── #5 Generic REST/JSON (Polisen + Kolada) + AIS position table ────────────
const POLISEN: &str = r#"[
{"id":123,"datetime":"2026-06-30 12:00:00","name":"Trafikolycka","type":"Trafik",
"summary":"x","url":"https://polisen.se/x","location":{"name":"Stockholm","gps":"59.3,18.0"}},
{"id":124,"datetime":"2026-06-30 13:00:00","name":"Stöld","type":"Stöld",
"summary":"y","url":"https://polisen.se/y","location":{"name":"Uppsala","gps":"59.8,17.6"}}
]"#;
/// The AIS document is the crate's OWN fixture — two GeoJSON features copied from the
/// live Digitraffic response on 2026-09-03 — rather than a second hand-written one that
/// could drift from the preset it is meant to prove (L5).
const AIS: &str = korp_collectors::sample::SYNTHETIC_AIS_JSON;
#[test]
fn rest_polisen_and_ais_map_to_arrow() {
let pol = RestCollector::polisen();
let json: serde_json::Value = serde_json::from_str(POLISEN).unwrap();
let batch = parse_records(&json, pol.records_pointer(), pol.fields()).expect("map polisen");
assert_eq!(batch.num_rows(), 2);
let id = batch
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let loc = batch
.column(6)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(id.value(0), 123);
assert_eq!(loc.value(1), "Uppsala", "nested /location/name pointer");
let ais = RestCollector::digitraffic_ais();
let json: serde_json::Value = serde_json::from_str(AIS).unwrap();
let batch = parse_records(&json, ais.records_pointer(), ais.fields()).expect("map ais");
assert_eq!(batch.num_rows(), 2, "two vessels in the Digitraffic fixture");
// Columns BY NAME, not by index: the field map gained `nav_stat`/`ts_second` when
// the placeholder feed became Digitraffic, and a positional lookup would then have
// been asserting the wrong column while still passing.
let mmsi = batch
.column_by_name("mmsi")
.and_then(|c| c.as_any().downcast_ref::<Int64Array>())
.expect("mmsi i64");
let lon = batch
.column_by_name("lon")
.and_then(|c| c.as_any().downcast_ref::<Float64Array>())
.expect("lon f64");
let lat = batch
.column_by_name("lat")
.and_then(|c| c.as_any().downcast_ref::<Float64Array>())
.expect("lat f64");
assert_eq!(mmsi.value(0), 265_676_640);
// coordinates = [lon, lat]: a swapped pair would put this vessel ~41° away.
assert!((lon.value(0) - 18.854_418).abs() < 1e-9, "lon {}", lon.value(0));
assert!((lat.value(0) - 59.654_993).abs() < 1e-9, "lat {}", lat.value(0));
emit_for(
"korp-collectors/rest",
"rest_json_field_map",
true,
"polisen 2 rows + 2 digitraffic AIS position rows",
);
}
// ── USGS earthquakes (GeoJSON FeatureCollection → quake rows) ────────────────
const USGS: &str = r#"{
"type": "FeatureCollection",
"metadata": {"count": 2},
"features": [
{"type":"Feature","id":"nc73012345",
"properties":{"mag":4.2,"place":"10km N of Reno","time":1752400000000,"type":"earthquake",
"url":"https://earthquake.usgs.gov/e/nc73012345"},
"geometry":{"type":"Point","coordinates":[-119.81,39.53,8.4]}},
{"type":"Feature","id":"line1","properties":{"mag":9.9},
"geometry":{"type":"LineString","coordinates":[[0,0],[1,1]]}},
{"type":"Feature","id":"us7000abcd",
"properties":{"mag":5.9,"place":"South of Fiji","time":1752400500000,"type":"earthquake"},
"geometry":{"type":"Point","coordinates":[178.12,-22.34,560.0]}}
]
}"#;
#[test]
fn usgs_geojson_flattens_to_quake_rows_and_arrow() {
use korp_collectors::usgs::{parse_usgs_geojson, quakes_batch};
let json: serde_json::Value = serde_json::from_str(USGS).unwrap();
let quakes = parse_usgs_geojson(&json).expect("parse USGS GeoJSON");
// the LineString is not a locatable Point → dropped
assert_eq!(
quakes.len(),
2,
"two Point features (LineString dropped): {quakes:#?}"
);
assert_eq!(quakes[0].id, "nc73012345");
assert_eq!(quakes[0].magnitude, Some(4.2));
assert!((quakes[0].longitude - -119.81).abs() < 1e-9);
assert!((quakes[0].latitude - 39.53).abs() < 1e-9);
assert_eq!(quakes[0].depth_km, Some(8.4));
assert_eq!(quakes[0].time_ms, Some(1752400000000));
let batch = quakes_batch(&quakes).expect("quakes batch");
assert_eq!(batch.num_rows(), 2);
let lon = batch
.column_by_name("longitude")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let time = batch
.column_by_name("time")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert!((lon.value(1) - 178.12).abs() < 1e-9);
assert_eq!(time.value(0), 1752400000000);
assert_eq!(
batch.schema().field_with_name("time").unwrap().data_type(),
&DataType::Int64
);
emit_for(
"korp-collectors/usgs-earthquakes",
"geojson_quake_rows",
true,
"2 Point quakes, LineString dropped, coords→lon/lat/depth",
);
}
// ── OpenAQ air quality (results envelope → measurement rows) ─────────────────
const OPENAQ: &str = r#"{
"meta": {"found": 3},
"results": [
{"locationId": 8118, "location": "Stockholm Torkel", "parameter": "pm25", "value": 7.4,
"unit": "µg/m³", "country": "SE", "city": "Stockholm",
"coordinates": {"latitude": 59.3162, "longitude": 18.0575},
"date": {"utc": "2026-07-14T08:00:00Z"}},
{"locationId": 999, "parameter": "pm10", "value": null,
"coordinates": {"latitude": 59.0, "longitude": 18.0}},
{"locationId": 2178, "location": "Göteborg Femman", "parameter": "no2", "value": "21.9",
"unit": "µg/m³", "country": "SE",
"coordinates": {"latitude": 57.7075, "longitude": 11.9675},
"date": {"utc": "2026-07-14T08:00:00Z"}}
]
}"#;
#[test]
fn openaq_json_flattens_to_measurement_rows_and_arrow() {
use korp_collectors::openaq::{measurements_batch, parse_openaq_json};
let json: serde_json::Value = serde_json::from_str(OPENAQ).unwrap();
let rows = parse_openaq_json(&json).expect("parse OpenAQ results");
// the valueless pm10 reading is dropped
assert_eq!(
rows.len(),
2,
"two usable located readings (valueless dropped): {rows:#?}"
);
assert_eq!(rows[0].parameter, "pm25");
assert_eq!(rows[0].value, 7.4);
assert!((rows[0].longitude - 18.0575).abs() < 1e-9);
assert!((rows[0].latitude - 59.3162).abs() < 1e-9);
assert_eq!(rows[0].date_utc.as_deref(), Some("2026-07-14T08:00:00Z"));
assert_eq!(rows[1].value, 21.9, "string value coerced to f64");
let batch = measurements_batch(&rows).expect("measurements batch");
assert_eq!(batch.num_rows(), 2);
let value = batch
.column_by_name("value")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let param = batch
.column_by_name("parameter")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(param.value(1), "no2");
assert!((value.value(1) - 21.9).abs() < 1e-9);
assert_eq!(
batch.schema().field_with_name("value").unwrap().data_type(),
&DataType::Float64
);
assert_eq!(
batch
.schema()
.field_with_name("location_id")
.unwrap()
.data_type(),
&DataType::Int64
);
emit_for(
"korp-collectors/openaq-air-quality",
"openaq_measurement_rows",
true,
"2 located readings, valueless dropped, coordinates+date.utc flattened",
);
}
// ── Energy-Charts (parallel arrays → transposed power rows) ─────────────────
const ENERGY_CHARTS: &str = r#"{
"unix_seconds": [1751328000, 1751328900, 1751329800],
"production_types": [
{"name": "Solar", "data": [1200.5, 1810.0, null]},
{"name": "Wind onshore", "data": [640.0, 655.5]},
{"name": "Hydro", "data": ["33.5", "n/a", 40.0]}
]
}"#;
#[test]
fn energy_charts_parallel_arrays_transpose_to_power_rows_and_arrow() {
use korp_collectors::energy_charts::{parse_energy_charts_json, power_batch};
let json: serde_json::Value = serde_json::from_str(ENERGY_CHARTS).unwrap();
let readings = parse_energy_charts_json(&json).expect("transpose energy-charts");
// Solar: t0,t1 (null t2 dropped) = 2; Wind: short data → 2; Hydro: "33.5"+40.0 = 2.
assert_eq!(
readings.len(),
6,
"ragged/null/non-numeric cells guarded: {readings:#?}"
);
// The value is recovered by zipping the two parallel arrays by index.
let solar_t1 = readings
.iter()
.find(|r| r.production_type == "Solar" && r.time_s == 1751328900)
.expect("Solar t1");
assert!(
(solar_t1.power_mw - 1810.0).abs() < 1e-9,
"Solar t1 = data[1]"
);
// Wind's short array never fabricates a third instant.
assert!(
!readings
.iter()
.any(|r| r.production_type == "Wind onshore" && r.time_s == 1751329800)
);
let batch = power_batch(&readings).expect("power batch");
assert_eq!(batch.num_rows(), 6);
let time = batch
.column_by_name("time")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let power = batch
.column_by_name("power_mw")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(
time.value(0),
1751328000,
"row order is type-major then time"
);
assert!((power.value(0) - 1200.5).abs() < 1e-9);
assert_eq!(
batch.schema().field_with_name("time").unwrap().data_type(),
&DataType::Int64
);
assert_eq!(
batch
.schema()
.field_with_name("power_mw")
.unwrap()
.data_type(),
&DataType::Float64
);
emit_for(
"korp-collectors/energy-charts-power",
"parallel_array_transpose",
true,
"6 (time,type) rows, index-zip of parallel arrays, ragged/null guarded",
);
}
// ── GTFS-Realtime (protobuf FeedMessage → vehicle position rows) ────────────
#[test]
fn gtfs_rt_protobuf_decodes_to_vehicle_position_rows_and_arrow() {
use korp_collectors::gtfs_rt::{VehicleFixture, build_gtfs_rt, parse_gtfs_rt, vehicles_batch};
// Encode a two-vehicle FeedMessage (header + vehicles) and decode it back —
// the binary protobuf wire round-trip, no network.
let pb = build_gtfs_rt(&[
VehicleFixture {
entity_id: "veh-1",
vehicle_id: Some("9031001"),
trip_id: Some("trip-A"),
route_id: Some("17"),
latitude: 59.33107,
longitude: 18.05977,
bearing: Some(137.5),
speed: Some(11.4),
timestamp: Some(1_752_400_123),
stop_id: Some("9022001"),
},
VehicleFixture {
entity_id: "veh-2",
vehicle_id: Some("9031002"),
route_id: Some("96"),
latitude: 57.70750,
longitude: 11.96750,
..Default::default()
},
]);
let rows = parse_gtfs_rt(&pb).expect("decode GTFS-RT protobuf");
assert_eq!(rows.len(), 2, "two vehicle entities → two rows: {rows:#?}");
assert_eq!(rows[0].vehicle_id.as_deref(), Some("9031001"));
assert_eq!(rows[0].route_id.as_deref(), Some("17"));
assert!(
(rows[0].latitude - 59.33107).abs() < 1e-3,
"lat from float wire"
);
assert!(
(rows[0].longitude - 18.05977).abs() < 1e-3,
"lon from float wire"
);
assert_eq!(rows[0].timestamp, Some(1_752_400_123));
// The second vehicle's absent bearing/speed/stop stay None (not fabricated).
assert!(rows[1].bearing.is_none() && rows[1].speed.is_none() && rows[1].stop_id.is_none());
let batch = vehicles_batch(&rows).expect("vehicles batch");
assert_eq!(batch.num_rows(), 2);
let lat = batch
.column_by_name("latitude")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let route = batch
.column_by_name("route_id")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let ts = batch
.column_by_name("timestamp")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert!((lat.value(1) - 57.7075).abs() < 1e-3);
assert_eq!(route.value(0), "17");
assert_eq!(ts.value(0), 1_752_400_123);
assert_eq!(
batch
.schema()
.field_with_name("latitude")
.unwrap()
.data_type(),
&DataType::Float64
);
assert_eq!(
batch
.schema()
.field_with_name("timestamp")
.unwrap()
.data_type(),
&DataType::Int64
);
// A truncated feed is a hard error, not a panic (the malformed-input guard).
assert!(parse_gtfs_rt(&[0x80]).is_err(), "truncated varint → Err");
emit_for(
"korp-collectors/gtfs-rt-vehicles",
"protobuf_vehicle_positions",
true,
"2 vehicle rows from a hand-decoded GTFS-RT protobuf FeedMessage, header/non-vehicle skipped",
);
}
// ── Trafikverket (POST XML → JSON) ──────────────────────────────────────────
const TV: &str = r#"{"RESPONSE":{"RESULT":[{"TrainPosition":[
{"Train":{"OperationalTrainNumber":"527"},"Position":{"WGS84":"POINT (16.18 59.61)"},
"TimeStamp":"2026-06-30T12:00:00Z","Bearing":270,"Speed":118,"Status":{"Active":true}}
]}]}}"#;
#[test]
fn trafikverket_trainpos_and_key_injection() {
let c = TrafikverketCollector::train_positions();
assert_eq!(c.records_pointer(), "/RESPONSE/RESULT/0/TrainPosition");
let json: serde_json::Value = serde_json::from_str(TV).unwrap();
let batch = parse_records(&json, &c.records_pointer(), c.fields()).expect("map trainpos");
assert_eq!(batch.num_rows(), 1);
let train = batch
.column(0)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let speed = batch
.column(4)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(train.value(0), "527");
assert_eq!(speed.value(0), 118.0);
let body = build_request("SECRET", "<QUERY objecttype=\"TrainPosition\"/>");
assert!(body.contains("authenticationkey=\"SECRET\""));
assert!(body.starts_with("<REQUEST><LOGIN"));
emit_for(
"korp-collectors/trafikverket",
"trainpos_and_key_in_body",
true,
"1 position row, key injected in <LOGIN>",
);
}
// ── OSM admin boundaries (pure tag-filter + trie, no PBF) ────────────────────
#[test]
fn osm_admin_per_country_levels_filter_and_trie() {
assert_eq!(country_division_name("SE", 7), Some("kommun"));
assert_eq!(country_division_name("NO", 4), Some("fylke"));
assert_eq!(country_division_name("FI", 8), Some("kunta/kommun"));
assert_eq!(country_division_name("DE", 4), Some("state/region"));
let kommun = r#"{"boundary":"administrative","admin_level":"7","name":"Solna kommun","ISO3166-2":"SE-AB"}"#;
let u = admin_unit_from_tags(111, kommun, None).expect("admin unit");
assert_eq!(u.country.as_deref(), Some("SE"));
assert_eq!(u.division.as_deref(), Some("kommun"));
// a non-boundary feature is dropped
assert!(admin_unit_from_tags(9, r#"{"highway":"primary"}"#, None).is_none());
let features = vec![
(1, r#"{"boundary":"administrative","admin_level":"7","name":"Solna","ISO3166-2":"SE-AB"}"#.to_string(), None),
(2, r#"{"boundary":"administrative","admin_level":"7","name":"Sundbyberg","ISO3166-2":"SE-AB"}"#.to_string(), Some(vec![1u8, 2, 3])),
(3, r#"{"boundary":"administrative","admin_level":"4","name":"Stockholms län","ISO3166-2":"SE-AB"}"#.to_string(), None),
(4, r#"{"boundary":"administrative","admin_level":"7","name":"Oslo","ISO3166-2":"NO-03"}"#.to_string(), None),
(5, r#"{"highway":"primary"}"#.to_string(), None),
];
let units = extract_admin_units(features);
assert_eq!(units.len(), 4, "the road is filtered out");
let batch = admin_batch(&units).unwrap();
assert_eq!(batch.num_rows(), 4);
let country = batch
.column(5)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(country.value(0), "SE");
let trie = WorldAdminTrie::build(&units);
assert_eq!(trie.countries(), 2, "SE + NO");
assert_eq!(trie.count("SE", 7), 2);
assert_eq!(trie.count("NO", 7), 1);
emit_for(
"korp-collectors/osm_admin",
"admin_tag_filter_and_trie",
true,
"4 admin units, SE/NO trie, road dropped",
);
}
// ── registry / direction discovery ──────────────────────────────────────────
#[test]
fn registry_is_discovery_complete_and_sources() {
let names: Vec<String> = registry().iter().map(|c| c.meta().name).collect();
for want in [
"dcat-ap-se",
"pxweb-jsonstat2",
"entsoe",
"polisen-events",
"kolada-kpi",
"smhi-metobs",
] {
assert!(
names.contains(&want.to_string()),
"registry missing {want}: {names:?}"
);
}
assert!(by_name("entsoe").is_some());
// DIRECTION marker: every collector is a data SOURCE.
for c in registry() {
assert_eq!(
c.meta().direction,
Direction::Source,
"{} must be a Source",
c.meta().name
);
}
emit_for(
"korp-collectors",
"registry_discovery_complete",
true,
&format!("{} collectors, all Direction::Source", names.len()),
);
}
// ── #1 DCAT SPARQL results path (data.europa.eu discovery) ───────────────────
// The SPARQL-results-JSON branch of the DCAT crawler is wired into
// `DcatCrawler::collect` but was previously untested (only the RDF path was).
const SPARQL_RESULTS: &str = r#"{
"head": { "vars": ["dataset","title","publisher","license","distribution","downloadURL","accessURL","format","mediaType"] },
"results": { "bindings": [
{
"dataset": {"type":"uri","value":"https://data.europa.eu/set/1"},
"title": {"type":"literal","value":"Population by municipality"},
"publisher": {"type":"literal","value":"SCB"},
"license": {"type":"uri","value":"http://creativecommons.org/publicdomain/zero/1.0/"},
"distribution":{"type":"uri","value":"https://data.europa.eu/set/1/dist"},
"downloadURL": {"type":"uri","value":"https://data.europa.eu/set/1.csv"},
"accessURL": {"type":"uri","value":"https://data.europa.eu/set/1/access"},
"format": {"type":"literal","value":"CSV"},
"mediaType": {"type":"literal","value":"text/csv"}
},
{
"title": {"type":"literal","value":"no dataset binding → skipped"}
}
]}
}"#;
#[test]
fn dcat_sparql_results_map_to_registry_rows() {
let json: serde_json::Value = serde_json::from_str(SPARQL_RESULTS).unwrap();
let rows = parse_sparql_results(&json).expect("parse SPARQL results");
// the binding without a `dataset` value is skipped
assert_eq!(
rows.len(),
1,
"binding missing `dataset` is dropped: {rows:#?}"
);
let r = &rows[0];
assert_eq!(r.dataset_uri, "https://data.europa.eu/set/1");
assert_eq!(
r.dataset_title.as_deref(),
Some("Population by municipality")
);
assert_eq!(r.publisher.as_deref(), Some("SCB"));
assert_eq!(
r.distribution_uri.as_deref(),
Some("https://data.europa.eu/set/1/dist")
);
assert_eq!(
r.download_url.as_deref(),
Some("https://data.europa.eu/set/1.csv")
);
assert_eq!(
r.access_url.as_deref(),
Some("https://data.europa.eu/set/1/access")
);
assert_eq!(r.format.as_deref(), Some("CSV"));
assert_eq!(r.media_type.as_deref(), Some("text/csv"));
// the single `license` binding maps to BOTH dataset_license and dist_license
let lic = Some("http://creativecommons.org/publicdomain/zero/1.0/");
assert_eq!(r.dataset_license.as_deref(), lic);
assert_eq!(r.dist_license.as_deref(), lic);
// it flows through the same Arrow batch contract as the RDF path
let batch = registry_batch(&rows).expect("registry batch");
assert_eq!(batch.num_rows(), 1);
assert_eq!(&batch.schema(), &std::sync::Arc::new(registry_schema()));
// a non-object / missing results.bindings is a hard error, not a silent empty
let bad: serde_json::Value = serde_json::json!({"results": {}});
assert!(
parse_sparql_results(&bad).is_err(),
"no results.bindings → Err"
);
emit_for(
"korp-collectors/dcat",
"sparql_results_to_registry_rows",
true,
"1 row (missing-dataset binding skipped), license→both",
);
}
#[test]
fn eu_discovery_query_has_prefixes_optionals_and_limit() {
let q = eu_discovery_query(37);
assert!(
q.contains("PREFIX dcat: <http://www.w3.org/ns/dcat#>"),
"dcat PREFIX: {q}"
);
assert!(
q.contains("PREFIX dct: <http://purl.org/dc/terms/>"),
"dct PREFIX"
);
assert!(q.contains("?dataset a dcat:Dataset"), "selects datasets");
assert!(
q.contains("OPTIONAL { ?dataset dcat:distribution ?distribution"),
"OPTIONAL distribution block"
);
assert!(
q.contains("OPTIONAL { ?distribution dcat:downloadURL ?downloadURL }"),
"nested downloadURL optional"
);
assert!(q.trim_end().ends_with("LIMIT 37"), "carries the LIMIT: {q}");
emit_for(
"korp-collectors/dcat",
"eu_discovery_query_shape",
true,
"dcat/dct prefixes, OPTIONAL distribution block, LIMIT interpolated",
);
}
// ── #5 Generic REST custom adapter (the config-driven escape hatch) ──────────
#[test]
fn rest_custom_adapter_maps_nested_records() {
// A custom adapter over a nested records pointer with mixed field types, and
// string↔number coercion + null handling exercised per column.
let custom = RestCollector::custom(
"myapi",
"/payload/records",
vec![
FieldMap::new("name", "/name", FieldType::Str),
FieldMap::new("count", "/stats/count", FieldType::I64),
FieldMap::new("ratio", "/stats/ratio", FieldType::F64),
],
);
assert_eq!(custom.records_pointer(), "/payload/records");
assert_eq!(custom.fields().len(), 3);
let json = serde_json::json!({
"payload": { "records": [
{ "name": "alpha", "stats": { "count": "5", "ratio": 1.5 } }, // count as STRING → coerced
{ "name": "beta", "stats": { "count": 10, "ratio": "2.25" } }, // ratio as STRING → coerced
{ "name": null, "stats": {} } // nulls / missing pointers
]}
});
let rb = parse_records(&json, custom.records_pointer(), custom.fields())
.expect("parse custom records");
assert_eq!(rb.num_rows(), 3);
// schema types come from the FieldMap declaration
let schema = rb.schema();
assert_eq!(schema.field(0).data_type(), &DataType::Utf8);
assert_eq!(schema.field(1).data_type(), &DataType::Int64);
assert_eq!(schema.field(2).data_type(), &DataType::Float64);
let name = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
let count = rb.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
let ratio = rb
.column(2)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(name.value(0), "alpha");
assert_eq!(count.value(0), 5, "string \"5\" coerced to i64");
assert_eq!(ratio.value(0), 1.5);
assert_eq!(count.value(1), 10);
assert_eq!(ratio.value(1), 2.25, "string \"2.25\" coerced to f64");
// row 3: null name + absent nested pointers → null cells
assert!(name.is_null(2), "explicit JSON null → null cell");
assert!(count.is_null(2), "missing /stats/count pointer → null cell");
assert!(ratio.is_null(2), "missing /stats/ratio pointer → null cell");
emit_for(
"korp-collectors/rest",
"rest_custom_adapter",
true,
"nested pointer, mixed types, string↔number coercion + nulls",
);
}
// ── CDC / incremental Bounds seam (RestCollector) ────────────────────────────
#[test]
fn bounds_narrow_output() {
// Records spanning a full year, plus one undated record.
let json = serde_json::json!({ "rows": [
{ "id": 1, "ts": "2026-01-01T00:00:00Z" },
{ "id": 2, "ts": "2026-06-15T12:00:00Z" },
{ "id": 3, "ts": "2026-12-31T23:59:59Z" },
{ "id": 4 }
]});
let fields = vec![
FieldMap::new("id", "/id", FieldType::I64),
FieldMap::new("ts", "/ts", FieldType::Str),
];
// A closed [since, until] window keeps only the mid-year record.
let window = Bounds {
since: Some("2026-03-01T00:00:00Z".into()),
until: Some("2026-09-01T00:00:00Z".into()),
..Default::default()
};
let recs = json.pointer("/rows").unwrap().as_array().unwrap();
let kept = filter_by_bounds(recs, "/ts", &window);
assert_eq!(
kept.len(),
1,
"one in-range record (undated dropped when filtering)"
);
let rb = parse_records_bounded(&json, "/rows", &fields, "/ts", &window).expect("bounded parse");
assert_eq!(rb.num_rows(), 1);
let id = rb.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
assert_eq!(id.value(0), 2, "only the in-window record lands");
// Empty bounds is a NO-OP: every record (incl. the undated one) is kept.
let all = parse_records_bounded(&json, "/rows", &fields, "/ts", &Bounds::default()).unwrap();
assert_eq!(all.num_rows(), 4, "empty bounds keeps everything");
// Open-ended upper bound: `since` only → records 2 and 3.
let since_only = Bounds {
since: Some("2026-06-01T00:00:00Z".into()),
until: None,
..Default::default()
};
let rb2 = parse_records_bounded(&json, "/rows", &fields, "/ts", &since_only).unwrap();
assert_eq!(
rb2.num_rows(),
2,
"since-only keeps everything at/after the mark"
);
emit_for(
"korp-collectors/rest",
"bounds_narrow_output",
true,
"CDC bounds filter narrows in-range rows; empty bounds = no-op",
);
}