tests/collectors_http.rs
//! Offline mock-server tests for the shared HTTP core's **resilience loop**
//! ([`HttpCore::send`]) and the 0-row **RAGNARÖK reason** on the network
//! collectors — the parts the module comment admitted were only covered by
//! non-existent `#[ignore]` network tests. A tiny in-process HTTP/1.1 server
//! (the `spawn_pbf_server` pattern from `collectors_osm.rs`) serves a scripted
//! sequence of responses and counts requests, so retry/no-retry is proven by the
//! request count with no external network.
#![cfg(feature = "parsers")]
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use korp_collectors::http::{HttpCore, NoAuth};
use korp_collectors::{BatchSink, Bounds, Collector, CollectorCfg};
mod common;
use common::emit_for;
/// Serve `responses` in order — one `(status, body)` per accepted connection —
/// then exit. Returns `(base_url, hit_counter, join_handle)`. The server is sized
/// to exactly the number of requests the test expects, so a spurious extra
/// request would block (never served) and `handle.join()` would prove the miss.
fn spawn_http(
responses: Vec<(u16, Vec<u8>)>,
) -> (String, Arc<AtomicUsize>, std::thread::JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
let addr = listener.local_addr().unwrap();
let hits = Arc::new(AtomicUsize::new(0));
let hits_srv = Arc::clone(&hits);
let handle = std::thread::spawn(move || {
for (code, body) in responses {
let Ok((mut stream, _)) = listener.accept() else {
break;
};
// Drain the request (headers) before replying.
let mut buf = [0u8; 2048];
let _ = stream.read(&mut buf);
hits_srv.fetch_add(1, Ordering::SeqCst);
let reason = match code {
200 => "OK",
404 => "Not Found",
429 => "Too Many Requests",
500 => "Internal Server Error",
_ => "Status",
};
let header = format!(
"HTTP/1.1 {code} {reason}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.write_all(&body);
let _ = stream.flush();
}
});
(format!("http://{addr}"), hits, handle)
}
#[tokio::test]
async fn retries_5xx_then_succeeds() {
// First connection → 500 (retryable), second → 200: get_text must retry and
// return the eventual body. Two requests reach the server.
let (url, hits, handle) = spawn_http(vec![(500, b"boom".to_vec()), (200, b"ok".to_vec())]);
let http = HttpCore::new(0); // no rate-limit; default 4 retries
let text = http
.get_text(&url, &NoAuth)
.await
.expect("retry past 500 then succeed");
assert_eq!(text, "ok");
assert_eq!(
hits.load(Ordering::SeqCst),
2,
"one 500 retried, then the 200"
);
handle.join().unwrap();
emit_for(
"korp-collectors/http",
"retries_5xx_then_succeeds",
true,
"500→retry→200, 2 requests",
);
}
#[tokio::test]
async fn does_not_retry_4xx() {
// 404 is fatal: exactly one request, then a hard error — no retry storm.
let (url, hits, handle) = spawn_http(vec![(404, b"nope".to_vec())]);
let http = HttpCore::new(0);
let err = match http.get_text(&url, &NoAuth).await {
Ok(_) => panic!("404 must be fatal"),
Err(e) => e,
};
assert!(
err.to_string().contains("404"),
"error carries the status: {err}"
);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"4xx not retried: exactly one request"
);
handle.join().unwrap();
emit_for(
"korp-collectors/http",
"does_not_retry_4xx",
true,
"404 fatal, 1 request",
);
}
#[tokio::test]
async fn with_max_retries_zero_does_not_retry_5xx() {
// The retry knob wired: max_retries(0) fast-fails even a normally-retryable
// 500 after a single attempt.
let (url, hits, handle) = spawn_http(vec![(500, b"boom".to_vec())]);
let http = HttpCore::new(0).with_max_retries(0);
let err = match http.get_text(&url, &NoAuth).await {
Ok(_) => panic!("max_retries(0) must fast-fail"),
Err(e) => e,
};
assert!(
err.to_string().contains("500"),
"error carries the status: {err}"
);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"max_retries(0): exactly one request even on 5xx"
);
handle.join().unwrap();
emit_for(
"korp-collectors/http",
"with_max_retries_zero_fast_fail",
true,
"500 + max_retries(0) → 1 request",
);
}
// ── 0-row RAGNARÖK reason on the network collectors (REST + DCAT) ────────────
// The empty-but-valid payload path sets `CollectStats.reason` (mirroring the
// csv/mainframe `empty_*_ragnarok` tests) — asserted here end-to-end through
// `collect()` against the mock server.
#[tokio::test]
async fn rest_empty_payload_sets_reason() {
use korp_collectors::rest::RestCollector;
// Polisen's records pointer is "" (the doc itself is the array) → an empty
// array is a valid 0-record payload.
let (url, _hits, handle) = spawn_http(vec![(200, b"[]".to_vec())]);
let cfg = CollectorCfg::new(url, "polisen_events");
let sink = BatchSink::memory();
let stats = RestCollector::polisen()
.collect(&cfg, Bounds::default(), &sink)
.await
.expect("collect an empty payload");
assert_eq!(stats.rows, 0);
assert_eq!(sink.batches(), 0, "no batch pushed for an empty payload");
assert!(
stats.reason.is_some(),
"0-row REST collect is RED unless reason set (RAGNARÖK)"
);
handle.join().unwrap();
emit_for(
"korp-collectors/rest",
"rest_empty_payload_ragnarok",
true,
"[] → 0 rows, reason set",
);
}
#[tokio::test]
async fn dcat_empty_graph_sets_reason() {
use korp_collectors::dcat::DcatCrawler;
// A valid but empty RDF/XML graph (zero dcat:Dataset) → 0 registry rows.
let empty_rdf =
br#"<?xml version="1.0" encoding="utf-8"?><rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"></rdf:RDF>"#
.to_vec();
let (url, _hits, handle) = spawn_http(vec![(200, empty_rdf)]);
let cfg = CollectorCfg::new(url, "dcat_registry"); // default mode "rdf" → get_bytes
let sink = BatchSink::memory();
let stats = DcatCrawler
.collect(&cfg, Bounds::default(), &sink)
.await
.expect("collect an empty graph");
assert_eq!(stats.rows, 0);
assert_eq!(sink.batches(), 0, "no batch pushed for an empty graph");
assert!(
stats.reason.is_some(),
"0-row DCAT collect is RED unless reason set (RAGNARÖK)"
);
handle.join().unwrap();
emit_for(
"korp-collectors/dcat",
"dcat_empty_graph_ragnarok",
true,
"empty RDF → 0 rows, reason set",
);
}
// ── CDC seam end-to-end: bounded collect() + checkpoint() round-trip ─────────
#[tokio::test]
async fn rest_bounds_and_checkpoint_roundtrip() {
use korp_collectors::rest::RestCollector;
// The AISStream preset declares `/MetaData/time_utc` as its incremental (CDC)
// bounds field — an ISO-8601 string, so lexical order IS chronological order.
//
// This used to ride on the old `ais-positions` preset. That preset became
// `digitraffic-ais` on 2026-09-03, and Digitraffic's per-record time is an epoch
// INT (`timestampExternal`), which is not a lexical CDC key — so it now declares
// NONE, and the round-trip moved to the AIS preset that still has one rather than
// pretending an int compares like a date.
let payload = serde_json::json!({ "messages": [
{ "MetaData": { "MMSI": 1, "time_utc": "2026-06-30T10:00:00Z" } },
{ "MetaData": { "MMSI": 2, "time_utc": "2026-06-30T12:00:00Z" } },
{ "MetaData": { "MMSI": 3, "time_utc": "2026-06-30T14:00:00Z" } }
]})
.to_string()
.into_bytes();
let (url, _hits, handle) = spawn_http(vec![(200, payload)]);
let cfg = CollectorCfg::new(url, "aisstream_positions");
let sink = BatchSink::memory();
let collector = RestCollector::aisstream();
assert_eq!(
collector.bounds_field(),
Some("/MetaData/time_utc"),
"AISStream wires the CDC key"
);
// …and the Digitraffic preset honestly declares none, rather than an int key that
// a lexical comparison would silently mis-order.
assert_eq!(
RestCollector::digitraffic_ais().bounds_field(),
None,
"an epoch-int timestamp is not a lexical CDC key"
);
// Collect only positions at/after 11:00 — the earliest is filtered out.
let bounds = Bounds {
since: Some("2026-06-30T11:00:00Z".into()),
until: None,
..Default::default()
};
let stats = collector
.collect(&cfg, bounds, &sink)
.await
.expect("bounded collect");
assert_eq!(stats.rows, 2, "the 10:00 position is out of bounds");
assert!(stats.reason.is_none());
assert_eq!(sink.rows(), 2, "only the in-bounds rows landed in the sink");
// checkpoint() reports the newest processed timestamp — the resume mark.
let cp = collector
.checkpoint()
.await
.expect("checkpoint ok")
.expect("a mark exists after a bounded collect over dated records");
assert_eq!(
cp.mark, "2026-06-30T14:00:00Z",
"high-watermark of the processed slice"
);
handle.join().unwrap();
emit_for(
"korp-collectors/rest",
"bounds_and_checkpoint_roundtrip",
true,
"bounded collect kept 2/3, checkpoint mark = newest processed ts",
);
}