nordisk/korp-collectors

tests/collectors_znippy.rs

//! The **znippy converter** end-to-end contract (behind `--features znippy`): a
//! REAL collector's parsed Arrow rows → a znippy archive → back, byte-identical.
//! This is the storage/transport form korp uses to ship a collected dataset
//! compactly; the round-trip is asserted as DATA (schema + values) AND at the
//! stored-blob byte level (blake3-verified), not "didn't panic".
#![cfg(feature = "znippy")]

use arrow_array::{Array, Float64Array, StringArray};
use korp_collectors::rest::{RestCollector, parse_records};
use korp_collectors::znippy_pack::{
    batches_to_ipc, pack_blobs, pack_tables, unpack_blob, unpack_tables,
};

/// Parse a real collector fixture (the OpenSky flight-position table) → Arrow.
fn opensky_flight_batch() -> arrow_array::RecordBatch {
    let json: serde_json::Value = serde_json::json!({
        "time": 1_700_000_000i64,
        "states": [
            ["4b1815", "SWR123  ", "Switzerland", 1_700_000_000i64, 1_700_000_000i64,
             8.55, 47.45, 10972.8, false, 231.0, 87.3, 0.0, null, 11277.6, "1000", false, 0],
            ["3c6750", "DLH456  ", "Germany", 1_700_000_000i64, 1_700_000_000i64,
             11.78, 48.35, 3200.0, false, 180.0, 270.0, 5.2, null, 3300.0, "2000", false, 0]
        ]
    });
    let c = RestCollector::opensky();
    parse_records(&json, c.records_pointer(), c.fields()).expect("parse opensky states")
}

#[test]
fn collected_flight_rows_pack_and_unpack_byte_identical() {
    let rb = opensky_flight_batch();
    assert_eq!(rb.num_rows(), 2, "two collected flight rows");

    // Pack the collected table into a znippy archive (compact storage/transport).
    let tables = vec![("opensky_flights".to_string(), vec![rb.clone()])];
    let archive = pack_tables(&tables).expect("pack collected table into znippy");
    assert!(!archive.is_empty(), "archive carries bytes");

    // Read the tables back: name + schema + values reconstruct exactly.
    let back = unpack_tables(&archive).expect("unpack tables");
    assert_eq!(back.len(), 1);
    assert_eq!(back[0].0, "opensky_flights", "table name round-trips");
    let got = &back[0].1[0];
    assert_eq!(got.schema(), rb.schema(), "schema round-trips");
    assert_eq!(got.num_rows(), 2);

    let lat_in = rb
        .column_by_name("lat")
        .unwrap()
        .as_any()
        .downcast_ref::<Float64Array>()
        .unwrap();
    let lat_out = got
        .column_by_name("lat")
        .unwrap()
        .as_any()
        .downcast_ref::<Float64Array>()
        .unwrap();
    assert!(
        (lat_in.value(0) - lat_out.value(0)).abs() < 1e-12,
        "lat value round-trips"
    );
    let cs_in = rb
        .column_by_name("callsign")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let cs_out = got
        .column_by_name("callsign")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    assert_eq!(cs_in.value(0), cs_out.value(0), "callsign round-trips");

    // Byte-identity: the blob znippy stored == the Arrow-IPC we packed (blake3-verified).
    let ipc = batches_to_ipc(&[rb]).expect("ipc");
    let blob = unpack_blob(&archive, "opensky_flights.arrow").expect("extract stored blob");
    assert_eq!(
        blob, ipc,
        "stored Arrow-IPC blob is byte-identical after round-trip"
    );

    korp_collectors::functional_status(
        "korp-collectors/znippy",
        "arrow_table_round_trip",
        true,
        "collected flight rows pack→unpack byte-identical (blake3-verified)",
    );
}

#[test]
fn raw_collected_payload_round_trips_byte_identical() {
    // A raw collected source payload (arbitrary bytes incl. NULs) round-trips exactly.
    let raw = b"raw source payload \x00\x01\x02\xff\xfe compact-shipped".to_vec();
    let archive = pack_blobs(&[("dataset.bin".to_string(), raw.clone())]).expect("pack blob");
    let got = unpack_blob(&archive, "dataset.bin").expect("unpack blob");
    assert_eq!(
        got, raw,
        "raw collected bytes are byte-identical after znippy round-trip"
    );

    korp_collectors::functional_status(
        "korp-collectors/znippy",
        "raw_blob_round_trip",
        true,
        "raw collected payload packs→unpacks byte-identical",
    );
}