//! Shared HTTP core for every collector: one `reqwest` client with **retry**
//! (exponential backoff on 429 / 5xx / transport errors), **rate-limiting**
//! (governor — honour SCB ≤30/10s, ENTSO-E ≤400/min, Trafiklab caps), and a
//! **pluggable [`Auth`]** so the same client serves no-auth, query-key
//! (ENTSO-E `securityToken`, Trafiklab key), header-key (Trafikverket) and
//! bearer-token sources. Secrets come from env (see
//! [`AuthConfig`](super::AuthConfig)) — never hardcoded.
//!
//! The live fetch is only exercised by `#[ignore]` network tests; the offline
//! fixture tests drive the per-collector *parsers* directly, so CI never hits
//! the network.
use std::time::Duration;
use anyhow::{Result, anyhow};
use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
/// Pluggable authentication — applied to every outgoing request.
pub trait Auth: Send + Sync {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder;
}
/// No credential (most Swedish open data).
pub struct NoAuth;
impl Auth for NoAuth {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
req
}
}
/// `?{param}={value}` query credential — ENTSO-E `securityToken`, Trafiklab key.
pub struct QueryKey {
param: String,
value: String,
}
impl QueryKey {
pub fn new(param: impl Into<String>, value: impl Into<String>) -> Self {
Self {
param: param.into(),
value: value.into(),
}
}
}
impl Auth for QueryKey {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
req.query(&[(self.param.as_str(), self.value.as_str())])
}
}
/// `{header}: {value}` credential — Trafikverket-style key.
pub struct HeaderKey {
header: String,
value: String,
}
impl HeaderKey {
pub fn new(header: impl Into<String>, value: impl Into<String>) -> Self {
Self {
header: header.into(),
value: value.into(),
}
}
}
impl Auth for HeaderKey {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
req.header(self.header.as_str(), self.value.as_str())
}
}
/// **Constant, non-secret request headers, composed over another [`Auth`].**
///
/// Some open APIs mandate an *identification* header that is not a credential:
/// Fintraffic's Digitraffic gateway requires `Digitraffic-User: <anything that
/// identifies the client>` and answers **HTTP 502** without it (measured live
/// 2026-09-03). That is not a secret — there is nothing to store in a vault, no
/// `SecretReq` to declare, and declaring one would make a keyless collector read as
/// `NeedsConfig` forever.
///
/// So it rides here instead: the collector names the headers it must send, this
/// wrapper applies them and then delegates to whatever real auth the config carries
/// (`NoAuth` for a keyless source, a bearer token for a credentialled one). One
/// writer, and a keyed source that *also* needs an identification header composes
/// without a second code path.
pub struct WithHeaders {
headers: Vec<(String, String)>,
inner: Box<dyn Auth>,
}
impl WithHeaders {
/// Wrap `inner`, adding each `(name, value)` to every request.
pub fn new(inner: Box<dyn Auth>, headers: Vec<(String, String)>) -> Self {
Self { headers, inner }
}
}
impl Auth for WithHeaders {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
let mut req = req;
for (name, value) in &self.headers {
req = req.header(name.as_str(), value.as_str());
}
self.inner.apply(req)
}
}
/// `Authorization: Bearer {token}`.
pub struct Bearer {
token: String,
}
impl Bearer {
pub fn new(token: impl Into<String>) -> Self {
Self {
token: token.into(),
}
}
}
impl Auth for Bearer {
fn apply(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
req.bearer_auth(&self.token)
}
}
/// Fetch an OAuth2 **client-credentials** bearer token. Form-POSTs
/// `grant_type=client_credentials` (+ `client_id`/`client_secret`) to `token_url`
/// and returns the `access_token`. OpenSky migrated its live API to a Keycloak
/// OAuth2 client-credentials flow in 2025 (the old anonymous / basic-auth path
/// now 401s for authenticated quotas), so the flights collector obtains a token
/// this way when `OPENSKY_CLIENT_ID`/`OPENSKY_CLIENT_SECRET` are configured. The
/// token then rides as `Authorization: Bearer` via [`Bearer`]. Never blocks past
/// its own 30s timeout; the secret comes from env (see [`AuthConfig`]).
///
/// [`AuthConfig`]: super::AuthConfig
pub async fn oauth2_client_credentials_token(
token_url: &str,
client_id: &str,
client_secret: &str,
) -> Result<String> {
let client = tls_roots(reqwest::Client::builder())
.user_agent("korp-collectors/0.1 (+https://git.gunnar.rs/nordisk)")
.timeout(Duration::from_secs(30))
.build()?;
let params = [
("grant_type", "client_credentials"),
("client_id", client_id),
("client_secret", client_secret),
];
let resp = client.post(token_url).form(¶ms).send().await?;
let status = resp.status();
let body = resp.text().await?;
if !status.is_success() {
return Err(anyhow!("OAuth2 token endpoint {token_url} → HTTP {status}"));
}
let json: serde_json::Value = serde_json::from_str(&body)
.map_err(|e| anyhow!("OAuth2 token response from {token_url} is not JSON: {e}"))?;
json.get("access_token")
.and_then(|v| v.as_str())
.map(str::to_string)
.ok_or_else(|| anyhow!("OAuth2 token response from {token_url} missing `access_token`"))
}
/// **Fetch a document the way its freshness plan says to** — the ONE writer of the
/// conditional-fetch path, so no collector rolls its own `If-None-Match` handling.
///
/// Returns `(body, checkpoint)`: `body` is `None` exactly when the source answered
/// `304`, and the checkpoint then carries `unchanged = true` and `bytes = 0`. On a
/// `200` the checkpoint carries the validators to send next time and the bytes actually
/// transferred.
///
/// A non-[`Conditional`](crate::freshness::Plan::Conditional) plan simply sends no
/// validators — the same call, an unconditional GET, and the validators are still
/// recorded so the NEXT run can be conditional. That is what makes a first run able to
/// become a `304` on the second without the collector knowing anything about it.
pub async fn fetch_planned(
http: &HttpCore,
url: &str,
auth: &dyn Auth,
plan: &crate::freshness::Plan,
) -> Result<(Option<String>, crate::Checkpoint)> {
let (inm, ims) = match plan {
crate::freshness::Plan::Conditional {
if_none_match,
if_modified_since,
} => (if_none_match.clone(), if_modified_since.clone()),
_ => (None, None),
};
let fetched = http
.get_conditional(url, auth, inm.as_deref(), ims.as_deref())
.await?;
let bytes = fetched.transferred();
if fetched.unchanged() {
return Ok((None, crate::Checkpoint::default().unchanged(bytes)));
}
let (etag, last_modified) = fetched.validators();
Ok((
fetched.text(),
crate::Checkpoint::default()
.with_validators(etag, last_modified)
.with_bytes(bytes),
))
}
/// **Read a source that is either a URL or a local path** — the ONE writer of that
/// choice.
///
/// It was four byte-identical copies (`csv`, `fixedwidth`, `mainframe`, `gtfs`), which
/// is exactly the shape that lets a seam land in three places and be silently missing
/// from the fourth: each copy built its own [`HttpCore`] and so ignored
/// [`CollectorCfg::canned`](crate::CollectorCfg::canned). Here it is one function that
/// takes the cfg, so a fixture reaches every flat-file collector by construction.
pub async fn read_source(cfg: &crate::CollectorCfg, src: &str) -> Result<Vec<u8>> {
if src.starts_with("http://") || src.starts_with("https://") {
cfg.http(10).get_bytes(src, &NoAuth).await
} else {
Ok(std::fs::read(std::path::Path::new(src))?)
}
}
/// Minimal RFC-3986 percent-encoding for query values (avoids a `url` dep).
pub fn percent_encode(s: &str) -> String {
let mut out = String::with_capacity(s.len() * 3);
for b in s.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(b as char)
}
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
/// **What a conditional GET came back with.** Either the source proved unchanged and
/// nothing crossed the wire, or a body arrived with the validators for next time.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Fetched {
/// `304 Not Modified` — no body. The run lands nothing and says why; RAGNARÖK reads
/// it as an explained empty, never as a failure.
NotModified,
/// A body, plus whatever validators the source served with it.
Body {
/// The response body.
bytes: Vec<u8>,
/// The `ETag`, for the next `If-None-Match`.
etag: Option<String>,
/// The `Last-Modified`, for the next `If-Modified-Since`.
last_modified: Option<String>,
},
}
impl Fetched {
/// **Bytes actually transferred** — `0` for a `304`, which is the whole point of
/// asking conditionally and the number a report must be able to show.
pub fn transferred(&self) -> u64 {
match self {
Fetched::NotModified => 0,
Fetched::Body { bytes, .. } => bytes.len() as u64,
}
}
/// Did the source prove unchanged?
pub fn unchanged(&self) -> bool {
matches!(self, Fetched::NotModified)
}
/// The body as text, or `None` when there was no body.
pub fn text(&self) -> Option<String> {
match self {
Fetched::NotModified => None,
Fetched::Body { bytes, .. } => Some(String::from_utf8_lossy(bytes).into_owned()),
}
}
/// The validators to carry into the next run.
pub fn validators(&self) -> (Option<String>, Option<String>) {
match self {
Fetched::NotModified => (None, None),
Fetched::Body {
etag,
last_modified,
..
} => (etag.clone(), last_modified.clone()),
}
}
}
/// A response header as an owned `String`, when present and valid UTF-8.
fn header_str(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.map(str::to_string)
}
/// **A canned response table — the offline seam for `collect()`.**
///
/// `Fetcher::from_map` gave the *daemon* leaf a hermetic fixture, but a
/// [`Collector::collect`](crate::Collector::collect) builds its own [`HttpCore`]
/// internally, so routing a daemon tick through its registry twin discarded the
/// fixture entirely and went to the live source (measured 2026-09-04: a hermetic
/// polisen daemon test that asserted 2 canned events landed **500** — the live page
/// size). This is that missing seam, and it rides on [`CollectorCfg`](crate::CollectorCfg)
/// so it reaches every collector through config rather than through 22 twins.
///
/// **Matching is exact url first, then the LONGEST key that is a prefix of the url.**
/// One rule, not a fallback chain. It exists because a query-window plan appends
/// `?DateTime=<mark>` to the base url at run time, so the exact url a tick asks for is
/// not knowable when the fixture is written; longest-prefix lets a more specific
/// fixture still beat a broader one on the same host. An un-mapped url is a loud
/// `Err` naming the url — never a silent empty.
#[derive(Debug, Clone, Default)]
pub struct Canned {
table: std::sync::Arc<std::collections::BTreeMap<String, Vec<u8>>>,
}
impl Canned {
/// Build from `url → bytes`.
pub fn new(table: std::collections::BTreeMap<String, Vec<u8>>) -> Self {
Self {
table: std::sync::Arc::new(table),
}
}
/// Exact url, else the longest key that prefixes it.
pub fn get(&self, url: &str) -> Option<&[u8]> {
if let Some(v) = self.table.get(url) {
return Some(v.as_slice());
}
self.table
.iter()
.filter(|(k, _)| url.starts_with(k.as_str()))
.max_by_key(|(k, _)| k.len())
.map(|(_, v)| v.as_slice())
}
fn resolve(&self, url: &str) -> Result<Vec<u8>> {
self.get(url).map(<[u8]>::to_vec).ok_or_else(|| {
anyhow!(
"canned HTTP core has no fixture for {url} (keys: {})",
self.table.keys().cloned().collect::<Vec<_>>().join(", ")
)
})
}
}
impl FromIterator<(String, Vec<u8>)> for Canned {
fn from_iter<I: IntoIterator<Item = (String, Vec<u8>)>>(iter: I) -> Self {
Self::new(iter.into_iter().collect())
}
}
/// The shared HTTP client — retry + rate-limit live here so every collector
/// inherits them.
pub struct HttpCore {
client: reqwest::Client,
limiter: Option<DefaultDirectRateLimiter>,
max_retries: u32,
/// When set, every verb reads from this table and no socket is opened.
pub(crate) canned: Option<Canned>,
}
impl Default for HttpCore {
fn default() -> Self {
Self::new(10)
}
}
impl HttpCore {
/// `max_per_sec` requests/second (best-effort, smoothing burst load).
pub fn new(max_per_sec: u32) -> Self {
let client = tls_roots(reqwest::Client::builder())
.user_agent("korp-collectors/0.1 (+https://git.gunnar.rs/nordisk)")
.timeout(Duration::from_secs(60))
.build()
.expect("reqwest client");
let limiter = std::num::NonZeroU32::new(max_per_sec)
.map(|q| RateLimiter::direct(Quota::per_second(q)));
Self {
client,
limiter,
max_retries: 4,
canned: None,
}
}
pub fn with_max_retries(mut self, n: u32) -> Self {
self.max_retries = n;
self
}
/// **Serve every request from `canned` instead of the network.** `None` leaves the
/// core live, so the shipped path is untouched. See [`Canned`] for why this exists
/// and how a url is matched.
pub fn with_canned(mut self, canned: Option<Canned>) -> Self {
self.canned = canned;
self
}
async fn throttle(&self) {
if let Some(l) = &self.limiter {
l.until_ready().await;
}
}
/// GET `url` (auth applied) → body text, with retry/rate-limit.
pub async fn get_text(&self, url: &str, auth: &dyn Auth) -> Result<String> {
let bytes = self.send(reqwest::Method::GET, url, auth, None).await?;
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
/// **A conditional GET** — the fetch half of
/// [`Freshness::MutableDocument`](crate::freshness::Freshness::MutableDocument).
///
/// Sends whatever validators the [`Plan`](crate::freshness::Plan) holds and
/// returns [`Fetched::NotModified`] on a `304`, which transfers no body at all.
/// Both live keyless sources on this box honour it (USGS `all_hour.geojson` and
/// SMHI metobs answered a real `304` to `If-Modified-Since` when this was written);
/// polisen serves neither validator, which is why polisen is a query window and not
/// a mutable document.
///
/// `304` is treated as a terminal SUCCESS, not as a retryable failure — the retry
/// loop's "not success ⇒ maybe retry" rule would otherwise turn every unchanged
/// source into four extra requests and then an error.
pub async fn get_conditional(
&self,
url: &str,
auth: &dyn Auth,
if_none_match: Option<&str>,
if_modified_since: Option<&str>,
) -> Result<Fetched> {
// A fixture answers before any socket is opened, and never claims `304`: it
// carries no validators, so the run stays honest and asks unconditionally again.
if let Some(c) = &self.canned {
return Ok(Fetched::Body {
bytes: c.resolve(url)?,
etag: None,
last_modified: None,
});
}
let mut attempt = 0u32;
loop {
self.throttle().await;
let mut req = self.client.request(reqwest::Method::GET, url);
if let Some(v) = if_none_match {
req = req.header("if-none-match", v);
}
if let Some(v) = if_modified_since {
req = req.header("if-modified-since", v);
}
req = auth.apply(req);
match req.send().await {
Ok(resp) => {
let status = resp.status();
if status.as_u16() == 304 {
return Ok(Fetched::NotModified);
}
if status.is_success() {
let etag = header_str(resp.headers(), "etag");
let last_modified = header_str(resp.headers(), "last-modified");
let body = resp.bytes().await?.to_vec();
return Ok(Fetched::Body {
bytes: body,
etag,
last_modified,
});
}
let retryable = status.as_u16() == 429 || status.is_server_error();
if !retryable || attempt >= self.max_retries {
return Err(anyhow!("GET {url} → HTTP {status}"));
}
}
Err(e) => {
if attempt >= self.max_retries {
return Err(anyhow!("GET {url} → {e}"));
}
}
}
tokio::time::sleep(Duration::from_millis(200u64 << attempt.min(5))).await;
attempt += 1;
}
}
/// GET `url` (auth applied) → raw bytes (bulk files / protobuf).
pub async fn get_bytes(&self, url: &str, auth: &dyn Auth) -> Result<Vec<u8>> {
self.send(reqwest::Method::GET, url, auth, None).await
}
/// POST `body` (with `content_type`) to `url` → body text. PxWeb selection /
/// Trafikverket XML query go through here.
pub async fn post_text(
&self,
url: &str,
body: String,
content_type: &str,
auth: &dyn Auth,
) -> Result<String> {
let bytes = self
.send(
reqwest::Method::POST,
url,
auth,
Some((body, content_type.to_string())),
)
.await?;
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
async fn send(
&self,
method: reqwest::Method,
url: &str,
auth: &dyn Auth,
body: Option<(String, String)>,
) -> Result<Vec<u8>> {
// Every verb reads the SAME table, so a POST-body source (Trafikverket's
// `<REQUEST>`, Bolagsverket's `/organisationer`) is drivable offline too.
if let Some(c) = &self.canned {
return c.resolve(url);
}
let mut attempt = 0u32;
loop {
self.throttle().await;
let mut req = self.client.request(method.clone(), url);
if let Some((b, ct)) = &body {
req = req.header("content-type", ct.as_str()).body(b.clone());
}
req = auth.apply(req);
match req.send().await {
Ok(resp) => {
let status = resp.status();
if status.is_success() {
return Ok(resp.bytes().await?.to_vec());
}
// 429 + 5xx are retryable; 4xx (except 429) is fatal.
let retryable = status.as_u16() == 429 || status.is_server_error();
if !retryable || attempt >= self.max_retries {
return Err(anyhow!("{method} {url} → HTTP {status}"));
}
}
Err(e) => {
if attempt >= self.max_retries {
return Err(anyhow!("{method} {url} → {e}"));
}
}
}
// exponential backoff: 200ms, 400, 800, 1600, …
let backoff = Duration::from_millis(200u64 << attempt.min(5));
tokio::time::sleep(backoff).await;
attempt += 1;
}
}
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}
#[cfg(test)]
mod tests {
use super::*;
// Build `req` and materialise it into a `reqwest::Request` so we can inspect
// the URL query + headers the `Auth` impl injected.
fn built(auth: &dyn Auth) -> reqwest::Request {
let req = super::tls_roots(reqwest::Client::builder()).build().expect("reqwest client").get("http://example.test/data");
auth.apply(req).build().expect("build request")
}
#[test]
fn percent_encode_passes_unreserved_and_encodes_reserved() {
// RFC-3986 unreserved set passes through byte-for-byte.
let unreserved = "ABCabc019-_.~";
assert_eq!(percent_encode(unreserved), unreserved);
// reserved / delimiters / braces / '=' encode to their %XX byte.
assert_eq!(percent_encode(" "), "%20");
assert_eq!(percent_encode("?"), "%3F");
assert_eq!(percent_encode("&"), "%26");
assert_eq!(percent_encode("="), "%3D");
assert_eq!(percent_encode("{"), "%7B");
assert_eq!(percent_encode("}"), "%7D");
// a full SPARQL-ish fragment round-trips deterministically.
assert_eq!(percent_encode("a b&c"), "a%20b%26c");
// non-ASCII 'ä' (UTF-8 0xC3 0xA4) → two percent-escaped bytes.
assert_eq!(percent_encode("ä"), "%C3%A4");
}
#[test]
fn noauth_leaves_request_untouched() {
let r = built(&NoAuth);
assert_eq!(r.url().query(), None, "NoAuth adds no query");
assert!(r.headers().get(reqwest::header::AUTHORIZATION).is_none());
}
#[test]
fn query_key_appends_query_param() {
let r = built(&QueryKey::new("securityToken", "SECRET"));
assert_eq!(r.url().query(), Some("securityToken=SECRET"));
}
#[test]
fn header_key_sets_header() {
let r = built(&HeaderKey::new("X-Api-Key", "K123"));
assert_eq!(r.headers().get("X-Api-Key").unwrap(), "K123");
assert_eq!(r.url().query(), None, "header auth adds no query param");
}
#[test]
fn bearer_sets_authorization_header() {
let r = built(&Bearer::new("tok"));
assert_eq!(
r.headers().get(reqwest::header::AUTHORIZATION).unwrap(),
"Bearer tok"
);
}
/// **RED-when-broken: `WithHeaders` sends the constant header AND keeps the
/// wrapped credential.** Both halves are asserted on the SAME request, because a
/// wrapper that dropped the inner auth would still pass a header-only check, and
/// one that dropped the header would still pass a bearer-only check.
#[test]
fn with_headers_adds_constants_and_preserves_the_inner_auth() {
let a = WithHeaders::new(
Box::new(Bearer::new("tok")),
vec![
("Digitraffic-User".to_string(), "korp-collectors/0.1".to_string()),
("X-Trace".to_string(), "abc".to_string()),
],
);
let r = built(&a);
assert_eq!(r.headers().get("Digitraffic-User").unwrap(), "korp-collectors/0.1");
assert_eq!(r.headers().get("X-Trace").unwrap(), "abc");
assert_eq!(
r.headers().get(reqwest::header::AUTHORIZATION).unwrap(),
"Bearer tok",
"the wrapped credential still rides"
);
}
/// Over `NoAuth` (the keyless case Digitraffic actually is) the header is the
/// ONLY thing added — no invented Authorization, no query param.
#[test]
fn with_headers_over_noauth_adds_only_the_header() {
let a = WithHeaders::new(
Box::new(NoAuth),
vec![("Digitraffic-User".to_string(), "korp-collectors/0.1".to_string())],
);
let r = built(&a);
assert_eq!(r.headers().get("Digitraffic-User").unwrap(), "korp-collectors/0.1");
assert!(r.headers().get(reqwest::header::AUTHORIZATION).is_none());
assert_eq!(r.url().query(), None);
}
#[test]
fn with_max_retries_overrides_the_default() {
// The builder is a live knob: default is 4, with_max_retries changes it.
assert_eq!(HttpCore::default().max_retries, 4);
assert_eq!(HttpCore::new(0).with_max_retries(0).max_retries, 0);
assert_eq!(HttpCore::new(0).with_max_retries(7).max_retries, 7);
}
}
#[cfg(test)]
mod canned_tests {
use super::*;
fn map(pairs: &[(&str, &str)]) -> Canned {
Canned::new(
pairs
.iter()
.map(|(k, v)| ((*k).to_string(), v.as_bytes().to_vec()))
.collect(),
)
}
/// **The seam O2c is blocked on.** A `HttpCore` carrying a canned table serves the
/// canned bytes for an EXACT url and never opens a socket — the address is
/// deliberately one nothing can answer (`127.0.0.1:1` is closed on every box), so a
/// core that fell through to the network fails instead of passing quietly.
#[tokio::test]
async fn a_canned_core_serves_the_body_and_never_reaches_the_network() {
let http = HttpCore::new(0).with_canned(Some(map(&[(
"http://127.0.0.1:1/events",
"[{\"id\":1}]",
)])));
let body = http
.get_text("http://127.0.0.1:1/events", &NoAuth)
.await
.expect("canned body");
assert_eq!(body, "[{\"id\":1}]");
}
/// A query-window plan appends `?DateTime=…` to the base url, so the url a tick
/// asks for is NOT knowable when the fixture is written. Longest-prefix is what
/// makes a fixture usable for a parameterised source; it is one rule, not a
/// fallback chain, and the LONGEST key wins so a more specific fixture beats a
/// broader one on the same host.
#[tokio::test]
async fn a_canned_core_matches_the_longest_prefix_for_a_parameterised_url() {
let http = HttpCore::new(0).with_canned(Some(map(&[
("http://127.0.0.1:1/api", "broad"),
("http://127.0.0.1:1/api/events", "specific"),
])));
assert_eq!(
http.get_text("http://127.0.0.1:1/api/events?DateTime=2026-09-04", &NoAuth)
.await
.expect("canned body"),
"specific",
"the longest matching key wins, not the first"
);
assert_eq!(
http.get_text("http://127.0.0.1:1/api/other?x=1", &NoAuth)
.await
.expect("canned body"),
"broad"
);
}
/// An un-mapped url is a LOUD error, never a silent empty — the same contract
/// `Fetcher::from_map` already keeps, and the reason a mis-wired offline daemon
/// test fails instead of asserting nothing.
#[tokio::test]
async fn an_unmapped_url_errors_by_name() {
let http = HttpCore::new(0).with_canned(Some(map(&[("http://a.test/x", "hi")])));
let e = http
.get_text("http://b.test/y", &NoAuth)
.await
.expect_err("un-mapped url must error");
let msg = format!("{e:#}");
assert!(msg.contains("http://b.test/y"), "names the url asked for: {msg}");
assert!(msg.contains("canned"), "says it was the canned core: {msg}");
}
/// Every verb goes through the same table — a POST-body source (Trafikverket's
/// `<REQUEST>`, Bolagsverket's `/organisationer`) is exercised offline too, and a
/// conditional GET answers with the body and NO validators (a fixture has no
/// `ETag`, so the next run stays honest and asks unconditionally again).
#[tokio::test]
async fn post_and_conditional_get_read_the_same_table() {
let http = HttpCore::new(0).with_canned(Some(map(&[("http://a.test/q", "rows")])));
assert_eq!(
http.post_text("http://a.test/q", "<REQUEST/>".into(), "application/xml", &NoAuth)
.await
.expect("canned post"),
"rows"
);
let f = http
.get_conditional("http://a.test/q", &NoAuth, Some("\"etag\""), None)
.await
.expect("canned conditional");
assert!(!f.unchanged(), "a fixture never claims 304");
assert_eq!(f.text().as_deref(), Some("rows"));
assert_eq!(f.validators(), (None, None), "a fixture carries no validators");
assert_eq!(f.transferred(), 4);
}
/// **Red-when-broken for the wiring, not the table.** `CollectorCfg::http` is the
/// ONE constructor every collector's `collect()` uses, so the canned table reaches
/// all of them through config — the thing that was missing when the incremental
/// daemon tick went live against polisen.se and landed 500 rows.
#[tokio::test]
async fn the_cfg_carries_the_canned_table_into_the_core() {
let cfg = crate::CollectorCfg::new("http://127.0.0.1:1/api", "t")
.with_canned(map(&[("http://127.0.0.1:1/api", "from-cfg")]));
assert_eq!(
cfg.http(10).get_text("http://127.0.0.1:1/api", &NoAuth).await.unwrap(),
"from-cfg"
);
// …and a cfg WITHOUT a table builds a live core (no canned table to read).
let live = crate::CollectorCfg::new("http://127.0.0.1:1/api", "t");
assert!(live.http(10).canned.is_none(), "no fixture ⇒ the real network core");
}
}