tests/collector_daemons_register.rs
//! **The batch-6 daemons go RED→GREEN in the live register** (design §8, LAW 2).
//!
//! `polisen-events`, `smhi-weather` and `trafikverket-trainpos` are new [`Collector`]
//! daemons. The korp Collectors panel does not read them in-process: it opens the
//! register file with a [`RegisterReader`] (a DIFFERENT process, `O_RDONLY`, no lock) and
//! renders whatever row a supervisor wrote. So the property that actually matters is that
//! a daemon's phase transition is *visible through that reader* — and, red-first, that a
//! daemon which never wrote its phase reads as **absent/stale**, never as a silent green.
//!
//! This proves the whole arc for each of the three, hermetically (no korp, no network —
//! the fetch is a canned [`Fetcher::from_map`] fixture):
//!
//! 1. **absent** — before any publish, the reader has no row for the collector;
//! 2. **RED** — a bare (never-reconciled) daemon published Offline reads back not-online
//! and not-healthy;
//! 3. **GREEN** — after one reconcile to Online, the re-published row reads back Online +
//! healthy, carrying the collector's own facet (table) so the panel can render it.
//!
//! The out-of-process reader ITSELF is proven across two real pids in
//! `register_across_processes.rs`; here the reader is opened in-process (it still takes no
//! lock and reads the on-disk file), so the assertions are about the row content the
//! panel would see, for these three specific daemons.
#![cfg(feature = "daemon")]
use std::collections::BTreeMap;
use korp_collectors::daemon::{Collector, CollectorCtx, Fetcher, Phase};
use korp_collectors::iceberg_daemons::{PolisenDaemon, SmhiWeatherDaemon, TrafikverketDaemon};
use korp_collectors::register::{Register, RegisterReader, Writer};
/// A hermetic ctx whose fetch seam serves `bytes` for `url` (no network).
fn fixture_ctx(name: &str, url: &str, bytes: &[u8]) -> CollectorCtx {
let mut map = BTreeMap::new();
map.insert(url.to_string(), bytes.to_vec());
let mut ctx = CollectorCtx::in_memory(name, std::env::temp_dir(), serde_json::json!({}));
ctx.fetch = Fetcher::from_map(map);
ctx
}
/// Assert the absent → RED → GREEN arc for one daemon, read back through the reader.
/// `bare` is a fresh (never-reconciled) daemon; `online` is a daemon the caller has
/// already reconciled to Online. A [`RegisterReader`] is a point-in-time snapshot (it
/// opens `O_RDONLY` and never re-reads), exactly as a fresh `korp-collectors status`
/// process would — so each stage opens a NEW reader on `data_dir`, which is also proof
/// the row is on disk and not merely in the writer's memory.
fn assert_red_then_green(
reg: &Register,
data_dir: &std::path::Path,
name: &str,
table: &str,
bare: &dyn Collector,
online: &dyn Collector,
) {
let read = |name: &str| {
RegisterReader::open(data_dir)
.expect("open reader")
.expect("register exists")
.get(name)
.expect("read register")
};
// 1. ABSENT — nobody has published this collector yet.
assert!(
read(name).is_none(),
"{name}: no row before any publish (never wrote its phase ⇒ absent, not a fake ok)"
);
// 2. RED — the bare daemon, never reconciled, published in its Offline state.
assert_ne!(bare.phase(), Phase::Online, "{name}: bare daemon is not Online");
reg.publish(bare).expect("publish bare");
let view = read(name).unwrap_or_else(|| panic!("{name}: row present after publish"));
assert_ne!(
view.row.phase,
Phase::Online,
"{name}: RED — a never-run daemon does not read Online"
);
assert!(
!view.row.healthy,
"{name}: RED — a never-run daemon is not healthy"
);
// 3. GREEN — the reconciled daemon re-published; the reader flips to Online.
assert_eq!(online.phase(), Phase::Online, "{name}: reconciled to Online");
reg.publish(online).expect("publish online");
let view = read(name).unwrap_or_else(|| panic!("{name}: row present"));
assert_eq!(
view.row.phase,
Phase::Online,
"{name}: GREEN — the reader now reads Online"
);
assert!(view.row.healthy, "{name}: GREEN — Online is healthy");
assert!(view.row.enabled, "{name}: the enabled flag reached the row");
assert_eq!(
view.row.facet["table"], table,
"{name}: the row carries the collector's own facet (table) for the panel"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch6_daemons_go_red_then_green_in_the_register() {
let dir = tempfile::tempdir().expect("tempdir");
let writer = Writer::this_process("test-nursery", Some("collector-test.service".into()));
let reg = Register::open(dir.path(), writer).expect("open register for writing");
// polisen-events
{
let bare = PolisenDaemon::with_config(true, PolisenDaemon::default_config());
let online = PolisenDaemon::with_config(true, PolisenDaemon::default_config());
let ctx = fixture_ctx(
"polisen-events",
&online.url(),
korp_collectors::sample::SYNTHETIC_POLISEN_JSON.as_bytes(),
);
online.reconcile_once(&ctx).await.expect("polisen online");
assert_red_then_green(®, dir.path(), "polisen-events", "polisen_events", &bare, &online);
}
// smhi-weather
{
let bare = SmhiWeatherDaemon::with_config(true, SmhiWeatherDaemon::default_config());
let online = SmhiWeatherDaemon::with_config(true, SmhiWeatherDaemon::default_config());
let ctx = fixture_ctx(
"smhi-weather",
&online.url(),
korp_collectors::sample::SYNTHETIC_SMHI_WEATHER_JSON.as_bytes(),
);
online.reconcile_once(&ctx).await.expect("smhi online");
assert_red_then_green(®, dir.path(), "smhi-weather", "smhi_weather", &bare, &online);
}
// trafikverket-trainpos (a POST-body daemon; key supplied in config, no env, no net)
{
let cfg = serde_json::json!({ "key": "offline-test-key" });
let bare = TrafikverketDaemon::with_config(true, cfg.clone());
let online = TrafikverketDaemon::with_config(true, cfg);
let ctx = fixture_ctx(
"trafikverket-trainpos",
&online.url(),
korp_collectors::sample::SYNTHETIC_TRAFIKVERKET_JSON.as_bytes(),
);
online.reconcile_once(&ctx).await.expect("trafikverket online");
assert_red_then_green(
®,
dir.path(),
"trafikverket-trainpos",
"trafikverket_trainpos",
&bare,
&online,
);
}
}