nordisk/korp-collectors

tests/business_keys_name_real_columns.rs

//! **O3 — a business key must name columns the collector actually emits.**
//!
//! `BusinessKey::missing_from` has existed since the history seam was declared and
//! nothing called it across the registry, so a key was a list of hopeful strings: a
//! column name that no batch carries cannot identify anything, and the failure shows up
//! only as rows that never match — an Upsert that silently degrades to an append,
//! forever, with nothing red anywhere.
//!
//! Since O1 wired `Upsert` into the sink, that stopped being theoretical: the key
//! reaches a real `EqualityDelete` / `MERGE INTO`. So this sweeps the WHOLE registry and
//! asks each collector's own `schema()` whether its own key exists.

use korp_collectors::{CollectorCfg, registry};

/// Every collector's declared key, checked against its own declared schema.
///
/// A schema a collector cannot produce without config (a copybook, a cube selection) is
/// SKIPPED and named in the report — a skip that is counted is a hole an operator can
/// see, a skip that is silent is the same blindness this test exists to remove.
#[test]
fn every_declared_business_key_names_columns_the_collector_emits() {
    let mut checked = 0usize;
    let mut skipped: Vec<String> = Vec::new();
    let mut append_only: Vec<String> = Vec::new();
    let mut broken: Vec<String> = Vec::new();

    for c in registry() {
        let name = c.meta().name;
        let cfg0 = CollectorCfg::new("https://example.invalid/", "t");
        let key = c.business_key_for(&cfg0);
        if key.columns.is_empty() {
            append_only.push(name);
            continue;
        }
        let schema = match c.schema(&cfg0) {
            Ok(s) => s,
            Err(e) => {
                skipped.push(format!("{name} (schema needs config: {e})"));
                continue;
            }
        };
        checked += 1;
        let missing = key.missing_from(&schema);
        if !missing.is_empty() {
            let have: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
            broken.push(format!(
                "\n  {name}: key {:?} names {missing:?} which the schema does not have.\n     \
                 schema: {have:?}\n     why: {}",
                key.columns, key.why
            ));
        }
    }

    // The sweep must actually have swept. A registry that answered `Err` for every
    // schema, or one that had quietly declared every collector append-only, would
    // otherwise report a triumphant zero. Both halves are asserted: enough collectors
    // were VISITED, and enough of them still carry a real key for the merge path (lane
    // EC O1) to have anything to merge on.
    let visited = checked + append_only.len() + skipped.len();
    assert!(
        visited >= 50,
        "only {visited} collectors were visited — the registry is not being swept"
    );
    assert!(
        checked >= 20,
        "only {checked} of {visited} collectors carry a business key at all — \
         append-only is the safe answer but it is not a free one, and a registry that \
         merged nothing would pass a missing-column check trivially. \
         append-only: {append_only:?}, skipped: {skipped:?}"
    );
    assert!(
        broken.is_empty(),
        "{} of {checked} collectors declare a business key naming a column they do not \
         emit — an Upsert on such a key matches nothing and silently degrades to an \
         append:{}\n\n(append-only, correctly declared: {append_only:?})",
        broken.len(),
        broken.join("")
    );
}