Source code for ewoksxrpd.tests.test_link_bliss_scan

import logging

import h5py
import numpy
import pytest
from silx.io.dictdump import dicttonx
from silx.io.dictdump import nxtodict
from silx.io.url import DataUrl

from ..tasks.utils.data_utils import link_bliss_scan




def test_link_bliss_scan_with_broken_softlink_with_timeout(tmp_path, caplog):
    _test_link_bliss_scan_with_broken_softlink(tmp_path, caplog, retry_timeout=0)





@pytest.mark.timeout(30)  # without timeout it take 10 sec to skip a broken link
def test_link_bliss_scan_with_broken_softlink_without_timeout(tmp_path, caplog):
    _test_link_bliss_scan_with_broken_softlink(tmp_path, caplog, retry_timeout=None)



def _test_link_bliss_scan_with_broken_softlink(tmp_path, caplog, retry_timeout):
    scan_file = tmp_path / "scan.h5"
    out_file = tmp_path / "out.h5"

    expected = _create_bliss_scan(scan_file)

    with h5py.File(out_file, "w") as f:
        outentry = f.create_group("1.1")
        url = DataUrl(f"{scan_file}::/1.1")

        with caplog.at_level(logging.WARNING):
            link_bliss_scan(
                outentry, url, retry_timeout=retry_timeout, retry_period=0.1
            )

    result = nxtodict(out_file)["1.1"]

    assert _normalize(result) == _normalize(expected)

    expected = "Skip /1.1/measurement/detector2 from Bliss scan link (broken link)"
    warnings = {record.getMessage() for record in caplog.records}
    assert expected in warnings


def _normalize(data):
    """Convert numpy arrays to lists."""
    if isinstance(data, dict):
        return {k: _normalize(v) for k, v in data.items()}
    if isinstance(data, numpy.ndarray):
        return data.tolist()
    return data


def _create_bliss_scan(scan_file):
    scan_dict = {
        "1.1": {
            "@NX_class": "NXentry",
            "end_time": "...",
            "instrument": {
                "@NX_class": "NXinstrument",
                "detector1": {
                    "@NX_class": "NXdetector",
                    "data": [10, 20, 30, 40.0],
                },
                "detector2": {
                    "@NX_class": "NXdetector",
                    # intentionally missing data -> broken link
                },
            },
            "sample": {
                "@NX_class": "NXsample",
                "name": "sampleA",
            },
            "measurement": {
                "@NX_class": "NXcollection",
                ">detector1": "/1.1/instrument/detector1/data",
                ">detector2": "/1.1/instrument/detector2/data",
            },
        },
    }

    dicttonx(scan_dict, scan_file)

    expected = {
        "instrument": scan_dict["1.1"]["instrument"],
        "sample": scan_dict["1.1"]["sample"],
        "measurement": {
            "@NX_class": "NXcollection",
            "detector1": scan_dict["1.1"]["instrument"]["detector1"]["data"],
        },
    }

    return expected