Source code for ewokstomo.tests.test_linkresults

from pathlib import Path

import pytest

from ewokstomo.tasks.linkresults import LinkSlices, LinkVolumes




def test_linkslices_creates_collection_and_root_symlinks(tmp_path):
    slice_file = (
        tmp_path
        / "visitor"
        / "ma1234"
        / "id00"
        / "20251201"
        / "PROCESSED_DATA"
        / "sample"
        / "sample_dataset"
        / "slices"
        / "sample_dataset_absorption_xy_00008.hdf5"
    )
    slice_file.parent.mkdir(parents=True, exist_ok=True)
    slice_file.write_bytes(b"dummy")

    task = LinkSlices(inputs={"reconstructed_slice_path": str(slice_file)})
    task.execute()

    collection_link = Path(task.outputs.slices_link_collection_file)
    root_link = Path(task.outputs.slices_link_root_file)

    assert collection_link.is_symlink()
    assert root_link.is_symlink()
    assert collection_link.resolve() == slice_file.resolve()
    assert root_link.resolve() == slice_file.resolve()
    assert task.outputs.reconstructed_slice_path == str(slice_file)





def test_linkslices_is_noop_for_non_esrf_path(tmp_path):
    slice_file = tmp_path / "slices" / "slice_00008.hdf5"
    slice_file.parent.mkdir(parents=True, exist_ok=True)
    slice_file.write_bytes(b"dummy")

    task = LinkSlices(inputs={"reconstructed_slice_path": str(slice_file)})
    task.execute()

    assert task.outputs.reconstructed_slice_path == str(slice_file)
    assert task.outputs.slices_link_collection_file is None
    assert task.outputs.slices_link_root_file is None





def test_linkslices_overwrite_guard(tmp_path):
    slice_file = (
        tmp_path
        / "visitor"
        / "ma1234"
        / "id00"
        / "20251201"
        / "PROCESSED_DATA"
        / "sample"
        / "sample_dataset"
        / "slices"
        / "sample_dataset_absorption_xy_00008.hdf5"
    )
    slice_file.parent.mkdir(parents=True, exist_ok=True)
    slice_file.write_bytes(b"dummy")

    first = LinkSlices(
        inputs={"reconstructed_slice_path": str(slice_file), "overwrite": True}
    )
    first.execute()

    collection_link = Path(first.outputs.slices_link_collection_file)
    collection_link.unlink()
    collection_link.write_text("occupied")

    with pytest.raises(RuntimeError, match="overwrite is False"):
        second = LinkSlices(
            inputs={"reconstructed_slice_path": str(slice_file), "overwrite": False}
        )
        second.execute()





def test_linkvolumes_creates_collection_and_root_symlinks(tmp_path):
    volume_file = (
        tmp_path
        / "visitor"
        / "ma1234"
        / "id00"
        / "20251201"
        / "PROCESSED_DATA"
        / "sample"
        / "sample_dataset"
        / "volumes"
        / "sample_dataset_absorption_16Bit_tiff"
        / "sample_dataset.tiff"
    )
    volume_file.parent.mkdir(parents=True, exist_ok=True)
    volume_file.write_bytes(b"dummy")

    task = LinkVolumes(inputs={"reconstructed_volume_path": str(volume_file)})
    task.execute()

    collection_link = Path(task.outputs.volumes_link_collection_file)
    root_link = Path(task.outputs.volumes_link_root_file)

    assert collection_link.is_symlink()
    assert root_link.is_symlink()
    assert collection_link.resolve() == volume_file.parent.resolve()
    assert root_link.resolve() == volume_file.parent.resolve()
    assert task.outputs.reconstructed_volume_path == str(volume_file)





def test_linkvolumes_is_noop_for_non_esrf_path(tmp_path):
    volume_file = tmp_path / "volumes" / "vol_16bit" / "volume.tiff"
    volume_file.parent.mkdir(parents=True, exist_ok=True)
    volume_file.write_bytes(b"dummy")

    task = LinkVolumes(inputs={"reconstructed_volume_path": str(volume_file)})
    task.execute()

    assert task.outputs.reconstructed_volume_path == str(volume_file)
    assert task.outputs.volumes_link_collection_file is None
    assert task.outputs.volumes_link_root_file is None





def test_linkvolumes_overwrite_guard(tmp_path):
    volume_file = (
        tmp_path
        / "visitor"
        / "ma1234"
        / "id00"
        / "20251201"
        / "PROCESSED_DATA"
        / "sample"
        / "sample_dataset"
        / "volumes"
        / "sample_dataset_absorption_16Bit_tiff"
        / "sample_dataset.tiff"
    )
    volume_file.parent.mkdir(parents=True, exist_ok=True)
    volume_file.write_bytes(b"dummy")

    first = LinkVolumes(
        inputs={"reconstructed_volume_path": str(volume_file), "overwrite": True}
    )
    first.execute()

    collection_link = Path(first.outputs.volumes_link_collection_file)
    collection_link.unlink()
    collection_link.mkdir()

    with pytest.raises(RuntimeError, match="overwrite is False"):
        second = LinkVolumes(
            inputs={"reconstructed_volume_path": str(volume_file), "overwrite": False}
        )
        second.execute()