mirror of
https://github.com/ivan-usov-org/bec.git
synced 2025-04-21 10:10:02 +02:00
85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
from unittest import mock
|
|
|
|
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_ds 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},
|
|
),
|
|
BECMessage.DeviceInstructionMessage(
|
|
device="ring_current_sim",
|
|
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]
|
|
dev_man = device_server.device_manager.devices
|
|
for dev in devices:
|
|
if not hasattr(dev_man[dev].obj, "_staged"):
|
|
continue
|
|
assert device_server.device_manager.devices[dev].obj._staged == Staged.yes
|
|
device_server._unstage_device(instr)
|
|
for dev in devices:
|
|
if not hasattr(dev_man[dev].obj, "_staged"):
|
|
continue
|
|
assert device_server.device_manager.devices[dev].obj._staged == Staged.no
|
|
|
|
|
|
def test_stop_devices():
|
|
device_server = load_DeviceServerMock()
|
|
dev = device_server.device_manager.devices
|
|
assert len(dev) > len(dev.enabled_devices)
|
|
with mock.patch.object(dev.samx.obj, "stop") as stop:
|
|
device_server.stop_devices()
|
|
stop.assert_called_once()
|
|
|
|
with mock.patch.object(dev.motor1_disabled.obj, "stop") as stop:
|
|
device_server.stop_devices()
|
|
stop.assert_not_called()
|
|
|
|
with mock.patch.object(dev.motor1_disabled_set.obj, "stop") as stop:
|
|
device_server.stop_devices()
|
|
stop.assert_not_called()
|