mirror of
https://github.com/ivan-usov-org/bec.git
synced 2025-04-22 02:20:02 +02:00
test: refactored test to use dev fixture instead of bec_client
This commit is contained in:
parent
9986a72926
commit
fc39c4ad1e
@ -9,29 +9,30 @@ from bec_lib.devicemanager_client import RPCError
|
||||
from bec_lib.tests.utils import bec_client
|
||||
|
||||
|
||||
def test_nested_device_root(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
@pytest.fixture
|
||||
def dev(bec_client):
|
||||
yield bec_client.device_manager.devices
|
||||
|
||||
|
||||
def test_nested_device_root(dev):
|
||||
assert dev.dyn_signals.name == "dyn_signals"
|
||||
assert dev.dyn_signals.messages.name == "messages"
|
||||
assert dev.dyn_signals.root == dev.dyn_signals
|
||||
assert dev.dyn_signals.messages.root == dev.dyn_signals
|
||||
|
||||
|
||||
def test_run_rpc_call(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_run_rpc_call(dev):
|
||||
with mock.patch.object(dev.samx.setpoint, "_get_rpc_response") as mock_rpc:
|
||||
dev.samx.setpoint.set(1)
|
||||
mock_rpc.assert_called_once()
|
||||
|
||||
|
||||
def test_handle_rpc_response(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_handle_rpc_response(dev):
|
||||
msg = messages.DeviceRPCMessage(device="samx", return_val=1, out="done", success=True)
|
||||
assert dev.samx._handle_rpc_response(msg) == 1
|
||||
|
||||
|
||||
def test_handle_rpc_response_returns_status(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_handle_rpc_response_returns_status(dev, bec_client):
|
||||
msg = messages.DeviceRPCMessage(
|
||||
device="samx", return_val={"type": "status", "RID": "request_id"}, out="done", success=True
|
||||
)
|
||||
@ -40,8 +41,7 @@ def test_handle_rpc_response_returns_status(bec_client):
|
||||
)
|
||||
|
||||
|
||||
def test_handle_rpc_response_raises(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_handle_rpc_response_raises(dev):
|
||||
msg = messages.DeviceRPCMessage(
|
||||
device="samx",
|
||||
return_val={"type": "status", "RID": "request_id"},
|
||||
@ -56,8 +56,7 @@ def test_handle_rpc_response_raises(bec_client):
|
||||
dev.samx._handle_rpc_response(msg)
|
||||
|
||||
|
||||
def test_handle_rpc_response_returns_dict(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_handle_rpc_response_returns_dict(dev):
|
||||
msg = messages.DeviceRPCMessage(
|
||||
device="samx",
|
||||
return_val={"a": "b"},
|
||||
@ -67,8 +66,7 @@ def test_handle_rpc_response_returns_dict(bec_client):
|
||||
assert dev.samx._handle_rpc_response(msg) == {"a": "b"}
|
||||
|
||||
|
||||
def test_run_rpc_call_calls_stop_on_keyboardinterrupt(bec_client):
|
||||
dev = bec_client.device_manager.devices
|
||||
def test_run_rpc_call_calls_stop_on_keyboardinterrupt(dev):
|
||||
with mock.patch.object(dev.samx.setpoint, "_prepare_rpc_msg") as mock_rpc:
|
||||
mock_rpc.side_effect = [KeyboardInterrupt]
|
||||
with pytest.raises(RPCError):
|
||||
|
Loading…
x
Reference in New Issue
Block a user