Source code for ewokstomo.tests.test_convertvolume

from pathlib import Path
from unittest.mock import patch
import numpy as np
import pytest
from ewokstomo.tasks.convert_volume import ConvertVolumeTo16Bit




@pytest.fixture
def fake_hdf5_volume(tmp_path):
    vol_dir = tmp_path / "sample_absorption_32Bit_hdf5"
    vol_dir.mkdir()
    h5 = vol_dir / "sample.hdf5"
    h5.write_text("not really hdf5")
    return h5





@pytest.fixture
def fake_hdf5_volume_no_32bit(tmp_path):
    vol_dir = tmp_path / "sample_absorption_hdf5"
    vol_dir.mkdir()
    h5 = vol_dir / "sample.hdf5"
    h5.write_text("not really hdf5")
    return h5





@pytest.fixture
def fake_tiff_volume(tmp_path):
    vol_dir = tmp_path / "sample_absorption_32Bit_tiff"
    vol_dir.mkdir()
    for i in range(3):
        (vol_dir / f"sample_{i:06d}.tiff").write_text("fake")
    return vol_dir / "sample"





@pytest.fixture
def fake_tiff_volume_no_32bit(tmp_path):
    vol_dir = tmp_path / "sample_absorption_tiff"
    vol_dir.mkdir()
    for i in range(3):
        (vol_dir / f"sample_{i:06d}.tiff").write_text("fake")
    return vol_dir / "sample"



def _make_nabu_dict(fmt):
    return {"output": {"file_format": fmt}}


def _fake_cast_volume_hdf5(
    input_volume, output_volume, output_data_type, remove_input_volume
):
    """Simulate cast_volume for HDF5: write a dummy output file."""
    out = Path(output_volume.file_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("converted")


def _fake_cast_volume_tiff(
    input_volume, output_volume, output_data_type, remove_input_volume
):
    """Simulate cast_volume for tiff: create a dummy output directory.
    MultitiffFVolume exposes its directory via the .url attribute.
    """
    folder = Path(output_volume.url.file_path())
    folder.mkdir(parents=True, exist_ok=True)
    (folder / "slice_000000.tiff").write_bytes(b"dummy")




@patch("ewokstomo.tasks.convert_volume.get_first_hdf5_entry", return_value="/entry")
@patch(
    "ewokstomo.tasks.convert_volume.cast_volume",
    side_effect=_fake_cast_volume_hdf5,
)
def test_convert_volume_hdf5(mock_cast, mock_entry, fake_hdf5_volume):
    task = ConvertVolumeTo16Bit(
        inputs={
            "reconstructed_volume_path": str(fake_hdf5_volume),
            "nabu_dict": _make_nabu_dict("hdf5"),
            "output_format": "hdf5",
        }
    )
    task.execute()
    out = Path(task.outputs.converted_volume_path)

    assert out.exists()
    assert out.stat().st_size > 0
    assert out.suffix == ".hdf5"
    assert "16Bit" in str(out)

    mock_cast.assert_called_once()
    _, kwargs = mock_cast.call_args
    assert kwargs["output_data_type"] == np.uint16





@patch("ewokstomo.tasks.convert_volume.get_first_hdf5_entry", return_value="/entry")
@patch(
    "ewokstomo.tasks.convert_volume.cast_volume",
    side_effect=_fake_cast_volume_hdf5,
)
def test_convert_volume_hdf5_no_32bit_warns(
    mock_cast, mock_entry, fake_hdf5_volume_no_32bit
):
    """When '32Bit' is absent from the HDF5 path, a warning must be emitted and
    the output is placed in an '_16Bit' fallback directory."""
    with patch("ewokstomo.tasks.convert_volume.logger") as mock_logger:
        task = ConvertVolumeTo16Bit(
            inputs={
                "reconstructed_volume_path": str(fake_hdf5_volume_no_32bit),
                "nabu_dict": _make_nabu_dict("hdf5"),
                "output_format": "hdf5",
            }
        )
        task.execute()

        warning_calls = [
            call
            for call in mock_logger.warning.call_args_list
            if "32Bit" in str(call) and "Falling back" in str(call)
        ]
        assert warning_calls, "Expected a warning about missing '32Bit' in path"

    out = Path(task.outputs.converted_volume_path)
    assert out.exists()
    assert out.stat().st_size > 0
    assert out.suffix == ".hdf5"
    assert str(out.parent).endswith("_16Bit"), (
        f"Expected fallback '_16Bit' directory, got: {out.parent}"
    )





@patch(
    "ewokstomo.tasks.convert_volume.cast_volume",
    side_effect=_fake_cast_volume_tiff,
)
def test_convert_volume_tiff(mock_cast, fake_tiff_volume):
    task = ConvertVolumeTo16Bit(
        inputs={
            "reconstructed_volume_path": str(fake_tiff_volume),
            "nabu_dict": _make_nabu_dict("tiff"),
        }
    )
    task.execute()
    out = Path(task.outputs.converted_volume_path)

    assert "16Bit" in str(out)
    assert "sample_absorption_16Bit_tiff" == str(out.parent.stem)
    assert out.parent.exists()
    assert any(out.parent.iterdir())

    mock_cast.assert_called_once()
    _, kwargs = mock_cast.call_args
    assert kwargs["output_data_type"] == np.uint16





@patch(
    "ewokstomo.tasks.convert_volume.cast_volume",
    side_effect=_fake_cast_volume_tiff,
)
def test_convert_volume_tiff_no_32bit_warns(mock_cast, fake_tiff_volume_no_32bit):
    """When '32Bit' is absent from the tiff path, a warning must be emitted and
    the output is placed in an '_16Bit' fallback directory."""
    with patch("ewokstomo.tasks.convert_volume.logger") as mock_logger:
        task = ConvertVolumeTo16Bit(
            inputs={
                "reconstructed_volume_path": str(fake_tiff_volume_no_32bit),
                "nabu_dict": _make_nabu_dict("tiff"),
            }
        )
        task.execute()

        warning_calls = [
            call
            for call in mock_logger.warning.call_args_list
            if "32Bit" in str(call) and "Falling back" in str(call)
        ]
        assert warning_calls, "Expected a warning about missing '32Bit' in path"

    out = Path(task.outputs.converted_volume_path)
    assert "16Bit" in str(out)
    assert str(out.parent).endswith("_16Bit"), (
        f"Expected fallback '_16Bit' directory, got: {out.parent}"
    )
    assert out.parent.exists()
    assert any(out.parent.iterdir())

    mock_cast.assert_called_once()