nordisk/korp-collectors

tests/graph_mappings_agree.rs

//! **Every graph mapping the registry declares, checked as ONE set** (lane EC O7).
//!
//! The sink keys a node on `(label, id property, value)`. A label declared as a vertex
//! by one collector and referenced as an endpoint by another with a different id
//! property therefore makes the sink MERGE a SECOND node for the same entity: the graph
//! silently doubles and every count is wrong. `Organisation` is exactly that shape —
//! DCAT publishes to it, SMHI's stations are owned by it — so this is not hypothetical,
//! it is the one thing the cross-collector move made possible to get wrong.
//!
//! korp had this check as a test over its own hand-written array. Moving the
//! declaration down to the leaf must not lose it, so it moves down with it.

use korp_collectors::graph_mapping::{GraphMapping, key_conflicts};
use korp_collectors::registry;

fn declared() -> Vec<GraphMapping> {
    registry().iter().filter_map(|c| c.graph_mapping()).collect()
}

/// No label is keyed two ways, and every endpoint agrees with its label's key.
#[test]
fn the_registrys_graph_mappings_key_every_label_one_way() {
    let mappings = declared();
    // The check is worthless over an empty set, and an empty set would also mean the
    // registry stopped declaring graphs at all — which is the other thing worth
    // catching.
    assert!(
        mappings.len() >= 5,
        "only {} collectors declare a graph mapping — the registry fold is not finding \
         them",
        mappings.len()
    );
    let conflicts = key_conflicts(&mappings);
    assert!(
        conflicts.is_empty(),
        "{} label/endpoint key conflicts across the registry's graph mappings — each \
         one is a second node per entity:\n  {}",
        conflicts.len(),
        conflicts.join("\n  ")
    );
}

/// **A plan name is an identity, so two collectors must not claim one**, and a mapping
/// must name a table, a limitation and at least one vertex — a "mapping" with no vertex
/// produces nothing and would sit in the pipeline roster as a plan that runs to an empty
/// graph.
#[test]
fn every_declared_mapping_is_complete_and_uniquely_named() {
    let mut seen: std::collections::BTreeMap<String, String> = Default::default();
    for c in registry() {
        let name = c.meta().name;
        let Some(m) = c.graph_mapping() else { continue };
        assert!(!m.plan.is_empty(), "{name}: a mapping with no plan name");
        assert!(!m.table.is_empty(), "{name}: plan `{}` names no table", m.plan);
        assert!(
            !m.limitation.trim().is_empty(),
            "{name}: plan `{}` states no limitation — every mapping has one, and the \
             unwritten ones are what an operator discovers as a wrong number",
            m.plan
        );
        assert!(
            !m.vertices.is_empty(),
            "{name}: plan `{}` declares no vertex, so it would run to an empty graph",
            m.plan
        );
        if let Some(prev) = seen.insert(m.plan.clone(), name.clone()) {
            panic!("plan `{}` is claimed by both `{prev}` and `{name}`", m.plan);
        }
    }
}

/// **A derived id must not collide with a column the table already has**, and every
/// column a mapping reads must be one the collector emits, a derived id, or one of the
/// spine columns the join adds.
///
/// This is the graph-side twin of the business-key sweep, and it exists for the same
/// reason: a mapping naming a column that is not there produces nodes with NULL
/// identities, which the sink happily MERGEs into ONE node — a silent collapse, not an
/// error.
#[test]
fn every_column_a_mapping_reads_is_one_the_collector_emits() {
    let cfg = korp_collectors::CollectorCfg::new("https://example.invalid/", "t");
    let mut checked = 0usize;
    let mut bad: Vec<String> = Vec::new();

    for c in registry() {
        let name = c.meta().name;
        let Some(m) = c.graph_mapping() else { continue };
        let Ok(schema) = c.schema(&cfg) else { continue };
        checked += 1;
        let mut have: Vec<String> = schema.fields().iter().map(|f| f.name().clone()).collect();
        // The spine join adds these two, by construction, when it is declared.
        if m.spine.is_some() {
            have.push("municipality_code".into());
            have.push("county_code".into());
        }
        for d in &m.derived_ids {
            assert!(
                !have.iter().any(|h| h == &d.column),
                "{name}: derived id `{}` collides with a column the table already has",
                d.column
            );
            have.push(d.column.clone());
        }
        let mut want: Vec<(&str, String)> = Vec::new();
        for v in &m.vertices {
            want.push(("vertex id", v.id_column.clone()));
            for p in &v.properties {
                want.push(("vertex property", p.clone()));
            }
        }
        for e in &m.edges {
            want.push(("edge src column", e.src.column.clone()));
            want.push(("edge dst column", e.dst.column.clone()));
        }
        for (what, col) in want {
            if !have.iter().any(|h| h == &col) {
                bad.push(format!(
                    "{name} (plan `{}`): {what} `{col}` is not a column it emits — \
                     schema {have:?}",
                    m.plan
                ));
            }
        }
    }

    assert!(checked >= 5, "only {checked} mappings could be checked");
    assert!(
        bad.is_empty(),
        "{} graph-mapping columns do not exist — each one produces NULL-identity nodes \
         the sink collapses into ONE:\n  {}",
        bad.len(),
        bad.join("\n  ")
    );
}