From 6045908a144606e51ea3324c44c22c7205cfac67 Mon Sep 17 00:00:00 2001 From: bharnden <32446120+bharnden@users.noreply.github.com> Date: Thu, 24 Oct 2019 10:58:26 -0700 Subject: [PATCH] cleanup for test fixtures to help simplify test_gui fixture setup --- daemon/core/emulator/coreemu.py | 3 +- daemon/core/emulator/session.py | 1 + daemon/tests/conftest.py | 30 +- daemon/tests/test_gui.py | 468 ++++++++++++++++---------------- 4 files changed, 259 insertions(+), 243 deletions(-) diff --git a/daemon/core/emulator/coreemu.py b/daemon/core/emulator/coreemu.py index 237000c7..754ab771 100644 --- a/daemon/core/emulator/coreemu.py +++ b/daemon/core/emulator/coreemu.py @@ -49,7 +49,7 @@ class CoreEmu: self.config = config # session management - self.session_id_gen = IdGen(_id=0) + self.session_id_gen = IdGen() self.sessions = {} # load services @@ -79,6 +79,7 @@ class CoreEmu: :return: nothing """ logging.info("shutting down all sessions") + self.session_id_gen.id = 0 sessions = self.sessions.copy() self.sessions.clear() for _id in sessions: diff --git a/daemon/core/emulator/session.py b/daemon/core/emulator/session.py index 93a1e7d6..2a2c4f11 100644 --- a/daemon/core/emulator/session.py +++ b/daemon/core/emulator/session.py @@ -1368,6 +1368,7 @@ class Session: while self.nodes: _, node = self.nodes.popitem() node.shutdown() + self.node_id_gen.id = 0 def write_nodes(self): """ diff --git a/daemon/tests/conftest.py b/daemon/tests/conftest.py index b57c1d6c..99d641da 100644 --- a/daemon/tests/conftest.py +++ b/daemon/tests/conftest.py @@ -129,6 +129,12 @@ class PatchManager: p.stop() +class MockServer: + def __init__(self, config, coreemu): + self.config = config + self.coreemu = coreemu + + @pytest.fixture(scope="session") def patcher(request): patch_manager = PatchManager() @@ -138,6 +144,7 @@ def patcher(request): patch_manager.patch("core.nodes.netclient.get_net_client") patch_manager.patch_obj(CoreNode, "nodefile") patch_manager.patch_obj(Session, "write_state") + patch_manager.patch_obj(Session, "write_nodes") yield patch_manager patch_manager.shutdown() @@ -179,11 +186,14 @@ def module_grpc(global_coreemu): @pytest.fixture(scope="module") -def module_cored(patcher): - server = CoreServerTest() - server.setup_handler() - yield server - server.shutdown() +def module_coretlv(patcher, global_coreemu, global_session): + request_mock = MagicMock() + request_mock.fileno = MagicMock(return_value=1) + server = MockServer({"numthreads": "1"}, global_coreemu) + request_handler = CoreHandler(request_mock, "", server) + request_handler.session = global_session + request_handler.add_session_handlers() + yield request_handler @pytest.fixture @@ -204,10 +214,12 @@ def session(global_session): @pytest.fixture -def cored(module_cored): - session = module_cored.session - module_cored.server.coreemu.sessions[session.id] = session - yield module_cored +def coretlv(module_coretlv): + session = module_coretlv.session + coreemu = module_coretlv.coreemu + coreemu.sessions[session.id] = session + yield module_coretlv + coreemu.shutdown() session.clear() session.location.reset() session.services.reset() diff --git a/daemon/tests/test_gui.py b/daemon/tests/test_gui.py index 58a6d257..21c14c2c 100644 --- a/daemon/tests/test_gui.py +++ b/daemon/tests/test_gui.py @@ -43,10 +43,9 @@ class TestGui: (NodeTypes.SWITCH, None), (NodeTypes.WIRELESS_LAN, None), (NodeTypes.TUNNEL, None), - (NodeTypes.RJ45, None), ], ) - def test_node_add(self, cored, node_type, model): + def test_node_add(self, coretlv, node_type, model): node_id = 1 message = coreapi.CoreNodeMessage.create( MessageFlags.ADD.value, @@ -60,13 +59,13 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.get_node(node_id) is not None + assert coretlv.session.get_node(node_id) is not None - def test_node_update(self, cored): + def test_node_update(self, coretlv): node_id = 1 - cored.session.add_node(_id=node_id) + coretlv.session.add_node(_id=node_id) x = 50 y = 100 message = coreapi.CoreNodeMessage.create( @@ -78,30 +77,30 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - node = cored.session.get_node(node_id) + node = coretlv.session.get_node(node_id) assert node is not None assert node.position.x == x assert node.position.y == y - def test_node_delete(self, cored): + def test_node_delete(self, coretlv): node_id = 1 - cored.session.add_node(_id=node_id) + coretlv.session.add_node(_id=node_id) message = coreapi.CoreNodeMessage.create( MessageFlags.DELETE.value, [(NodeTlvs.NUMBER, node_id)] ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) with pytest.raises(CoreError): - cored.session.get_node(node_id) + coretlv.session.get_node(node_id) - def test_link_add_node_to_net(self, cored): + def test_link_add_node_to_net(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) switch = 2 - cored.session.add_node(_id=switch, _type=NodeTypes.SWITCH) + coretlv.session.add_node(_id=switch, _type=NodeTypes.SWITCH) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) message = coreapi.CoreLinkMessage.create( @@ -115,17 +114,17 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - switch_node = cored.session.get_node(switch) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 - def test_link_add_net_to_node(self, cored): + def test_link_add_net_to_node(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) switch = 2 - cored.session.add_node(_id=switch, _type=NodeTypes.SWITCH) + coretlv.session.add_node(_id=switch, _type=NodeTypes.SWITCH) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) message = coreapi.CoreLinkMessage.create( @@ -139,17 +138,17 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - switch_node = cored.session.get_node(switch) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 - def test_link_add_node_to_node(self, cored): + def test_link_add_node_to_node(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) node_two = 2 - cored.session.add_node(_id=node_two) + coretlv.session.add_node(_id=node_two) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) interface_two = ip_prefix.addr(node_two) @@ -167,19 +166,19 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) all_links = [] - for node_id in cored.session.nodes: - node = cored.session.nodes[node_id] + for node_id in coretlv.session.nodes: + node = coretlv.session.nodes[node_id] all_links += node.all_link_data(0) assert len(all_links) == 1 - def test_link_update(self, cored): + def test_link_update(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) switch = 2 - cored.session.add_node(_id=switch, _type=NodeTypes.SWITCH) + coretlv.session.add_node(_id=switch, _type=NodeTypes.SWITCH) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) message = coreapi.CoreLinkMessage.create( @@ -192,8 +191,8 @@ class TestGui: (LinkTlvs.INTERFACE1_IP4_MASK, 24), ], ) - cored.request_handler.handle_message(message) - switch_node = cored.session.get_node(switch) + coretlv.handle_message(message) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 link = all_links[0] @@ -209,19 +208,19 @@ class TestGui: (LinkTlvs.BANDWIDTH, bandwidth), ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - switch_node = cored.session.get_node(switch) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 link = all_links[0] assert link.bandwidth == bandwidth - def test_link_delete_node_to_node(self, cored): + def test_link_delete_node_to_node(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) node_two = 2 - cored.session.add_node(_id=node_two) + coretlv.session.add_node(_id=node_two) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) interface_two = ip_prefix.addr(node_two) @@ -237,10 +236,10 @@ class TestGui: (LinkTlvs.INTERFACE2_IP4_MASK, 24), ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) all_links = [] - for node_id in cored.session.nodes: - node = cored.session.nodes[node_id] + for node_id in coretlv.session.nodes: + node = coretlv.session.nodes[node_id] all_links += node.all_link_data(0) assert len(all_links) == 1 @@ -253,19 +252,19 @@ class TestGui: (LinkTlvs.INTERFACE2_NUMBER, 0), ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) all_links = [] - for node_id in cored.session.nodes: - node = cored.session.nodes[node_id] + for node_id in coretlv.session.nodes: + node = coretlv.session.nodes[node_id] all_links += node.all_link_data(0) assert len(all_links) == 0 - def test_link_delete_node_to_net(self, cored): + def test_link_delete_node_to_net(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) switch = 2 - cored.session.add_node(_id=switch, _type=NodeTypes.SWITCH) + coretlv.session.add_node(_id=switch, _type=NodeTypes.SWITCH) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) message = coreapi.CoreLinkMessage.create( @@ -278,8 +277,8 @@ class TestGui: (LinkTlvs.INTERFACE1_IP4_MASK, 24), ], ) - cored.request_handler.handle_message(message) - switch_node = cored.session.get_node(switch) + coretlv.handle_message(message) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 @@ -291,17 +290,17 @@ class TestGui: (LinkTlvs.INTERFACE1_NUMBER, 0), ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - switch_node = cored.session.get_node(switch) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 0 - def test_link_delete_net_to_node(self, cored): + def test_link_delete_net_to_node(self, coretlv): node_one = 1 - cored.session.add_node(_id=node_one) + coretlv.session.add_node(_id=node_one) switch = 2 - cored.session.add_node(_id=switch, _type=NodeTypes.SWITCH) + coretlv.session.add_node(_id=switch, _type=NodeTypes.SWITCH) ip_prefix = Ipv4Prefix("10.0.0.0/24") interface_one = ip_prefix.addr(node_one) message = coreapi.CoreLinkMessage.create( @@ -314,8 +313,8 @@ class TestGui: (LinkTlvs.INTERFACE1_IP4_MASK, 24), ], ) - cored.request_handler.handle_message(message) - switch_node = cored.session.get_node(switch) + coretlv.handle_message(message) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 1 @@ -327,58 +326,58 @@ class TestGui: (LinkTlvs.INTERFACE2_NUMBER, 0), ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - switch_node = cored.session.get_node(switch) + switch_node = coretlv.session.get_node(switch) all_links = switch_node.all_link_data(0) assert len(all_links) == 0 - def test_session_update(self, cored): - session_id = cored.session.id + def test_session_update(self, coretlv): + session_id = coretlv.session.id name = "test" message = coreapi.CoreSessionMessage.create( 0, [(SessionTlvs.NUMBER, str(session_id)), (SessionTlvs.NAME, name)] ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.name == name + assert coretlv.session.name == name - def test_session_query(self, cored): - cored.request_handler.dispatch_replies = mock.MagicMock() + def test_session_query(self, coretlv): + coretlv.dispatch_replies = mock.MagicMock() message = coreapi.CoreSessionMessage.create(MessageFlags.STRING.value, []) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - args, _ = cored.request_handler.dispatch_replies.call_args + args, _ = coretlv.dispatch_replies.call_args replies = args[0] assert len(replies) == 1 - def test_session_join(self, cored): - cored.request_handler.dispatch_replies = mock.MagicMock() - session_id = cored.session.id + def test_session_join(self, coretlv): + coretlv.dispatch_replies = mock.MagicMock() + session_id = coretlv.session.id message = coreapi.CoreSessionMessage.create( MessageFlags.ADD.value, [(SessionTlvs.NUMBER, str(session_id))] ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.request_handler.session.id == session_id + assert coretlv.session.id == session_id - def test_session_delete(self, cored): - assert len(cored.server.coreemu.sessions) == 1 - session_id = cored.session.id + def test_session_delete(self, coretlv): + assert len(coretlv.coreemu.sessions) == 1 + session_id = coretlv.session.id message = coreapi.CoreSessionMessage.create( MessageFlags.DELETE.value, [(SessionTlvs.NUMBER, str(session_id))] ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert len(cored.server.coreemu.sessions) == 0 + assert len(coretlv.coreemu.sessions) == 0 - def test_file_hook_add(self, cored): + def test_file_hook_add(self, coretlv): state = EventTypes.DATACOLLECT_STATE.value - assert cored.session._hooks.get(state) is None + assert coretlv.session._hooks.get(state) is None file_name = "test.sh" file_data = "echo hello" message = coreapi.CoreFileMessage.create( @@ -390,16 +389,16 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - hooks = cored.session._hooks.get(state) + hooks = coretlv.session._hooks.get(state) assert len(hooks) == 1 name, data = hooks[0] assert file_name == name assert file_data == data - def test_file_service_file_set(self, cored): - node = cored.session.add_node() + def test_file_service_file_set(self, coretlv): + node = coretlv.session.add_node() service = "DefaultRoute" file_name = "defaultroute.sh" file_data = "echo hello" @@ -413,14 +412,16 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - service_file = cored.session.services.get_service_file(node, service, file_name) + service_file = coretlv.session.services.get_service_file( + node, service, file_name + ) assert file_data == service_file.data - def test_file_node_file_copy(self, request, cored): + def test_file_node_file_copy(self, request, coretlv): file_name = "/var/log/test/node.log" - node = cored.session.add_node() + node = coretlv.session.add_node() node.makenodedir() file_data = "echo hello" message = coreapi.CoreFileMessage.create( @@ -432,7 +433,7 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) if not request.config.getoption("mock"): directory, basename = os.path.split(file_name) @@ -440,9 +441,9 @@ class TestGui: create_path = os.path.join(node.nodedir, created_directory, basename) assert os.path.exists(create_path) - def test_exec_node_tty(self, cored): - cored.request_handler.dispatch_replies = mock.MagicMock() - node = cored.session.add_node() + def test_exec_node_tty(self, coretlv): + coretlv.dispatch_replies = mock.MagicMock() + node = coretlv.session.add_node() message = coreapi.CoreExecMessage.create( MessageFlags.TTY.value, [ @@ -452,18 +453,18 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - args, _ = cored.request_handler.dispatch_replies.call_args + args, _ = coretlv.dispatch_replies.call_args replies = args[0] assert len(replies) == 1 - def test_exec_local_command(self, request, cored): + def test_exec_local_command(self, request, coretlv): if request.config.getoption("mock"): pytest.skip("mocking calls") - cored.request_handler.dispatch_replies = mock.MagicMock() - node = cored.session.add_node() + coretlv.dispatch_replies = mock.MagicMock() + node = coretlv.session.add_node() cmd = "echo hello" message = coreapi.CoreExecMessage.create( MessageFlags.TEXT.value | MessageFlags.LOCAL.value, @@ -474,15 +475,15 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - args, _ = cored.request_handler.dispatch_replies.call_args + args, _ = coretlv.dispatch_replies.call_args replies = args[0] assert len(replies) == 1 - def test_exec_node_command(self, cored): - cored.request_handler.dispatch_replies = mock.MagicMock() - node = cored.session.add_node() + def test_exec_node_command(self, coretlv): + coretlv.dispatch_replies = mock.MagicMock() + node = coretlv.session.add_node() cmd = "echo hello" message = coreapi.CoreExecMessage.create( MessageFlags.TEXT.value, @@ -494,7 +495,7 @@ class TestGui: ) node.cmd = MagicMock(return_value="hello") - cored.request_handler.handle_message(message) + coretlv.handle_message(message) node.cmd.assert_called_with(cmd) @@ -508,16 +509,16 @@ class TestGui: EventTypes.DEFINITION_STATE, ], ) - def test_event_state(self, cored, state): + def test_event_state(self, coretlv, state): message = coreapi.CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)]) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.state == state.value + assert coretlv.session.state == state.value - def test_event_schedule(self, cored): - cored.session.add_event = mock.MagicMock() - node = cored.session.add_node() + def test_event_schedule(self, coretlv): + coretlv.session.add_event = mock.MagicMock() + node = coretlv.session.add_node() message = coreapi.CoreEventMessage.create( MessageFlags.ADD.value, [ @@ -529,37 +530,37 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.session.add_event.assert_called_once() + coretlv.session.add_event.assert_called_once() - def test_event_save_xml(self, cored, tmpdir): - xml_file = tmpdir.join("session.xml") + def test_event_save_xml(self, coretlv, tmpdir): + xml_file = tmpdir.join("coretlv.session.xml") file_path = xml_file.strpath - cored.session.add_node() + coretlv.session.add_node() message = coreapi.CoreEventMessage.create( 0, [(EventTlvs.TYPE, EventTypes.FILE_SAVE.value), (EventTlvs.NAME, file_path)], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) assert os.path.exists(file_path) - def test_event_open_xml(self, cored, tmpdir): - xml_file = tmpdir.join("session.xml") + def test_event_open_xml(self, coretlv, tmpdir): + xml_file = tmpdir.join("coretlv.session.xml") file_path = xml_file.strpath - node = cored.session.add_node() - cored.session.save_xml(file_path) - cored.session.delete_node(node.id) + node = coretlv.session.add_node() + coretlv.session.save_xml(file_path) + coretlv.session.delete_node(node.id) message = coreapi.CoreEventMessage.create( 0, [(EventTlvs.TYPE, EventTypes.FILE_OPEN.value), (EventTlvs.NAME, file_path)], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.get_node(node.id) + assert coretlv.session.get_node(node.id) @pytest.mark.parametrize( "state", @@ -571,10 +572,9 @@ class TestGui: EventTypes.RECONFIGURE, ], ) - def test_event_service(self, cored, state): - cored.session.broadcast_event = mock.MagicMock() - node = cored.session.add_node() - node.startup() + def test_event_service(self, coretlv, state): + coretlv.session.broadcast_event = mock.MagicMock() + node = coretlv.session.add_node() message = coreapi.CoreEventMessage.create( 0, [ @@ -584,9 +584,9 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.session.broadcast_event.assert_called_once() + coretlv.session.broadcast_event.assert_called_once() @pytest.mark.parametrize( "state", @@ -598,54 +598,54 @@ class TestGui: EventTypes.RECONFIGURE, ], ) - def test_event_mobility(self, cored, state): + def test_event_mobility(self, coretlv, state): message = coreapi.CoreEventMessage.create( 0, [(EventTlvs.TYPE, state.value), (EventTlvs.NAME, "mobility:ns2script")] ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - def test_register_gui(self, cored): - cored.request_handler.master = False + def test_register_gui(self, coretlv): + coretlv.master = False message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")]) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.request_handler.master is True + assert coretlv.master is True - def test_register_xml(self, cored, tmpdir): - xml_file = tmpdir.join("session.xml") + def test_register_xml(self, coretlv, tmpdir): + xml_file = tmpdir.join("coretlv.session.xml") file_path = xml_file.strpath - node = cored.session.add_node() - cored.session.save_xml(file_path) - cored.session.delete_node(node.id) + node = coretlv.session.add_node() + coretlv.session.save_xml(file_path) + coretlv.session.delete_node(node.id) message = coreapi.CoreRegMessage.create( 0, [(RegisterTlvs.EXECUTE_SERVER, file_path)] ) - cored.session.instantiate() + coretlv.session.instantiate() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.server.coreemu.sessions[2].get_node(node.id) + assert coretlv.coreemu.sessions[1].get_node(node.id) - def test_register_python(self, cored, tmpdir): + def test_register_python(self, coretlv, tmpdir): xml_file = tmpdir.join("test.py") file_path = xml_file.strpath with open(file_path, "w") as f: f.write("coreemu = globals()['coreemu']\n") - f.write("session = coreemu.sessions[1]\n") + f.write(f"session = coreemu.sessions[{coretlv.session.id}]\n") f.write("session.add_node()\n") message = coreapi.CoreRegMessage.create( 0, [(RegisterTlvs.EXECUTE_SERVER, file_path)] ) - cored.session.instantiate() + coretlv.session.instantiate() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert len(cored.session.nodes) == 1 + assert len(coretlv.session.nodes) == 1 - def test_config_all(self, cored): - node = cored.session.add_node() + def test_config_all(self, coretlv): + node = coretlv.session.add_node() message = coreapi.CoreConfMessage.create( MessageFlags.ADD.value, [ @@ -654,13 +654,13 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.RESET.value), ], ) - cored.session.location.refxyz = (10, 10, 10) + coretlv.session.location.refxyz = (10, 10, 10) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.location.refxyz == (0, 0, 0) + assert coretlv.session.location.refxyz == (0, 0, 0) - def test_config_options_request(self, cored): + def test_config_options_request(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -668,13 +668,13 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_options_update(self, cored): + def test_config_options_update(self, coretlv): test_key = "test" test_value = "test" values = {test_key: test_value} @@ -687,11 +687,11 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.options.get_config(test_key) == test_value + assert coretlv.session.options.get_config(test_key) == test_value - def test_config_location_reset(self, cored): + def test_config_location_reset(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -699,13 +699,13 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.RESET.value), ], ) - cored.session.location.refxyz = (10, 10, 10) + coretlv.session.location.refxyz = (10, 10, 10) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.location.refxyz == (0, 0, 0) + assert coretlv.session.location.refxyz == (0, 0, 0) - def test_config_location_update(self, cored): + def test_config_location_update(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -715,13 +715,13 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.location.refxyz == (10, 10, 0.0) - assert cored.session.location.refgeo == (70, 50, 0) - assert cored.session.location.refscale == 0.5 + assert coretlv.session.location.refxyz == (10, 10, 0.0) + assert coretlv.session.location.refgeo == (70, 50, 0) + assert coretlv.session.location.refscale == 0.5 - def test_config_metadata_request(self, cored): + def test_config_metadata_request(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -729,13 +729,13 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_metadata_update(self, cored): + def test_config_metadata_update(self, coretlv): test_key = "test" test_value = "test" values = {test_key: test_value} @@ -748,11 +748,11 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.metadata.get_config(test_key) == test_value + assert coretlv.session.metadata.get_config(test_key) == test_value - def test_config_broker_request(self, cored): + def test_config_broker_request(self, coretlv): server = "test" host = "10.0.0.1" port = 50000 @@ -764,13 +764,13 @@ class TestGui: (ConfigTlvs.VALUES, f"{server}:{host}:{port}"), ], ) - cored.session.distributed.add_server = mock.MagicMock() + coretlv.session.distributed.add_server = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.session.distributed.add_server.assert_called_once_with(server, host) + coretlv.session.distributed.add_server.assert_called_once_with(server, host) - def test_config_services_request_all(self, cored): + def test_config_services_request_all(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -778,14 +778,14 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_services_request_specific(self, cored): - node = cored.session.add_node() + def test_config_services_request_specific(self, coretlv): + node = coretlv.session.add_node() message = coreapi.CoreConfMessage.create( 0, [ @@ -795,14 +795,14 @@ class TestGui: (ConfigTlvs.OPAQUE, "service:DefaultRoute"), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_services_request_specific_file(self, cored): - node = cored.session.add_node() + def test_config_services_request_specific_file(self, coretlv): + node = coretlv.session.add_node() message = coreapi.CoreConfMessage.create( 0, [ @@ -812,16 +812,16 @@ class TestGui: (ConfigTlvs.OPAQUE, "service:DefaultRoute:defaultroute.sh"), ], ) - cored.session.broadcast_file = mock.MagicMock() + coretlv.session.broadcast_file = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.session.broadcast_file.assert_called_once() + coretlv.session.broadcast_file.assert_called_once() - def test_config_services_reset(self, cored): - node = cored.session.add_node() + def test_config_services_reset(self, coretlv): + node = coretlv.session.add_node() service = "DefaultRoute" - cored.session.services.set_service(node.id, service) + coretlv.session.services.set_service(node.id, service) message = coreapi.CoreConfMessage.create( 0, [ @@ -829,14 +829,14 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.RESET.value), ], ) - assert cored.session.services.get_service(node.id, service) is not None + assert coretlv.session.services.get_service(node.id, service) is not None - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.services.get_service(node.id, service) is None + assert coretlv.session.services.get_service(node.id, service) is None - def test_config_services_set(self, cored): - node = cored.session.add_node() + def test_config_services_set(self, coretlv): + node = coretlv.session.add_node() service = "DefaultRoute" values = {"meta": "metadata"} message = coreapi.CoreConfMessage.create( @@ -849,14 +849,14 @@ class TestGui: (ConfigTlvs.VALUES, dict_to_str(values)), ], ) - assert cored.session.services.get_service(node.id, service) is None + assert coretlv.session.services.get_service(node.id, service) is None - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert cored.session.services.get_service(node.id, service) is not None + assert coretlv.session.services.get_service(node.id, service) is not None - def test_config_mobility_reset(self, cored): - wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) + def test_config_mobility_reset(self, coretlv): + wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN) message = coreapi.CoreConfMessage.create( 0, [ @@ -864,15 +864,15 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.RESET.value), ], ) - cored.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {}) - assert len(cored.session.mobility.node_configurations) == 1 + coretlv.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {}) + assert len(coretlv.session.mobility.node_configurations) == 1 - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - assert len(cored.session.mobility.node_configurations) == 0 + assert len(coretlv.session.mobility.node_configurations) == 0 - def test_config_mobility_model_request(self, cored): - wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) + def test_config_mobility_model_request(self, coretlv): + wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN) message = coreapi.CoreConfMessage.create( 0, [ @@ -881,14 +881,14 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_mobility_model_update(self, cored): - wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) + def test_config_mobility_model_update(self, coretlv): + wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN) config_key = "range" config_value = "1000" values = {config_key: config_value} @@ -902,13 +902,15 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - config = cored.session.mobility.get_model_config(wlan.id, BasicRangeModel.name) + config = coretlv.session.mobility.get_model_config( + wlan.id, BasicRangeModel.name + ) assert config[config_key] == config_value - def test_config_emane_model_request(self, cored): - wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) + def test_config_emane_model_request(self, coretlv): + wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN) message = coreapi.CoreConfMessage.create( 0, [ @@ -917,14 +919,14 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_emane_model_update(self, cored): - wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) + def test_config_emane_model_update(self, coretlv): + wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN) config_key = "distance" config_value = "50051" values = {config_key: config_value} @@ -938,14 +940,14 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - config = cored.session.emane.get_model_config( + config = coretlv.session.emane.get_model_config( wlan.id, EmaneIeee80211abgModel.name ) assert config[config_key] == config_value - def test_config_emane_request(self, cored): + def test_config_emane_request(self, coretlv): message = coreapi.CoreConfMessage.create( 0, [ @@ -953,13 +955,13 @@ class TestGui: (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), ], ) - cored.request_handler.handle_broadcast_config = mock.MagicMock() + coretlv.handle_broadcast_config = mock.MagicMock() - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - cored.request_handler.handle_broadcast_config.assert_called_once() + coretlv.handle_broadcast_config.assert_called_once() - def test_config_emane_update(self, cored): + def test_config_emane_update(self, coretlv): config_key = "eventservicedevice" config_value = "eth4" values = {config_key: config_value} @@ -972,7 +974,7 @@ class TestGui: ], ) - cored.request_handler.handle_message(message) + coretlv.handle_message(message) - config = cored.session.emane.get_configs() + config = coretlv.session.emane.get_configs() assert config[config_key] == config_value