Source code for blissoda.tests.test_xrpd

import re

import pytest

from ..xrpd.processor import XrpdProcessor




def test_lima_conflicts(mock_persistent, mock_processing_triggers):
    config = {
        "name": "xrpd_processor1",
        "plugin": "generic",
        "class": "XrpdProcessor",
        "package": "blissoda.xrpd.processor",
        "default_parameters": {"lima_names": ["unknown_detector1"], "_enabled": True},
        "shared": False,
        "singleton": False,
    }
    xrpd_processor1 = XrpdProcessor(config=config)
    assert xrpd_processor1.lima_names == ["unknown_detector1"]
    assert xrpd_processor1.get_all_lima_names() == {"unknown_detector1"}

    config = {
        "name": "xrpd_processor2",
        "plugin": "generic",
        "class": "XrpdProcessor",
        "package": "blissoda.xrpd.processor",
        "default_parameters": {"lima_names": ["unknown_detector2"], "_enabled": True},
        "shared": False,
        "singleton": False,
    }
    xrpd_processor2 = XrpdProcessor(config=config)
    assert xrpd_processor2.lima_names == ["unknown_detector2"]
    assert xrpd_processor1.get_all_lima_names() == {
        "unknown_detector1",
        "unknown_detector2",
    }
    assert xrpd_processor2.get_all_lima_names() == {
        "unknown_detector1",
        "unknown_detector2",
    }

    config = {
        "name": "xrpd_processor3",
        "plugin": "generic",
        "class": "XrpdProcessor",
        "package": "blissoda.xrpd.processor",
        "default_parameters": {"lima_names": ["unknown_detector2"], "_enabled": True},
        "shared": False,
        "singleton": False,
    }

    # Fails to instantiate with conflicting Lima name
    expected_error = re.escape(
        "Lima detectors ['unknown_detector2'] are already processed by another XrpdProcessor"
    )
    with pytest.raises(ValueError, match=expected_error):
        _ = XrpdProcessor(config=config)

    # Succeeds to instantiate because Lima name is no longer used
    del xrpd_processor2
    xrpd_processor3 = XrpdProcessor(config=config)
    assert xrpd_processor3.lima_names == ["unknown_detector2"]
    assert xrpd_processor1.get_all_lima_names() == {
        "unknown_detector1",
        "unknown_detector2",
    }
    assert xrpd_processor3.get_all_lima_names() == {
        "unknown_detector1",
        "unknown_detector2",
    }

    # Fails to set Lima name because of conflict
    expected_error = re.escape(
        "Lima detectors ['unknown_detector1'] are already processed by another XrpdProcessor"
    )
    with pytest.raises(ValueError, match=expected_error):
        xrpd_processor3.lima_names = ["unknown_detector1"]

    # Succeeds to set Lima name because Lima name is no longer used
    del xrpd_processor1
    xrpd_processor3.lima_names = ["unknown_detector1"]
    assert xrpd_processor3.lima_names == ["unknown_detector1"]
    assert xrpd_processor3.get_all_lima_names() == {"unknown_detector1"}

    # Succeeds to instantiate because processor is enabled
    config = {
        "name": "xrpd_processor4",
        "plugin": "generic",
        "class": "XrpdProcessor",
        "package": "blissoda.xrpd.processor",
        "default_parameters": {
            "lima_names": ["unknown_detector1"],
        },
        "shared": False,
        "singleton": False,
    }
    xrpd_processor4 = XrpdProcessor(config=config)
    assert xrpd_processor3.get_all_lima_names() == {"unknown_detector1"}
    assert xrpd_processor4.get_all_lima_names() == {"unknown_detector1"}

    # Failes to enable the processing with conflicting Lima name
    expected_error = re.escape(
        "Lima detectors ['unknown_detector1'] are already processed by another XrpdProcessor"
    )
    with pytest.raises(ValueError, match=expected_error):
        xrpd_processor4.enable()