mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 11:41:49 +02:00
test(cli/rpc_register): rpc_register tests added
This commit is contained in:
@ -1,8 +1,15 @@
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from bec_widgets.cli.rpc_register import RPCRegister
|
||||||
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
|
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def rpc_register():
|
||||||
|
yield RPCRegister()
|
||||||
|
RPCRegister.reset_singleton()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
def bec_dispatcher(threads_check):
|
def bec_dispatcher(threads_check):
|
||||||
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
|
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
|
||||||
|
52
tests/unit_tests/test_rpc_register.py
Normal file
52
tests/unit_tests/test_rpc_register.py
Normal file
@ -0,0 +1,52 @@
|
|||||||
|
from bec_widgets.cli.rpc_register import RPCRegister
|
||||||
|
|
||||||
|
|
||||||
|
class FakeObject:
|
||||||
|
def __init__(self, gui_id):
|
||||||
|
self.gui_id = gui_id
|
||||||
|
|
||||||
|
|
||||||
|
def test_add_connection(rpc_register):
|
||||||
|
obj1 = FakeObject("id1")
|
||||||
|
obj2 = FakeObject("id2")
|
||||||
|
|
||||||
|
rpc_register.add_rpc(obj1)
|
||||||
|
rpc_register.add_rpc(obj2)
|
||||||
|
|
||||||
|
all_connections = rpc_register.list_all_connections()
|
||||||
|
|
||||||
|
assert len(all_connections) == 2
|
||||||
|
assert all_connections["id1"] == obj1
|
||||||
|
assert all_connections["id2"] == obj2
|
||||||
|
|
||||||
|
|
||||||
|
def test_remove_connection(rpc_register):
|
||||||
|
|
||||||
|
obj1 = FakeObject("id1")
|
||||||
|
obj2 = FakeObject("id2")
|
||||||
|
|
||||||
|
rpc_register.add_rpc(obj1)
|
||||||
|
rpc_register.add_rpc(obj2)
|
||||||
|
|
||||||
|
rpc_register.remove_rpc(obj1)
|
||||||
|
|
||||||
|
all_connections = rpc_register.list_all_connections()
|
||||||
|
|
||||||
|
assert len(all_connections) == 1
|
||||||
|
assert all_connections["id2"] == obj2
|
||||||
|
|
||||||
|
|
||||||
|
def test_reset_singleton(rpc_register):
|
||||||
|
obj1 = FakeObject("id1")
|
||||||
|
obj2 = FakeObject("id2")
|
||||||
|
|
||||||
|
rpc_register.add_rpc(obj1)
|
||||||
|
rpc_register.add_rpc(obj2)
|
||||||
|
|
||||||
|
rpc_register.reset_singleton()
|
||||||
|
rpc_register = RPCRegister()
|
||||||
|
|
||||||
|
all_connections = rpc_register.list_all_connections()
|
||||||
|
|
||||||
|
assert len(all_connections) == 0
|
||||||
|
assert all_connections == {}
|
Reference in New Issue
Block a user