Source code for ewoksscxrd.tests.test_eiger2cbf

from pathlib import Path
from os import PathLike

import fabio
import h5py
import numpy as np
import pytest

from ewoksscxrd.tasks.eiger2cbf import Eiger2CBF

EIGER_FILE = Path(__file__).parent / "data" / "eiger_0000.h5"




@pytest.fixture(scope="module")
def eiger_h5_path():
    assert EIGER_FILE.exists(), f"Test data file not found: {EIGER_FILE}"
    return str(EIGER_FILE)





def test_eiger2cbf_task(tmp_path: PathLike, eiger_h5_path):
    with h5py.File(eiger_h5_path, "r") as h5file:
        frames = h5file["/entry_0000/measurement/data"][()]

    output_dir = tmp_path / "cbf"
    output_pattern = output_dir / "frame_{index}.cbf"

    task = Eiger2CBF(
        inputs={
            "images": [eiger_h5_path],
            "output": str(output_pattern),
            "offset": 1,
            "rotation": 180,
            "flip_lr": True,
            "wavelength": 0.2846,
            "distance": 0.1518,
            "beam": [1052, 1102],
            "alpha": 50,
            "phi": "0",
            "omega": "-36.000000-index*-0.500000",
        }
    )
    task.execute()
    result = task.get_output_values()

    assert result["output_path"] == str(output_dir)

    generated_files = sorted(output_dir.glob("*.cbf"))
    assert len(generated_files) == frames.shape[0]

    expected_frame = np.fliplr(np.rot90(frames[0].astype(np.int32), k=2))
    generated_frame = fabio.open(str(generated_files[0])).data
    assert np.array_equal(generated_frame, expected_frame)





def test_eiger2cbf_reports_input_open_failure(tmp_path: PathLike, monkeypatch):
    input_file = tmp_path / "not_an_image.h5"
    input_file.write_text("not hdf5", encoding="ascii")

    def failing_open(filename):
        raise RuntimeError("bad reader")

    monkeypatch.setattr("ewoksscxrd.tasks.eiger2cbf.fabio_open", failing_open)

    task = Eiger2CBF(
        inputs={
            "images": [str(input_file)],
            "output": str(tmp_path / "cbf" / "frame_{index}.cbf"),
        }
    )

    with pytest.raises(RuntimeError) as excinfo:
        task.execute()

    errors = []
    error = excinfo.value
    while error is not None:
        errors.append(error)
        error = error.__cause__

    messages = "\n".join(str(error) for error in errors)
    assert "Fabio cannot open input file" in messages
    assert str(input_file) in messages
    assert "bad reader" in messages