nordisk/korp-collectors

tests/collectors_csv.rs

//! Offline INJECT/ASSERT contracts for the **CSV** collector (feature `csv`) —
//! one fixture per equivalence class the test-matrix law demands (RFC-4180,
//! quoted specials, EU `;`+decimal-comma, Latin-1, UTF-8 BOM, ragged, empty).
//! Each feeds `parse_csv` canned bytes and asserts the round-tripped Arrow as
//! DATA (downcast + value), and emits a functional-status row under
//! `--features testmatrix`.
#![cfg(feature = "csv")]

use arrow_array::{Array, BooleanArray, Date32Array, Float64Array, Int64Array, StringArray};
use arrow_schema::DataType;

use korp_collectors::csv::CsvCollector;
use korp_collectors::csv::{CsvDialect, CsvEncoding, parse_csv, sniff_dialect};
use korp_collectors::{BatchSink, Bounds, CollectStats, Collector, CollectorCfg};

mod common;
use common::emit_for;

fn col_types(schema: &arrow_schema::Schema) -> Vec<DataType> {
    schema
        .fields()
        .iter()
        .map(|f| f.data_type().clone())
        .collect()
}

// ── #1 RFC-4180: CRLF, header, typed sniff → [Utf8, Int64, Float64, Date32, Boolean]
const BASIC: &str = "name,age,score,born,active\r\n\
Anna,30,4.5,2020-01-01,true\r\n\
Björn,25,3.0,2019-06-15,false\r\n";

#[test]
fn rfc4180_basic_typed_sniff() {
    let dialect = sniff_dialect(BASIC.as_bytes());
    assert_eq!(dialect.delimiter, b',');
    assert!(dialect.has_headers);
    let (schema, batches) = parse_csv(BASIC.as_bytes(), &dialect).expect("parse basic");
    assert_eq!(
        col_types(&schema),
        vec![
            DataType::Utf8,
            DataType::Int64,
            DataType::Float64,
            DataType::Date32,
            DataType::Boolean
        ],
        "narrowest-type sniff per column"
    );
    let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
    assert_eq!(names, vec!["name", "age", "score", "born", "active"]);

    let rb = &batches[0];
    assert_eq!(rb.num_rows(), 2);
    let name = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
    let age = rb.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
    let score = rb
        .column(2)
        .as_any()
        .downcast_ref::<Float64Array>()
        .unwrap();
    let born = rb.column(3).as_any().downcast_ref::<Date32Array>().unwrap();
    let active = rb
        .column(4)
        .as_any()
        .downcast_ref::<BooleanArray>()
        .unwrap();
    assert_eq!(name.value(0), "Anna");
    assert_eq!(name.value(1), "Björn");
    assert_eq!(age.value(0), 30);
    assert_eq!(score.value(0), 4.5);
    assert_eq!(born.value(0), 18262, "2020-01-01 = 18262 days since epoch");
    assert!(active.value(0));
    assert!(!active.value(1));

    emit_for(
        "korp-collectors/csv",
        "rfc4180_basic_typed_sniff",
        true,
        "5 typed cols, values by downcast",
    );
}

// ── #2 quoted: embedded comma, doubled-quote "", embedded newline ─────────────
const QUOTED: &str = "id,note\r\n1,\"a, b \"\"q\"\" \ncont\"\r\n2,plain\r\n";

#[test]
fn quoted_embedded_specials() {
    let dialect = sniff_dialect(QUOTED.as_bytes());
    let (schema, batches) = parse_csv(QUOTED.as_bytes(), &dialect).expect("parse quoted");
    assert_eq!(col_types(&schema), vec![DataType::Int64, DataType::Utf8]);
    let rb = &batches[0];
    assert_eq!(
        rb.num_rows(),
        2,
        "embedded newline does NOT split the record"
    );
    let note = rb.column(1).as_any().downcast_ref::<StringArray>().unwrap();
    let v0 = note.value(0);
    assert!(v0.contains("a, b"), "embedded comma preserved: {v0:?}");
    assert!(
        v0.contains("\"q\""),
        "doubled-quote unescaped to a literal quote: {v0:?}"
    );
    assert!(
        v0.contains('\n'),
        "embedded newline preserved inside the field: {v0:?}"
    );
    assert_eq!(note.value(1), "plain");

    emit_for(
        "korp-collectors/csv",
        "quoted_embedded_specials",
        true,
        "comma+doubled-quote+newline in one field",
    );
}

// ── #3 EU: `;` delimiter + decimal-comma + space thousands ────────────────────
const EU: &str = "stad;befolkning;andel\r\nSolna;82 000;12,5\r\nSundbyberg;51 000;9,0\r\n";

#[test]
fn semicolon_decimal_comma_eu() {
    let dialect = sniff_dialect(EU.as_bytes());
    assert_eq!(dialect.delimiter, b';');
    assert!(dialect.decimal_comma, "`;` ⇒ EU decimal-comma dialect");
    let (schema, batches) = parse_csv(EU.as_bytes(), &dialect).expect("parse eu");
    assert_eq!(
        col_types(&schema),
        vec![DataType::Utf8, DataType::Int64, DataType::Float64]
    );
    let rb = &batches[0];
    let stad = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
    let pop = rb.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
    let andel = rb
        .column(2)
        .as_any()
        .downcast_ref::<Float64Array>()
        .unwrap();
    assert_eq!(stad.value(0), "Solna");
    assert_eq!(pop.value(0), 82_000, "space thousands stripped");
    assert_eq!(andel.value(0), 12.5, "decimal-comma parsed");
    assert_eq!(andel.value(1), 9.0);

    emit_for(
        "korp-collectors/csv",
        "semicolon_decimal_comma_eu",
        true,
        "; delim, 82 000→82000, 12,5→12.5",
    );
}

// ── #4 Latin-1 / ISO-8859-1 transcode (åäö) ───────────────────────────────────
#[test]
fn latin1_transcode() {
    // "stad,lan\r\nMalmö,Skåne\r\n" in Latin-1 (ö=0xF6, å=0xE5).
    let mut bytes = Vec::new();
    bytes.extend_from_slice(b"stad,lan\r\n");
    bytes.extend_from_slice(&[
        b'M', b'a', b'l', b'm', 0xF6, b',', b'S', b'k', 0xE5, b'n', b'e',
    ]);
    bytes.extend_from_slice(b"\r\n");

    let dialect = sniff_dialect(&bytes);
    assert_eq!(
        dialect.encoding,
        CsvEncoding::Latin1,
        "invalid UTF-8 ⇒ Latin-1"
    );
    let (_schema, batches) = parse_csv(&bytes, &dialect).expect("parse latin1");
    let rb = &batches[0];
    let stad = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
    let lan = rb.column(1).as_any().downcast_ref::<StringArray>().unwrap();
    assert_eq!(stad.value(0), "Malmö", "0xF6 → ö");
    assert_eq!(lan.value(0), "Skåne", "0xE5 → å");

    emit_for(
        "korp-collectors/csv",
        "latin1_transcode",
        true,
        "ISO-8859-1 åäö → UTF-8",
    );
}

// ── #5 UTF-8 BOM strip (no BOM leaking into the first header name) ─────────────
#[test]
fn utf8_bom_strip() {
    let mut bytes = vec![0xEF, 0xBB, 0xBF];
    bytes.extend_from_slice("kommun,invanare\r\nSolna,82000\r\n".as_bytes());

    let dialect = sniff_dialect(&bytes);
    assert_eq!(dialect.encoding, CsvEncoding::Utf8);
    let (schema, batches) = parse_csv(&bytes, &dialect).expect("parse bom");
    assert_eq!(
        schema.field(0).name(),
        "kommun",
        "BOM stripped from header name"
    );
    assert!(!schema.field(0).name().starts_with('\u{feff}'));
    let rb = &batches[0];
    let kommun = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
    assert_eq!(kommun.value(0), "Solna");

    emit_for(
        "korp-collectors/csv",
        "utf8_bom_strip",
        true,
        "EF BB BF stripped, header clean",
    );
}

// ── #6 ragged rows (flexible → graceful, not a panic) ─────────────────────────
const RAGGED: &str = "a,b,c\r\n1,2,3\r\n4,5\r\n6,7,8,9\r\n";

#[test]
fn ragged_flexible_no_panic() {
    let dialect = sniff_dialect(RAGGED.as_bytes());
    let (schema, batches) =
        parse_csv(RAGGED.as_bytes(), &dialect).expect("ragged parses, no panic");
    assert_eq!(
        schema.fields().len(),
        3,
        "header width governs column count"
    );
    let rb = &batches[0];
    assert_eq!(rb.num_rows(), 3);
    let c = rb.column(2).as_any().downcast_ref::<Int64Array>().unwrap();
    assert_eq!(c.value(0), 3);
    assert!(c.is_null(1), "short row → trailing column is null");
    assert_eq!(c.value(2), 8, "extra 4th column on row 3 is dropped");

    emit_for(
        "korp-collectors/csv",
        "ragged_flexible_no_panic",
        true,
        "short→null, extra→dropped, 3 rows",
    );
}

// ── #7 empty / header-only → 0 rows (RAGNARÖK reason) ─────────────────────────
#[tokio::test]
async fn empty_and_header_only_ragnarok() {
    // pure parse: header-only yields a schema but no batches
    let (schema, batches) = parse_csv(b"a,b,c\r\n", &CsvDialect::default()).unwrap();
    assert_eq!(schema.fields().len(), 3);
    assert!(batches.is_empty(), "header-only ⇒ 0 data batches");

    // empty bytes → empty schema + no batches
    let (schema, batches) = parse_csv(b"", &CsvDialect::default()).unwrap();
    assert_eq!(schema.fields().len(), 0);
    assert!(batches.is_empty());

    // through the collector: header-only file must set CollectStats.reason
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("header_only.csv");
    std::fs::write(&path, "a,b,c\r\n").unwrap();
    let cfg = CollectorCfg::new(path.to_string_lossy().to_string(), "t");
    let sink = BatchSink::memory();
    let stats: CollectStats = CsvCollector::default()
        .collect(&cfg, Bounds::default(), &sink)
        .await
        .unwrap();
    assert_eq!(stats.rows, 0);
    assert!(
        stats.reason.is_some(),
        "0-row collect is RED unless reason set (RAGNARÖK)"
    );

    emit_for(
        "korp-collectors/csv",
        "empty_header_only_ragnarok",
        true,
        "0 rows → reason set",
    );
}

// ── descriptor: the CSV collector declares a FlatFile → warehouse target ──────
// The source→target descriptor seam is default-only in `collectors_sample.rs`
// (csv is feature-gated), so assert the file-format collector's declaration here,
// where `--features csv` puts it in the registry.
#[test]
fn csv_collector_declares_flatfile_warehouse_descriptor() {
    use korp_collectors::{SourceEndpoint, TargetKind, registry_descriptors};

    // Direct declaration on the collector.
    let st = CsvCollector::default().source_target();
    assert_eq!(st.source, SourceEndpoint::FlatFile);
    assert_eq!(st.target, TargetKind::WarehouseTable);
    assert_eq!(st.target_table, "csv_import");
    assert!(
        st.map_scene.is_none(),
        "a bulk CSV import feeds no map scene"
    );
    assert!(
        st.label().contains("flat-file"),
        "label carries the source tag: {}",
        st.label()
    );

    // Under `--features csv` it JOINS the registry, so `registry_descriptors()`
    // must enumerate its (source → target) row.
    let descs = registry_descriptors();
    let (_, csv) = descs
        .iter()
        .find(|(m, _)| m.name == "csv")
        .expect("csv collector present in registry under feature csv");
    assert_eq!(csv.source, SourceEndpoint::FlatFile);
    assert_eq!(csv.target, TargetKind::WarehouseTable);
    assert_eq!(csv.target_table, "csv_import");

    emit_for(
        "korp-collectors/csv",
        "declares_source_target",
        true,
        &csv.label(),
    );
}