Source code for blissoda.demo.tests.itest_fluo

import pathlib

from ...bliss_globals import setup_globals
from .. import testing
from ..processors.fluo import DemoFluoProcessor




@testing.integration_fixture
def fluo_processor():
    processor = DemoFluoProcessor()
    processor.enable()
    yield processor
    processor.disable()





@testing.integration_test
def testfluo_xrfmap(fluo_processor, expo=0.2, shape=(3, 4)):
    robx_start = 0
    robx_step_size = 0.01
    robx_intervals = shape[1] - 1
    robx_end = robx_start + robx_step_size * robx_intervals

    roby_start = 0
    roby_step_size = -0.01
    roby_intervals = shape[0] - 1
    roby_end = roby_start + roby_step_size * roby_intervals

    _ = setup_globals.amesh(
        setup_globals.robx,
        robx_start,
        robx_end,
        robx_intervals,
        setup_globals.roby,
        roby_start,
        roby_end,
        roby_intervals,
        expo,
        setup_globals.mca1,
        setup_globals.mca2,
        setup_globals.diode1,
    )

    assert fluo_processor._last_future is not None, "Fluo workflow not triggered"
    result = fluo_processor._last_future.result(timeout=20)

    for key in ("output_root_uri", "xrf_results_uri"):
        file_path, data_path = result[key].split("::")
        testing.assert_hdf5_group_exists(pathlib.Path(file_path), data_path)

    file_path, data_path = result["xrf_results_uri"].split("::")
    data_path += "/parameters/Cr_K"
    testing.assert_hdf5_dataset_exists(pathlib.Path(file_path), data_path, shape)