facett-demo/tests/region_picker_connector.rs
//! **Region picker fed by a korp connector** (feature `region-connector`).
//!
//! Proves the 🌍 Region picker can be "fed by a connector fetching data": a
//! `korp_collectors::region::HttpOsmSource` FETCHES a region's `.osm.pbf` from an
//! in-process HTTP server, the picker converts the fetched extract to GeoParquet
//! IN-PROCESS (osm-katana) and loads a NON-EMPTY geometry — then a re-pick is a
//! proven cache hit (a request counter never increments; the connector is never
//! touched a second time). Real inject/assert (facett FC-1/FC-7): we inject a real
//! PBF over a real socket and assert the picker's observable `state_json` +
//! `take_loaded()` — no wall-clock, no "didn't panic".
//!
//! Run:
//! cargo test -p facett-demo --features region-connector
//! cargo test -p facett-demo --features region-connector,testmatrix # feeds the matrix
#![cfg(feature = "region-connector")]
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use facett_demo::region_picker::{RegionPicker, Status};
use korp_collectors::region::HttpOsmSource;
/// The tiny real OSM extract korp-collectors ships (454-byte zstd PBF, 4 nodes / 2
/// ways). Reused straight from the sibling leaf so this test needs no new fixture
/// and asserts REAL OSM bytes (LAW 2), not a synthetic stand-in.
const FIXTURE_PBF: &[u8] =
include_bytes!("../../../korp-collectors/tests/fixtures/testland-latest.osm.pbf");
/// A minimal HTTP/1.1 server on `127.0.0.1:0` that serves `body` for `served`
/// requests then exits, counting hits. Returns `(base_url, hit_counter, handle)`.
/// Sized to the calls a test expects — a cache hit that (wrongly) re-fetched would
/// find the socket closed and error, so the counter is a hard cache-hit proof.
fn spawn_pbf_server(
body: Vec<u8>,
served: usize,
) -> (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 _ in 0..served {
let Ok((mut stream, _)) = listener.accept() else { break };
let mut buf = [0u8; 1024];
let _ = stream.read(&mut buf);
hits_srv.fetch_add(1, Ordering::SeqCst);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\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)
}
/// Pump `poll()` until the pipeline reaches a terminal status (Ready/Error) or a
/// bounded number of polls elapses. The short sleep only PACES the polling of the
/// background worker (same pattern as the preset test) — the assertion is on the
/// observable *status transition*, never on elapsed wall-clock.
fn drive_to_terminal(rp: &mut RegionPicker) {
for _ in 0..4_000 {
rp.poll();
match rp.status() {
Status::Ready | Status::Error(_) => return,
_ => std::thread::sleep(std::time::Duration::from_millis(5)),
}
}
}
#[test]
fn connector_feeds_picker_fetch_then_cache_hit_loads_nonempty() {
// ── isolate the GeoParquet cache dir (used by region_cache_dir_named) ───────
let xdg = std::env::temp_dir().join(format!("facett-conn-xdg-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&xdg);
std::fs::create_dir_all(&xdg).unwrap();
let prev_xdg = std::env::var("XDG_CACHE_HOME").ok();
// SAFETY: this integration test binary runs this single test; no other thread
// reads the environment concurrently while we set it up.
unsafe {
std::env::set_var("XDG_CACHE_HOME", &xdg);
}
// ── the connector: an HttpOsmSource pointed at our in-process server ────────
// The server serves AT MOST ONE request; a cache hit must not need a second.
let (base_url, hits, handle) = spawn_pbf_server(FIXTURE_PBF.to_vec(), 1);
let cache = std::env::temp_dir().join(format!("facett-conn-cache-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&cache);
std::fs::create_dir_all(&cache).unwrap();
let src = HttpOsmSource::new(&base_url, &cache);
let mut rp = RegionPicker::with_source(src);
assert_eq!(rp.connector_name(), Some("http-osm"), "picker reports the connector name");
// ── FIRST pick: the connector FETCHES `testland`, convert+load ─────────────
rp.pick_region("testland");
// The observable surface is set synchronously by the pick (mirrors cache_hit).
let s = rp.state_json();
assert_eq!(s["region"], "testland", "region name flows into state_json");
assert_eq!(s["source"], "http-osm", "state_json.source NAMES the connector");
assert_eq!(s["connector"], "http-osm");
assert_eq!(s["cache_hit"], false, "fresh cache → miss on first pick");
assert_eq!(s["fetched"], true, "first pick invokes the connector (a fetch)");
drive_to_terminal(&mut rp);
let s = rp.state_json();
assert_eq!(s["status"], "ready", "first pick reaches ready (got {})", s["status"]);
assert_eq!(hits.load(Ordering::SeqCst), 1, "exactly one HTTP fetch");
// LAW 2: the fetched extract loads REAL geometry (non-empty ways + points).
let geo = rp.take_loaded().expect("first pick yields loaded geometry");
let ways = geo.ways_2d.len();
let points = geo.points.len();
assert!(ways >= 1, "fetched extract yields ≥1 real way (got {ways})");
assert!(points >= 1, "fetched extract yields ≥1 node point (got {points})");
// ── RE-PICK: GeoParquet cache HIT — connector never touched, no HTTP ────────
rp.pick_region("testland");
let s = rp.state_json();
assert_eq!(s["cache_hit"], true, "GeoParquet now cached → hit on re-pick");
assert_eq!(s["fetched"], false, "cache hit short-circuits before the connector");
drive_to_terminal(&mut rp);
let s = rp.state_json();
assert_eq!(s["status"], "ready", "re-pick loads from cache (got {})", s["status"]);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"re-pick is a cache hit — NO second HTTP request (counter stayed 1)"
);
let geo2 = rp.take_loaded().expect("re-pick yields loaded geometry from cache");
assert!(geo2.ways_2d.len() >= 1, "cache-hit load is still non-empty");
let handle_ok = handle.join().is_ok();
// ── FEED THE NORNIR TEST MATRIX (compiled out without `testmatrix`) ─────────
#[cfg(feature = "testmatrix")]
{
facett::core::testmatrix::emit(
"facett-region-picker",
"connector_fetched",
hits.load(Ordering::SeqCst) == 1 && handle_ok,
&format!("http-osm fetched testland in 1 request (bytes={})", FIXTURE_PBF.len()),
);
facett::core::testmatrix::emit(
"facett-region-picker",
"cache_hit",
hits.load(Ordering::SeqCst) == 1,
"re-pick resolved from GeoParquet cache with no second HTTP request",
);
facett::core::testmatrix::emit(
"facett-region-picker",
"region_loaded_nonempty",
ways >= 1 && points >= 1,
&format!("loaded {ways} ways / {points} points from the fetched extract"),
);
}
let _ = handle_ok;
// ── restore env + clean up ─────────────────────────────────────────────────
unsafe {
match prev_xdg {
Some(v) => std::env::set_var("XDG_CACHE_HOME", v),
None => std::env::remove_var("XDG_CACHE_HOME"),
}
}
let _ = std::fs::remove_dir_all(&xdg);
let _ = std::fs::remove_dir_all(&cache);
}