afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

cleanup for test fixtures to help simplify test_gui fixture setup

Browse source
This commit is contained in:
bharnden 2019-10-24 10:58:26 -07:00
parent 27be86f175
commit 6045908a14
4 changed files with 259 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
daemon/core/emulator/coreemu.py
View file

@ -49,7 +49,7 @@ class CoreEmu:
self.config = config self.config = config
# session management # session management
self.session_id_gen = IdGen(_id=0) self.session_id_gen = IdGen()
self.sessions = {} self.sessions = {}
# load services # load services
@ -79,6 +79,7 @@ class CoreEmu:
:return: nothing :return: nothing
""" """
logging.info("shutting down all sessions") logging.info("shutting down all sessions")
self.session_id_gen.id = 0
sessions = self.sessions.copy() sessions = self.sessions.copy()
self.sessions.clear() self.sessions.clear()
for _id in sessions: for _id in sessions:

1
daemon/core/emulator/session.py
View file

@ -1368,6 +1368,7 @@ class Session:
while self.nodes: while self.nodes:
_, node = self.nodes.popitem() _, node = self.nodes.popitem()
node.shutdown() node.shutdown()
self.node_id_gen.id = 0
def write_nodes(self): def write_nodes(self):
""" """

30
daemon/tests/conftest.py
View file

@ -129,6 +129,12 @@ class PatchManager:
p.stop() p.stop()
class MockServer:
def __init__(self, config, coreemu):
self.config = config
self.coreemu = coreemu
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def patcher(request): def patcher(request):
patch_manager = PatchManager() patch_manager = PatchManager()
@ -138,6 +144,7 @@ def patcher(request):
patch_manager.patch("core.nodes.netclient.get_net_client") patch_manager.patch("core.nodes.netclient.get_net_client")
patch_manager.patch_obj(CoreNode, "nodefile") patch_manager.patch_obj(CoreNode, "nodefile")
patch_manager.patch_obj(Session, "write_state") patch_manager.patch_obj(Session, "write_state")
patch_manager.patch_obj(Session, "write_nodes")
yield patch_manager yield patch_manager
patch_manager.shutdown() patch_manager.shutdown()
@ -179,11 +186,14 @@ def module_grpc(global_coreemu):
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def module_cored(patcher): def module_coretlv(patcher, global_coreemu, global_session):
server = CoreServerTest() request_mock = MagicMock()
server.setup_handler() request_mock.fileno = MagicMock(return_value=1)
yield server server = MockServer({"numthreads": "1"}, global_coreemu)
server.shutdown() request_handler = CoreHandler(request_mock, "", server)
request_handler.session = global_session
request_handler.add_session_handlers()
yield request_handler
@pytest.fixture @pytest.fixture
@ -204,10 +214,12 @@ def session(global_session):
@pytest.fixture @pytest.fixture
def cored(module_cored): def coretlv(module_coretlv):
session = module_cored.session session = module_coretlv.session
module_cored.server.coreemu.sessions[session.id] = session coreemu = module_coretlv.coreemu
yield module_cored coreemu.sessions[session.id] = session
yield module_coretlv
coreemu.shutdown()
session.clear() session.clear()
session.location.reset() session.location.reset()
session.services.reset() session.services.reset()

468
daemon/tests/test_gui.py
View file

@ -43,10 +43,9 @@ class TestGui:
(NodeTypes.SWITCH, None), (NodeTypes.SWITCH, None),
(NodeTypes.WIRELESS_LAN, None), (NodeTypes.WIRELESS_LAN, None),
(NodeTypes.TUNNEL, 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 node_id = 1
message = coreapi.CoreNodeMessage.create( message = coreapi.CoreNodeMessage.create(
MessageFlags.ADD.value, 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 node_id = 1
cored.session.add_node(_id=node_id) coretlv.session.add_node(_id=node_id)
x = 50 x = 50
y = 100 y = 100
message = coreapi.CoreNodeMessage.create( 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 is not None
assert node.position.x == x assert node.position.x == x
assert node.position.y == y assert node.position.y == y
def test_node_delete(self, cored): def test_node_delete(self, coretlv):
node_id = 1 node_id = 1
cored.session.add_node(_id=node_id) coretlv.session.add_node(_id=node_id)
message = coreapi.CoreNodeMessage.create( message = coreapi.CoreNodeMessage.create(
MessageFlags.DELETE.value, [(NodeTlvs.NUMBER, node_id)] MessageFlags.DELETE.value, [(NodeTlvs.NUMBER, node_id)]
) )
cored.request_handler.handle_message(message) coretlv.handle_message(message)
with pytest.raises(CoreError): 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
switch = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
message = coreapi.CoreLinkMessage.create( 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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
switch = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
message = coreapi.CoreLinkMessage.create( 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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
node_two = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
interface_two = ip_prefix.addr(node_two) 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 = [] all_links = []
for node_id in cored.session.nodes: for node_id in coretlv.session.nodes:
node = cored.session.nodes[node_id] node = coretlv.session.nodes[node_id]
all_links += node.all_link_data(0) all_links += node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
def test_link_update(self, cored): def test_link_update(self, coretlv):
node_one = 1 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
switch = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
message = coreapi.CoreLinkMessage.create( message = coreapi.CoreLinkMessage.create(
@ -192,8 +191,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
link = all_links[0] link = all_links[0]
@ -209,19 +208,19 @@ class TestGui:
(LinkTlvs.BANDWIDTH, bandwidth), (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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
link = all_links[0] link = all_links[0]
assert link.bandwidth == bandwidth 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
node_two = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
interface_two = ip_prefix.addr(node_two) interface_two = ip_prefix.addr(node_two)
@ -237,10 +236,10 @@ class TestGui:
(LinkTlvs.INTERFACE2_IP4_MASK, 24), (LinkTlvs.INTERFACE2_IP4_MASK, 24),
], ],
) )
cored.request_handler.handle_message(message) coretlv.handle_message(message)
all_links = [] all_links = []
for node_id in cored.session.nodes: for node_id in coretlv.session.nodes:
node = cored.session.nodes[node_id] node = coretlv.session.nodes[node_id]
all_links += node.all_link_data(0) all_links += node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
@ -253,19 +252,19 @@ class TestGui:
(LinkTlvs.INTERFACE2_NUMBER, 0), (LinkTlvs.INTERFACE2_NUMBER, 0),
], ],
) )
cored.request_handler.handle_message(message) coretlv.handle_message(message)
all_links = [] all_links = []
for node_id in cored.session.nodes: for node_id in coretlv.session.nodes:
node = cored.session.nodes[node_id] node = coretlv.session.nodes[node_id]
all_links += node.all_link_data(0) all_links += node.all_link_data(0)
assert len(all_links) == 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
switch = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
message = coreapi.CoreLinkMessage.create( message = coreapi.CoreLinkMessage.create(
@ -278,8 +277,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
@ -291,17 +290,17 @@ class TestGui:
(LinkTlvs.INTERFACE1_NUMBER, 0), (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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 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 node_one = 1
cored.session.add_node(_id=node_one) coretlv.session.add_node(_id=node_one)
switch = 2 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") ip_prefix = Ipv4Prefix("10.0.0.0/24")
interface_one = ip_prefix.addr(node_one) interface_one = ip_prefix.addr(node_one)
message = coreapi.CoreLinkMessage.create( message = coreapi.CoreLinkMessage.create(
@ -314,8 +313,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 1 assert len(all_links) == 1
@ -327,58 +326,58 @@ class TestGui:
(LinkTlvs.INTERFACE2_NUMBER, 0), (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) all_links = switch_node.all_link_data(0)
assert len(all_links) == 0 assert len(all_links) == 0
def test_session_update(self, cored): def test_session_update(self, coretlv):
session_id = cored.session.id session_id = coretlv.session.id
name = "test" name = "test"
message = coreapi.CoreSessionMessage.create( message = coreapi.CoreSessionMessage.create(
0, [(SessionTlvs.NUMBER, str(session_id)), (SessionTlvs.NAME, name)] 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): def test_session_query(self, coretlv):
cored.request_handler.dispatch_replies = mock.MagicMock() coretlv.dispatch_replies = mock.MagicMock()
message = coreapi.CoreSessionMessage.create(MessageFlags.STRING.value, []) 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] replies = args[0]
assert len(replies) == 1 assert len(replies) == 1
def test_session_join(self, cored): def test_session_join(self, coretlv):
cored.request_handler.dispatch_replies = mock.MagicMock() coretlv.dispatch_replies = mock.MagicMock()
session_id = cored.session.id session_id = coretlv.session.id
message = coreapi.CoreSessionMessage.create( message = coreapi.CoreSessionMessage.create(
MessageFlags.ADD.value, [(SessionTlvs.NUMBER, str(session_id))] 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): def test_session_delete(self, coretlv):
assert len(cored.server.coreemu.sessions) == 1 assert len(coretlv.coreemu.sessions) == 1
session_id = cored.session.id session_id = coretlv.session.id
message = coreapi.CoreSessionMessage.create( message = coreapi.CoreSessionMessage.create(
MessageFlags.DELETE.value, [(SessionTlvs.NUMBER, str(session_id))] 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 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_name = "test.sh"
file_data = "echo hello" file_data = "echo hello"
message = coreapi.CoreFileMessage.create( 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 assert len(hooks) == 1
name, data = hooks[0] name, data = hooks[0]
assert file_name == name assert file_name == name
assert file_data == data assert file_data == data
def test_file_service_file_set(self, cored): def test_file_service_file_set(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
file_name = "defaultroute.sh" file_name = "defaultroute.sh"
file_data = "echo hello" 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 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" file_name = "/var/log/test/node.log"
node = cored.session.add_node() node = coretlv.session.add_node()
node.makenodedir() node.makenodedir()
file_data = "echo hello" file_data = "echo hello"
message = coreapi.CoreFileMessage.create( 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"): if not request.config.getoption("mock"):
directory, basename = os.path.split(file_name) directory, basename = os.path.split(file_name)
@ -440,9 +441,9 @@ class TestGui:
create_path = os.path.join(node.nodedir, created_directory, basename) create_path = os.path.join(node.nodedir, created_directory, basename)
assert os.path.exists(create_path) assert os.path.exists(create_path)
def test_exec_node_tty(self, cored): def test_exec_node_tty(self, coretlv):
cored.request_handler.dispatch_replies = mock.MagicMock() coretlv.dispatch_replies = mock.MagicMock()
node = cored.session.add_node() node = coretlv.session.add_node()
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TTY.value, 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] replies = args[0]
assert len(replies) == 1 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"): if request.config.getoption("mock"):
pytest.skip("mocking calls") pytest.skip("mocking calls")
cored.request_handler.dispatch_replies = mock.MagicMock() coretlv.dispatch_replies = mock.MagicMock()
node = cored.session.add_node() node = coretlv.session.add_node()
cmd = "echo hello" cmd = "echo hello"
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TEXT.value | MessageFlags.LOCAL.value, 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] replies = args[0]
assert len(replies) == 1 assert len(replies) == 1
def test_exec_node_command(self, cored): def test_exec_node_command(self, coretlv):
cored.request_handler.dispatch_replies = mock.MagicMock() coretlv.dispatch_replies = mock.MagicMock()
node = cored.session.add_node() node = coretlv.session.add_node()
cmd = "echo hello" cmd = "echo hello"
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TEXT.value, MessageFlags.TEXT.value,
@ -494,7 +495,7 @@ class TestGui:
) )
node.cmd = MagicMock(return_value="hello") node.cmd = MagicMock(return_value="hello")
cored.request_handler.handle_message(message) coretlv.handle_message(message)
node.cmd.assert_called_with(cmd) node.cmd.assert_called_with(cmd)
@ -508,16 +509,16 @@ class TestGui:
EventTypes.DEFINITION_STATE, 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)]) 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): def test_event_schedule(self, coretlv):
cored.session.add_event = mock.MagicMock() coretlv.session.add_event = mock.MagicMock()
node = cored.session.add_node() node = coretlv.session.add_node()
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
MessageFlags.ADD.value, 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): def test_event_save_xml(self, coretlv, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("coretlv.session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
cored.session.add_node() coretlv.session.add_node()
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
0, 0,
[(EventTlvs.TYPE, EventTypes.FILE_SAVE.value), (EventTlvs.NAME, file_path)], [(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) assert os.path.exists(file_path)
def test_event_open_xml(self, cored, tmpdir): def test_event_open_xml(self, coretlv, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("coretlv.session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
node = cored.session.add_node() node = coretlv.session.add_node()
cored.session.save_xml(file_path) coretlv.session.save_xml(file_path)
cored.session.delete_node(node.id) coretlv.session.delete_node(node.id)
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
0, 0,
[(EventTlvs.TYPE, EventTypes.FILE_OPEN.value), (EventTlvs.NAME, file_path)], [(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( @pytest.mark.parametrize(
"state", "state",
@ -571,10 +572,9 @@ class TestGui:
EventTypes.RECONFIGURE, EventTypes.RECONFIGURE,
], ],
) )
def test_event_service(self, cored, state): def test_event_service(self, coretlv, state):
cored.session.broadcast_event = mock.MagicMock() coretlv.session.broadcast_event = mock.MagicMock()
node = cored.session.add_node() node = coretlv.session.add_node()
node.startup()
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
0, 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( @pytest.mark.parametrize(
"state", "state",
@ -598,54 +598,54 @@ class TestGui:
EventTypes.RECONFIGURE, EventTypes.RECONFIGURE,
], ],
) )
def test_event_mobility(self, cored, state): def test_event_mobility(self, coretlv, state):
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
0, [(EventTlvs.TYPE, state.value), (EventTlvs.NAME, "mobility:ns2script")] 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): def test_register_gui(self, coretlv):
cored.request_handler.master = False coretlv.master = False
message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")]) 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): def test_register_xml(self, coretlv, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("coretlv.session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
node = cored.session.add_node() node = coretlv.session.add_node()
cored.session.save_xml(file_path) coretlv.session.save_xml(file_path)
cored.session.delete_node(node.id) coretlv.session.delete_node(node.id)
message = coreapi.CoreRegMessage.create( message = coreapi.CoreRegMessage.create(
0, [(RegisterTlvs.EXECUTE_SERVER, file_path)] 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") xml_file = tmpdir.join("test.py")
file_path = xml_file.strpath file_path = xml_file.strpath
with open(file_path, "w") as f: with open(file_path, "w") as f:
f.write("coreemu = globals()['coreemu']\n") 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") f.write("session.add_node()\n")
message = coreapi.CoreRegMessage.create( message = coreapi.CoreRegMessage.create(
0, [(RegisterTlvs.EXECUTE_SERVER, file_path)] 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): def test_config_all(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
MessageFlags.ADD.value, MessageFlags.ADD.value,
[ [
@ -654,13 +654,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (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( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -668,13 +668,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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_key = "test"
test_value = "test" test_value = "test"
values = {test_key: test_value} 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( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -699,13 +699,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (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( message = coreapi.CoreConfMessage.create(
0, 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 coretlv.session.location.refxyz == (10, 10, 0.0)
assert cored.session.location.refgeo == (70, 50, 0) assert coretlv.session.location.refgeo == (70, 50, 0)
assert cored.session.location.refscale == 0.5 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( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -729,13 +729,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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_key = "test"
test_value = "test" test_value = "test"
values = {test_key: test_value} 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" server = "test"
host = "10.0.0.1" host = "10.0.0.1"
port = 50000 port = 50000
@ -764,13 +764,13 @@ class TestGui:
(ConfigTlvs.VALUES, f"{server}:{host}:{port}"), (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( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -778,14 +778,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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): def test_config_services_request_specific(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -795,14 +795,14 @@ class TestGui:
(ConfigTlvs.OPAQUE, "service:DefaultRoute"), (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): def test_config_services_request_specific_file(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -812,16 +812,16 @@ class TestGui:
(ConfigTlvs.OPAQUE, "service:DefaultRoute:defaultroute.sh"), (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): def test_config_services_reset(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
cored.session.services.set_service(node.id, service) coretlv.session.services.set_service(node.id, service)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -829,14 +829,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (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): def test_config_services_set(self, coretlv):
node = cored.session.add_node() node = coretlv.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
values = {"meta": "metadata"} values = {"meta": "metadata"}
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
@ -849,14 +849,14 @@ class TestGui:
(ConfigTlvs.VALUES, dict_to_str(values)), (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): def test_config_mobility_reset(self, coretlv):
wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -864,15 +864,15 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (ConfigTlvs.TYPE, ConfigFlags.RESET.value),
], ],
) )
cored.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {}) coretlv.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {})
assert len(cored.session.mobility.node_configurations) == 1 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): def test_config_mobility_model_request(self, coretlv):
wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -881,14 +881,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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): def test_config_mobility_model_update(self, coretlv):
wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN)
config_key = "range" config_key = "range"
config_value = "1000" config_value = "1000"
values = {config_key: config_value} 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 assert config[config_key] == config_value
def test_config_emane_model_request(self, cored): def test_config_emane_model_request(self, coretlv):
wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -917,14 +919,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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): def test_config_emane_model_update(self, coretlv):
wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = coretlv.session.add_node(_type=NodeTypes.WIRELESS_LAN)
config_key = "distance" config_key = "distance"
config_value = "50051" config_value = "50051"
values = {config_key: config_value} 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 wlan.id, EmaneIeee80211abgModel.name
) )
assert config[config_key] == config_value assert config[config_key] == config_value
def test_config_emane_request(self, cored): def test_config_emane_request(self, coretlv):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -953,13 +955,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (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_key = "eventservicedevice"
config_value = "eth4" config_value = "eth4"
values = {config_key: config_value} 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 assert config[config_key] == config_value