clawhdf5/crates/clawhdf5/tests/local_heap_interop.rs

//! Old-style (symbol-table) groups keep link names in a local heap. libhdf5
//! validates the heap's free list when it loads the heap and refuses the
//! group ("bad heap free list") when the list points outside the heap; we
//! must refuse too instead of listing names read from a broken heap. Like
//! libhdf5, the check happens when a name is needed, so an empty group with
//! a broken heap still lists.
//!
//! h5py writes the files; skipped when python3 with h5py is unavailable,
//! unless `CLAWHDF5_REQUIRE_INTEROP=1`.

use std::process::Command;

use clawhdf5::File;

fn python() -> String {
    std::env::var("CLAWHDF5_PYTHON").unwrap_or_else(|_| "python3".to_string())
}

fn interop_required() -> bool {
    std::env::var("CLAWHDF5_REQUIRE_INTEROP").is_ok_and(|v| v == "1")
}

fn python_available() -> bool {
    Command::new(python())
        .args(["-c", "import h5py, numpy"])
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
}

macro_rules! skip_if_no_python {
    () => {
        if !python_available() {
            assert!(
                !interop_required(),
                "CLAWHDF5_REQUIRE_INTEROP=1 but python3 with h5py is not available"
            );
            eprintln!("SKIP: python3 with h5py not available");
            return;
        }
    };
}

fn run_python(script: &str) -> String {
    let output = Command::new(python())
        .args(["-c", script])
        .output()
        .expect("failed to run python");
    assert!(
        output.status.success(),
        "python failed:\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8_lossy(&output.stdout).into_owned()
}

#[test]
fn local_heap_free_list_checked_like_libhdf5() {
    skip_if_no_python!();
    let dir = tempfile::tempdir().unwrap();
    let good = dir.path().join("good.h5");
    // Writes `good.h5` (a deleted link leaves a real free block in the root
    // group's heap) and two copies whose root heap free list is broken; for
    // each prints what h5py lists, or `ERROR`.
    let script = format!(
        r#"
import h5py, struct
good = "{good}"
with h5py.File(good, "w", libver="earliest") as f:
    for name in ("alpha", "beta", "gamma"):
        f.create_group(name)
    del f["beta"]
data = bytearray(open(good, "rb").read())
heap = data.find(b"HEAP")  # the root group's heap is written first
size, head, seg = struct.unpack_from("<QQQ", data, heap + 8)
assert head != 1, "expected a free block"
bad_head = bytearray(data)
struct.pack_into("<Q", bad_head, heap + 16, size + 8)
bad_block = bytearray(data)
struct.pack_into("<Q", bad_block, seg + head + 8, size)  # block runs past the end
# libhdf5 only loads a heap when it needs a name: an empty group with the
# same damage still lists (as empty).
empty = good.replace("good.h5", "empty_src.h5")
with h5py.File(empty, "w", libver="earliest") as f:
    pass
bad_empty = bytearray(open(empty, "rb").read())
eheap = bad_empty.find(b"HEAP")
esize = struct.unpack_from("<Q", bad_empty, eheap + 8)[0]
struct.pack_into("<Q", bad_empty, eheap + 16, esize + 8)
for name, content in (("good", data), ("bad_head", bad_head), ("bad_block", bad_block),
                      ("bad_empty", bad_empty)):
    path = good.replace("good.h5", name + ".h5")
    open(path, "wb").write(content)
    try:
        with h5py.File(path, "r") as f:
            print(name, *sorted(f.keys()))
    except Exception as e:
        print(name, "ERROR")
"#,
        good = good.display()
    );
    let out = run_python(&script);
    let lines: Vec<&str> = out.lines().collect();
    assert_eq!(
        lines,
        [
            "good alpha gamma",
            "bad_head ERROR",
            "bad_block ERROR",
            "bad_empty"
        ],
        "h5py's view changed"
    );

    let file = File::open(&good).unwrap();
    let mut groups = file.root().groups().unwrap();
    groups.sort();
    assert_eq!(groups, ["alpha", "gamma"]);

    for name in ["bad_head", "bad_block"] {
        let file = File::open(dir.path().join(format!("{name}.h5"))).unwrap();
        let listed = file.root().groups();
        assert!(
            listed.is_err(),
            "{name}: listed {listed:?} from a heap libhdf5 rejects"
        );
    }

    let file = File::open(dir.path().join("bad_empty.h5")).unwrap();
    assert_eq!(file.root().groups().unwrap(), Vec::<String>::new());
}