nordisk/korp-collectors

tests/collectors_e2e_skade.rs

//! End-to-end through skade (feature `skade-sink`): a parsed DCAT fixture batch →
//! the real [`BatchSink::iceberg`] → a skade Iceberg commit → read-back via
//! `Table::count` + values. No network — the whole Arrow → Iceberg → read-back
//! contract is proven offline.
#![cfg(feature = "skade-sink")]

use korp_collectors::BatchSink;
use korp_collectors::dcat::{parse_dcat_rdfxml, registry_batch, registry_schema};

mod common;
use common::emit_for;

const SAMPLE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
         xmlns:dcat="http://www.w3.org/ns/dcat#"
         xmlns:dct="http://purl.org/dc/terms/">
  <dcat:Dataset rdf:about="https://data.example.se/dataset/a">
    <dct:title>A</dct:title>
    <dcat:distribution rdf:resource="https://data.example.se/dist/a.csv"/>
  </dcat:Dataset>
  <dcat:Distribution rdf:about="https://data.example.se/dist/a.csv">
    <dcat:downloadURL rdf:resource="https://data.example.se/files/a.csv"/>
    <dct:format>CSV</dct:format>
  </dcat:Distribution>
  <dcat:Dataset rdf:about="https://data.example.se/dataset/b">
    <dct:title>B</dct:title>
  </dcat:Dataset>
</rdf:RDF>"#;

#[tokio::test(flavor = "multi_thread")]
async fn dcat_registry_lands_in_skade_and_reads_back() {
    let warehouse =
        std::env::temp_dir().join(format!("korp_collectors_e2e_{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&warehouse);

    let rows = parse_dcat_rdfxml(SAMPLE.as_bytes()).unwrap();
    assert_eq!(rows.len(), 2, "two datasets → two registry rows");
    let batch = registry_batch(&rows).unwrap();

    // ── land through the real skade sink (korp-lite, no JVM) ────────────────
    let wh = skade::Warehouse::open(&warehouse)
        .await
        .expect("open warehouse");
    let table = wh
        .table_or_create("dcat_registry", &registry_schema())
        .await
        .expect("create table");
    let sink = BatchSink::iceberg(table);
    sink.push(batch).unwrap();
    sink.flush().await.expect("commit to iceberg");
    assert_eq!(sink.rows(), 2);

    // ── read it back (the hard contract) ────────────────────────────────────
    let table = wh.table("dcat_registry").await.expect("reopen table");
    let count = table.count().await.expect("count rows");
    assert_eq!(
        count, 2,
        "skade read-back sees the rows the collector wrote"
    );

    let back = table.read().await.expect("read back");
    let total: usize = back.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total, 2);

    let _ = std::fs::remove_dir_all(&warehouse);
    emit_for(
        "korp-collectors/sink",
        "skade_iceberg_roundtrip",
        count == 2,
        &format!("committed + read back {count} rows via skade"),
    );
}

// ── CSV through skade (feature `csv` + `skade-sink`) ──────────────────────────
// The "through-skade" law for the CSV collector: a real CSV file → the path-
// agnostic `run()` → skade Iceberg → read-back asserts the round-tripped rows.
#[cfg(feature = "csv")]
#[tokio::test(flavor = "multi_thread")]
async fn csv_collector_lands_in_skade_and_reads_back() {
    use korp_collectors::csv::CsvCollector;
    use korp_collectors::{CollectorCfg, run};

    let dir = tempfile::tempdir().unwrap();
    let csv_path = dir.path().join("befolkning.csv");
    std::fs::write(
        &csv_path,
        "kommun,invanare,andel\r\nSolna,82000,12.5\r\nSundbyberg,51000,9.0\r\n",
    )
    .unwrap();

    let warehouse =
        std::env::temp_dir().join(format!("korp_collectors_csv_e2e_{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&warehouse);

    let cfg = CollectorCfg::new(csv_path.to_string_lossy().to_string(), "befolkning");
    let collector = CsvCollector::default();
    let stats = run(&collector, &cfg, warehouse.to_str().unwrap())
        .await
        .expect("run csv → skade");
    assert_eq!(stats.rows, 2, "two data rows collected");

    let wh = skade::Warehouse::open(&warehouse)
        .await
        .expect("open warehouse");
    let table = wh.table("befolkning").await.expect("reopen table");
    let count = table.count().await.expect("count rows");
    assert_eq!(
        count, 2,
        "skade read-back sees the CSV rows the collector wrote"
    );

    let _ = std::fs::remove_dir_all(&warehouse);
    emit_for(
        "korp-collectors/csv",
        "csv_through_skade_roundtrip",
        count == 2,
        &format!("CSV → Iceberg → read back {count} rows"),
    );
}