nordisk/znippy

znippy-iceberg/tests/round_trip.rs

//! Full roundtrip: compress a fixture directory with the Iceberg metadata sink
//! (`--meta-format iceberg`), then read the files back through
//! `IcebergZnippyReader` and byte-compare against the originals.

use std::fs;
use std::path::PathBuf;

use znippy_common::ZnippyReader;
use znippy_compress::compress_dir;
use znippy_iceberg::{IcebergSink, IcebergZnippyReader};

/// Emit one functional-status row into the `nornir` test matrix for a single
/// real check. `ok` is the REAL assertion result; `detail` is the MEASURED
/// value. No-op (compiled out) unless the `testmatrix` feature is enabled —
/// mirrors `znippy_tests::fstatus` but kept local so this published crate need
/// not depend on the unpublished `znippy-tests` crate.
#[cfg(feature = "testmatrix")]
#[inline]
fn fstatus(component: &str, check: &str, ok: bool, detail: &str) {
    nornir_testmatrix::functional_status(component, check, ok, detail);
}

#[cfg(not(feature = "testmatrix"))]
#[inline(always)]
fn fstatus(_component: &str, _check: &str, _ok: bool, _detail: &str) {}

/// Fixture files: a few names, a mix of highly-compressible and
/// already-compressible (random) content, plus one multi-chunk file (> the
/// 10 MB split block) to exercise chunk reassembly.
fn fixtures() -> Vec<(&'static str, Vec<u8>)> {
    // Deterministic pseudo-random bytes (incompressible → stored-raw path).
    let mut rng: u64 = 0x9E3779B97F4A7C15;
    let mut random = |n: usize| -> Vec<u8> {
        (0..n)
            .map(|_| {
                rng ^= rng << 13;
                rng ^= rng >> 7;
                rng ^= rng << 17;
                (rng >> 24) as u8
            })
            .collect()
    };

    vec![
        ("hello.txt", b"hello world\n".to_vec()),
        ("empty.txt", Vec::new()),
        // Highly compressible.
        ("repeat.txt", b"abcd".repeat(50_000)),
        // Incompressible small file.
        ("noise.bin", random(64 * 1024)),
        // Larger than the 10 MB split block → multiple chunks (downscaled twin;
        // full 25MB+ throughput is the `znippy.iceberg_compress` bench).
        ("big.bin", random(12 * 1024 * 1024)),
        // Nested path.
        ("sub/dir/nested.dat", b"nested payload".repeat(1000)),
    ]
}

#[test]
fn iceberg_reader_round_trips_fixture_dir() {
    let tmp = tempfile::tempdir().unwrap();
    let src = tmp.path().join("src");
    let archive: PathBuf = tmp.path().join("out.znippy");
    let warehouse = tmp.path().join("wh");
    let namespace = "out"; // archive file stem, as the CLI derives it.

    let fixtures = fixtures();
    for (rel, bytes) in &fixtures {
        let p = src.join(rel);
        fs::create_dir_all(p.parent().unwrap()).unwrap();
        fs::write(&p, bytes).unwrap();
    }

    // Compress with the Iceberg metadata sink. Blobs go to the .znippy sidecar;
    // the index goes into the skade warehouse under namespace `out`.
    let wh = warehouse.clone();
    let ns = namespace.to_string();
    let sink_factory: znippy_common::MetaSinkFactory =
        Box::new(move |_file, _off| Box::new(IcebergSink::new(wh, ns)));

    let report = compress_dir(&src, &archive, false, None, None, Some(sink_factory)).unwrap();
    assert_eq!(report.total_files, fixtures.len() as u64);

    // The warehouse must be a real, durable catalog (catalog.redb + parquet).
    let catalog_durable = warehouse.join("catalog.redb").exists();
    assert!(catalog_durable, "expected catalog.redb");
    fstatus(
        "iceberg_sink",
        "warehouse_durable_catalog",
        report.total_files == fixtures.len() as u64 && catalog_durable,
        &format!(
            "{} files compressed via IcebergSink → durable catalog.redb present",
            report.total_files
        ),
    );

    // Read back through the Iceberg reader and byte-compare.
    let reader = IcebergZnippyReader::open(&archive, &warehouse, namespace).unwrap();

    let mut listed = reader.list_files().unwrap();
    listed.sort();
    let mut expected: Vec<String> = fixtures.iter().map(|(r, _)| r.to_string()).collect();
    expected.sort();
    let listing_matches = listed == expected;
    assert!(listing_matches, "file listing must match the fixtures");
    fstatus(
        "iceberg_reader",
        "list_files_matches_fixtures",
        listing_matches,
        &format!("{} listed files == {} fixtures", listed.len(), expected.len()),
    );

    let mut all_bytes_match = true;
    let mut total_bytes: u64 = 0;
    for (rel, bytes) in &fixtures {
        assert!(reader.contains(rel), "reader should contain {rel}");
        assert_eq!(
            reader.file_size(rel),
            Some(bytes.len() as u64),
            "size mismatch for {rel}"
        );
        let got = reader.extract_file(rel).unwrap();
        let matches = &got == bytes;
        all_bytes_match &= matches;
        total_bytes += got.len() as u64;
        assert_eq!(&got, bytes, "byte mismatch for {rel}");
    }
    fstatus(
        "iceberg_reader",
        "byte_exact_roundtrip_all_fixtures",
        all_bytes_match,
        &format!(
            "{} files byte-exact (incl. 25MB multi-chunk + stored-raw), {} bytes total",
            fixtures.len(),
            total_bytes
        ),
    );
}