mirror of
https://github.com/ivan-usov-org/bec.git
synced 2025-04-21 18:20:01 +02:00
55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
import pytest
|
|
from bec_utils import BECMessage
|
|
from bec_utils.tests.utils import ConnectorMock
|
|
from device_server import DeviceServer
|
|
from ophyd import Staged
|
|
|
|
from test_device_manager import load_device_manager
|
|
|
|
# pylint: disable=missing-function-docstring
|
|
# pylint: disable=protected-access
|
|
|
|
|
|
def load_DeviceServerMock():
|
|
connector = ConnectorMock("")
|
|
device_manager = load_device_manager()
|
|
return DeviceServerMock(device_manager, connector)
|
|
|
|
|
|
class DeviceServerMock(DeviceServer):
|
|
def __init__(self, device_manager, connector_cls) -> None:
|
|
super().__init__(bootstrap_server="dummy", connector_cls=ConnectorMock, scibec_url="dummy")
|
|
self.device_manager = device_manager
|
|
|
|
def _start_device_manager(self):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"instr",
|
|
[
|
|
BECMessage.DeviceInstructionMessage(
|
|
device="samx",
|
|
action="stage",
|
|
parameter={},
|
|
metadata={"stream": "primary", "DIID": 1},
|
|
),
|
|
BECMessage.DeviceInstructionMessage(
|
|
device=["samx", "samy"],
|
|
action="stage",
|
|
parameter={},
|
|
metadata={"stream": "primary", "DIID": 1},
|
|
),
|
|
],
|
|
)
|
|
def test_stage_device(instr):
|
|
device_server = load_DeviceServerMock()
|
|
device_server._stage_device(instr)
|
|
devices = instr.content["device"]
|
|
devices = devices if isinstance(devices, list) else [devices]
|
|
for dev in devices:
|
|
assert device_server.device_manager.devices[dev].obj._staged == Staged.yes
|
|
device_server._unstage_device(instr)
|
|
for dev in devices:
|
|
assert device_server.device_manager.devices[dev].obj._staged == Staged.no
|