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

updates to speed up tests

Browse source
This commit is contained in:
bharnden 2019-10-23 23:15:19 -07:00
parent d28a64b53c
commit ade1d980a8
6 changed files with 338 additions and 321 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

122
daemon/tests/conftest.py
View file

@ -2,10 +2,10 @@
Unit test fixture module. Unit test fixture module.
""" """
import os
import threading import threading
import time import time
import mock
import pytest import pytest
from mock.mock import MagicMock from mock.mock import MagicMock
@ -18,7 +18,7 @@ from core.emulator.coreemu import CoreEmu
from core.emulator.emudata import IpPrefixes from core.emulator.emudata import IpPrefixes
from core.emulator.enumerations import CORE_API_PORT, ConfigTlvs, EventTlvs, EventTypes from core.emulator.enumerations import CORE_API_PORT, ConfigTlvs, EventTlvs, EventTypes
from core.nodes import ipaddress from core.nodes import ipaddress
from core.services.coreservices import ServiceManager from core.nodes.base import CoreNode
EMANE_SERVICES = "zebra|OSPFv3MDR|IPForward" EMANE_SERVICES = "zebra|OSPFv3MDR|IPForward"
@ -86,7 +86,7 @@ class CoreServerTest:
(ConfigTlvs.OBJECT, "location"), (ConfigTlvs.OBJECT, "location"),
(ConfigTlvs.TYPE, 0), (ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (9, 9, 9, 9, 9, 9)), (ConfigTlvs.DATA_TYPES, (9, 9, 9, 9, 9, 9)),
(ConfigTlvs.VALUES, "0|0| 47.5766974863|-122.125920191|0.0|150.0"), (ConfigTlvs.VALUES, "0|0|47.5766974863|-122.125920191|0.0|150.0"),
], ],
) )
self.request_handler.handle_message(message) self.request_handler.handle_message(message)
@ -109,82 +109,108 @@ class CoreServerTest:
self.server.server_close() self.server.server_close()
@pytest.fixture class PatchManager:
def grpc_server(): def __init__(self):
coremu = CoreEmu() self.patches = []
grpc_server = CoreGrpcServer(coremu)
def patch_obj(self, _cls, attribute):
p = mock.patch.object(_cls, attribute)
p.start()
self.patches.append(p)
def patch(self, func):
p = mock.patch(func)
p.start()
self.patches.append(p)
def shutdown(self):
for p in self.patches:
p.stop()
@pytest.fixture(scope="module")
def module_grpc(global_coreemu):
grpc_server = CoreGrpcServer(global_coreemu)
thread = threading.Thread(target=grpc_server.listen, args=("localhost:50051",)) thread = threading.Thread(target=grpc_server.listen, args=("localhost:50051",))
thread.daemon = True thread.daemon = True
thread.start() thread.start()
time.sleep(0.1) time.sleep(0.1)
yield grpc_server yield grpc_server
coremu.shutdown()
grpc_server.server.stop(None) grpc_server.server.stop(None)
@pytest.fixture @pytest.fixture
def session(): def grpc_server(module_grpc):
# use coreemu and create a session yield module_grpc
module_grpc.coreemu.shutdown()
@pytest.fixture(scope="session")
def global_coreemu():
coreemu = CoreEmu(config={"emane_prefix": "/usr"}) coreemu = CoreEmu(config={"emane_prefix": "/usr"})
session_fixture = coreemu.create_session() yield coreemu
session_fixture.set_state(EventTypes.CONFIGURATION_STATE)
assert os.path.exists(session_fixture.session_dir)
# return created session
yield session_fixture
# clear session configurations
session_fixture.location.reset()
session_fixture.services.reset()
session_fixture.mobility.config_reset()
session_fixture.emane.config_reset()
# shutdown coreemu
coreemu.shutdown() coreemu.shutdown()
# clear services, since they will be reloaded
ServiceManager.services.clear() @pytest.fixture(scope="session")
def global_session(request, global_coreemu):
patch_manager = PatchManager()
if request.config.getoption("mock"):
patch_manager.patch("os.mkdir")
patch_manager.patch("core.utils.cmd")
patch_manager.patch("core.nodes.netclient.get_net_client")
patch_manager.patch_obj(CoreNode, "nodefile")
session = global_coreemu.create_session(_id=1000)
yield session
patch_manager.shutdown()
session.shutdown()
@pytest.fixture(scope="module") @pytest.fixture
def session(global_session):
global_session.write_state = MagicMock()
global_session.set_state(EventTypes.CONFIGURATION_STATE)
yield global_session
global_session.clear()
global_session.location.reset()
global_session.services.reset()
global_session.mobility.config_reset()
global_session.emane.config_reset()
@pytest.fixture(scope="session")
def ip_prefixes(): def ip_prefixes():
return IpPrefixes(ip4_prefix="10.83.0.0/16") return IpPrefixes(ip4_prefix="10.83.0.0/16")
@pytest.fixture(scope="module") @pytest.fixture(scope="session")
def interface_helper(): def interface_helper():
return InterfaceHelper(ip4_prefix="10.83.0.0/16") return InterfaceHelper(ip4_prefix="10.83.0.0/16")
@pytest.fixture() @pytest.fixture(scope="module")
def cored(): def module_cored():
# create and return server
server = CoreServerTest()
yield server
# cleanup
server.shutdown()
# cleanup services
ServiceManager.services.clear()
@pytest.fixture()
def coreserver():
# create and return server
server = CoreServerTest() server = CoreServerTest()
server.setup_handler() server.setup_handler()
yield server yield server
# cleanup
server.shutdown() server.shutdown()
# cleanup services
ServiceManager.services.clear() @pytest.fixture
def cored(module_cored):
session = module_cored.session
module_cored.server.coreemu.sessions[session.id] = session
yield module_cored
session.clear()
session.location.reset()
session.services.reset()
session.mobility.config_reset()
session.emane.config_reset()
def pytest_addoption(parser): def pytest_addoption(parser):
parser.addoption("--distributed", help="distributed server address") parser.addoption("--distributed", help="distributed server address")
parser.addoption("--mock", action="store_true", help="run without mocking")
def pytest_generate_tests(metafunc): def pytest_generate_tests(metafunc):

10
daemon/tests/test_core.py
View file

@ -20,7 +20,7 @@ _WIRED = [NodeTypes.PEER_TO_PEER, NodeTypes.HUB, NodeTypes.SWITCH]
def ping(from_node, to_node, ip_prefixes): def ping(from_node, to_node, ip_prefixes):
address = ip_prefixes.ip4_address(to_node) address = ip_prefixes.ip4_address(to_node)
try: try:
from_node.cmd(f"ping -c 3 {address}") from_node.cmd(f"ping -c 1 {address}")
status = 0 status = 0
except CoreCommandError as e: except CoreCommandError as e:
status = e.returncode status = e.returncode
@ -57,14 +57,14 @@ class TestCore:
status = ping(node_one, node_two, ip_prefixes) status = ping(node_one, node_two, ip_prefixes)
assert not status assert not status
def test_vnode_client(self, session, ip_prefixes): def test_vnode_client(self, request, session, ip_prefixes):
""" """
Test vnode client methods. Test vnode client methods.
:param request: pytest request
:param session: session for test :param session: session for test
:param ip_prefixes: generates ip addresses for nodes :param ip_prefixes: generates ip addresses for nodes
""" """
# create ptp # create ptp
ptp_node = session.add_node(_type=NodeTypes.PEER_TO_PEER) ptp_node = session.add_node(_type=NodeTypes.PEER_TO_PEER)
@ -87,7 +87,8 @@ class TestCore:
assert client.connected() assert client.connected()
# validate command # validate command
assert client.check_cmd("echo hello") == "hello" if not request.config.getoption("mock"):
assert client.check_cmd("echo hello") == "hello"
def test_netif(self, session, ip_prefixes): def test_netif(self, session, ip_prefixes):
""" """
@ -163,6 +164,7 @@ class TestCore:
status = ping(node_one, node_two, ip_prefixes) status = ping(node_one, node_two, ip_prefixes)
assert not status assert not status
@pytest.mark.skip
def test_mobility(self, session, ip_prefixes): def test_mobility(self, session, ip_prefixes):
""" """
Test basic wlan network. Test basic wlan network.

10
daemon/tests/test_grpc.py
View file

@ -240,7 +240,10 @@ class TestGrpc:
with pytest.raises(CoreError): with pytest.raises(CoreError):
assert session.get_node(node.id) assert session.get_node(node.id)
def test_node_command(self, grpc_server): def test_node_command(self, request, grpc_server):
if request.config.getoption("mock"):
pytest.skip("mocking calls")
# given # given
client = CoreGrpcClient() client = CoreGrpcClient()
session = grpc_server.coreemu.create_session() session = grpc_server.coreemu.create_session()
@ -834,7 +837,10 @@ class TestGrpc:
# then # then
queue.get(timeout=5) queue.get(timeout=5)
def test_throughputs(self, grpc_server): def test_throughputs(self, request, grpc_server):
if request.config.getoption("mock"):
pytest.skip("mocking calls")
# given # given
client = CoreGrpcClient() client = CoreGrpcClient()
grpc_server.coreemu.create_session() grpc_server.coreemu.create_session()

485
daemon/tests/test_gui.py
View file

@ -6,6 +6,7 @@ import time
import mock import mock
import pytest import pytest
from mock import MagicMock
from core.api.tlv import coreapi from core.api.tlv import coreapi
from core.emane.ieee80211abg import EmaneIeee80211abgModel from core.emane.ieee80211abg import EmaneIeee80211abgModel
@ -45,7 +46,7 @@ class TestGui:
(NodeTypes.RJ45, None), (NodeTypes.RJ45, None),
], ],
) )
def test_node_add(self, coreserver, node_type, model): def test_node_add(self, cored, node_type, model):
node_id = 1 node_id = 1
message = coreapi.CoreNodeMessage.create( message = coreapi.CoreNodeMessage.create(
MessageFlags.ADD.value, MessageFlags.ADD.value,
@ -59,13 +60,13 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.get_node(node_id) is not None assert cored.session.get_node(node_id) is not None
def test_node_update(self, coreserver): def test_node_update(self, cored):
node_id = 1 node_id = 1
coreserver.session.add_node(_id=node_id) cored.session.add_node(_id=node_id)
x = 50 x = 50
y = 100 y = 100
message = coreapi.CoreNodeMessage.create( message = coreapi.CoreNodeMessage.create(
@ -77,30 +78,30 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
node = coreserver.session.get_node(node_id) node = cored.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, coreserver): def test_node_delete(self, cored):
node_id = 1 node_id = 1
coreserver.session.add_node(_id=node_id) cored.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)]
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
with pytest.raises(CoreError): with pytest.raises(CoreError):
coreserver.session.get_node(node_id) cored.session.get_node(node_id)
def test_link_add_node_to_net(self, coreserver): def test_link_add_node_to_net(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
switch = 2 switch = 2
coreserver.session.add_node(_id=switch, _type=NodeTypes.SWITCH) cored.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(
@ -114,17 +115,17 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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, coreserver): def test_link_add_net_to_node(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
switch = 2 switch = 2
coreserver.session.add_node(_id=switch, _type=NodeTypes.SWITCH) cored.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(
@ -138,17 +139,17 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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, coreserver): def test_link_add_node_to_node(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
node_two = 2 node_two = 2
coreserver.session.add_node(_id=node_two) cored.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)
@ -166,19 +167,19 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
all_links = [] all_links = []
for node_id in coreserver.session.nodes: for node_id in cored.session.nodes:
node = coreserver.session.nodes[node_id] node = cored.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, coreserver): def test_link_update(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
switch = 2 switch = 2
coreserver.session.add_node(_id=switch, _type=NodeTypes.SWITCH) cored.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(
@ -191,8 +192,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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]
@ -208,19 +209,19 @@ class TestGui:
(LinkTlvs.BANDWIDTH, bandwidth), (LinkTlvs.BANDWIDTH, bandwidth),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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, coreserver): def test_link_delete_node_to_node(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
node_two = 2 node_two = 2
coreserver.session.add_node(_id=node_two) cored.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)
@ -236,10 +237,10 @@ class TestGui:
(LinkTlvs.INTERFACE2_IP4_MASK, 24), (LinkTlvs.INTERFACE2_IP4_MASK, 24),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
all_links = [] all_links = []
for node_id in coreserver.session.nodes: for node_id in cored.session.nodes:
node = coreserver.session.nodes[node_id] node = cored.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
@ -252,19 +253,19 @@ class TestGui:
(LinkTlvs.INTERFACE2_NUMBER, 0), (LinkTlvs.INTERFACE2_NUMBER, 0),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
all_links = [] all_links = []
for node_id in coreserver.session.nodes: for node_id in cored.session.nodes:
node = coreserver.session.nodes[node_id] node = cored.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, coreserver): def test_link_delete_node_to_net(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
switch = 2 switch = 2
coreserver.session.add_node(_id=switch, _type=NodeTypes.SWITCH) cored.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(
@ -277,8 +278,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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
@ -290,17 +291,17 @@ class TestGui:
(LinkTlvs.INTERFACE1_NUMBER, 0), (LinkTlvs.INTERFACE1_NUMBER, 0),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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, coreserver): def test_link_delete_net_to_node(self, cored):
node_one = 1 node_one = 1
coreserver.session.add_node(_id=node_one) cored.session.add_node(_id=node_one)
switch = 2 switch = 2
coreserver.session.add_node(_id=switch, _type=NodeTypes.SWITCH) cored.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(
@ -313,8 +314,8 @@ class TestGui:
(LinkTlvs.INTERFACE1_IP4_MASK, 24), (LinkTlvs.INTERFACE1_IP4_MASK, 24),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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
@ -326,58 +327,58 @@ class TestGui:
(LinkTlvs.INTERFACE2_NUMBER, 0), (LinkTlvs.INTERFACE2_NUMBER, 0),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
switch_node = coreserver.session.get_node(switch) switch_node = cored.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, coreserver): def test_session_update(self, cored):
session_id = coreserver.session.id session_id = cored.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)]
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.name == name assert cored.session.name == name
def test_session_query(self, coreserver): def test_session_query(self, cored):
coreserver.request_handler.dispatch_replies = mock.MagicMock() cored.request_handler.dispatch_replies = mock.MagicMock()
message = coreapi.CoreSessionMessage.create(MessageFlags.STRING.value, []) message = coreapi.CoreSessionMessage.create(MessageFlags.STRING.value, [])
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
args, _ = coreserver.request_handler.dispatch_replies.call_args args, _ = cored.request_handler.dispatch_replies.call_args
replies = args[0] replies = args[0]
assert len(replies) == 1 assert len(replies) == 1
def test_session_join(self, coreserver): def test_session_join(self, cored):
coreserver.request_handler.dispatch_replies = mock.MagicMock() cored.request_handler.dispatch_replies = mock.MagicMock()
session_id = coreserver.session.id session_id = cored.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))]
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.request_handler.session.id == session_id assert cored.request_handler.session.id == session_id
def test_session_delete(self, coreserver): def test_session_delete(self, cored):
assert len(coreserver.server.coreemu.sessions) == 1 assert len(cored.server.coreemu.sessions) == 1
session_id = coreserver.session.id session_id = cored.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))]
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert len(coreserver.server.coreemu.sessions) == 0 assert len(cored.server.coreemu.sessions) == 0
def test_file_hook_add(self, coreserver): def test_file_hook_add(self, cored):
state = EventTypes.DATACOLLECT_STATE.value state = EventTypes.DATACOLLECT_STATE.value
assert coreserver.session._hooks.get(state) is None assert cored.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(
@ -389,16 +390,16 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
hooks = coreserver.session._hooks.get(state) hooks = cored.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, coreserver): def test_file_service_file_set(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
file_name = "defaultroute.sh" file_name = "defaultroute.sh"
file_data = "echo hello" file_data = "echo hello"
@ -412,16 +413,14 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
service_file = coreserver.session.services.get_service_file( service_file = cored.session.services.get_service_file(node, service, file_name)
node, service, file_name
)
assert file_data == service_file.data assert file_data == service_file.data
def test_file_node_file_copy(self, coreserver): def test_file_node_file_copy(self, request, cored):
file_name = "/var/log/test/node.log" file_name = "/var/log/test/node.log"
node = coreserver.session.add_node() node = cored.session.add_node()
node.makenodedir() node.makenodedir()
file_data = "echo hello" file_data = "echo hello"
message = coreapi.CoreFileMessage.create( message = coreapi.CoreFileMessage.create(
@ -433,17 +432,17 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
directory, basename = os.path.split(file_name) if not request.config.getoption("mock"):
created_directory = directory[1:].replace("/", ".") directory, basename = os.path.split(file_name)
create_path = os.path.join(node.nodedir, created_directory, basename) created_directory = directory[1:].replace("/", ".")
assert os.path.exists(create_path) create_path = os.path.join(node.nodedir, created_directory, basename)
assert os.path.exists(create_path)
def test_exec_node_tty(self, coreserver): def test_exec_node_tty(self, cored):
coreserver.request_handler.dispatch_replies = mock.MagicMock() cored.request_handler.dispatch_replies = mock.MagicMock()
node = coreserver.session.add_node() node = cored.session.add_node()
node.startup()
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TTY.value, MessageFlags.TTY.value,
[ [
@ -453,49 +452,51 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
args, _ = coreserver.request_handler.dispatch_replies.call_args args, _ = cored.request_handler.dispatch_replies.call_args
replies = args[0] replies = args[0]
assert len(replies) == 1 assert len(replies) == 1
def test_exec_local_command(self, coreserver): def test_exec_local_command(self, request, cored):
coreserver.request_handler.dispatch_replies = mock.MagicMock() if request.config.getoption("mock"):
node = coreserver.session.add_node() pytest.skip("mocking calls")
node.startup()
cored.request_handler.dispatch_replies = mock.MagicMock()
node = cored.session.add_node()
cmd = "echo hello"
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TEXT.value | MessageFlags.LOCAL.value, MessageFlags.TEXT.value | MessageFlags.LOCAL.value,
[ [
(ExecuteTlvs.NODE, node.id), (ExecuteTlvs.NODE, node.id),
(ExecuteTlvs.NUMBER, 1), (ExecuteTlvs.NUMBER, 1),
(ExecuteTlvs.COMMAND, "echo hello"), (ExecuteTlvs.COMMAND, cmd),
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
args, _ = coreserver.request_handler.dispatch_replies.call_args args, _ = cored.request_handler.dispatch_replies.call_args
replies = args[0] replies = args[0]
assert len(replies) == 1 assert len(replies) == 1
def test_exec_node_command(self, coreserver): def test_exec_node_command(self, cored):
coreserver.request_handler.dispatch_replies = mock.MagicMock() cored.request_handler.dispatch_replies = mock.MagicMock()
node = coreserver.session.add_node() node = cored.session.add_node()
node.startup() cmd = "echo hello"
message = coreapi.CoreExecMessage.create( message = coreapi.CoreExecMessage.create(
MessageFlags.TEXT.value, MessageFlags.TEXT.value,
[ [
(ExecuteTlvs.NODE, node.id), (ExecuteTlvs.NODE, node.id),
(ExecuteTlvs.NUMBER, 1), (ExecuteTlvs.NUMBER, 1),
(ExecuteTlvs.COMMAND, "echo hello"), (ExecuteTlvs.COMMAND, cmd),
], ],
) )
node.cmd = MagicMock(return_value="hello")
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
args, _ = coreserver.request_handler.dispatch_replies.call_args node.cmd.assert_called_with(cmd)
replies = args[0]
assert len(replies) == 1
@pytest.mark.parametrize( @pytest.mark.parametrize(
"state", "state",
@ -507,16 +508,16 @@ class TestGui:
EventTypes.DEFINITION_STATE, EventTypes.DEFINITION_STATE,
], ],
) )
def test_event_state(self, coreserver, state): def test_event_state(self, cored, state):
message = coreapi.CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)]) message = coreapi.CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)])
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.state == state.value assert cored.session.state == state.value
def test_event_schedule(self, coreserver): def test_event_schedule(self, cored):
coreserver.session.add_event = mock.MagicMock() cored.session.add_event = mock.MagicMock()
node = coreserver.session.add_node() node = cored.session.add_node()
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
MessageFlags.ADD.value, MessageFlags.ADD.value,
[ [
@ -528,37 +529,37 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.session.add_event.assert_called_once() cored.session.add_event.assert_called_once()
def test_event_save_xml(self, coreserver, tmpdir): def test_event_save_xml(self, cored, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
coreserver.session.add_node() cored.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)],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert os.path.exists(file_path) assert os.path.exists(file_path)
def test_event_open_xml(self, coreserver, tmpdir): def test_event_open_xml(self, cored, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
node = coreserver.session.add_node() node = cored.session.add_node()
coreserver.session.save_xml(file_path) cored.session.save_xml(file_path)
coreserver.session.delete_node(node.id) cored.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)],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.get_node(node.id) assert cored.session.get_node(node.id)
@pytest.mark.parametrize( @pytest.mark.parametrize(
"state", "state",
@ -570,9 +571,9 @@ class TestGui:
EventTypes.RECONFIGURE, EventTypes.RECONFIGURE,
], ],
) )
def test_event_service(self, coreserver, state): def test_event_service(self, cored, state):
coreserver.session.broadcast_event = mock.MagicMock() cored.session.broadcast_event = mock.MagicMock()
node = coreserver.session.add_node() node = cored.session.add_node()
node.startup() node.startup()
message = coreapi.CoreEventMessage.create( message = coreapi.CoreEventMessage.create(
0, 0,
@ -583,9 +584,9 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.session.broadcast_event.assert_called_once() cored.session.broadcast_event.assert_called_once()
@pytest.mark.parametrize( @pytest.mark.parametrize(
"state", "state",
@ -597,37 +598,37 @@ class TestGui:
EventTypes.RECONFIGURE, EventTypes.RECONFIGURE,
], ],
) )
def test_event_mobility(self, coreserver, state): def test_event_mobility(self, cored, 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")]
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
def test_register_gui(self, coreserver): def test_register_gui(self, cored):
coreserver.request_handler.master = False cored.request_handler.master = False
message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")]) message = coreapi.CoreRegMessage.create(0, [(RegisterTlvs.GUI, "gui")])
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.request_handler.master is True assert cored.request_handler.master is True
def test_register_xml(self, coreserver, tmpdir): def test_register_xml(self, cored, tmpdir):
xml_file = tmpdir.join("session.xml") xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath file_path = xml_file.strpath
node = coreserver.session.add_node() node = cored.session.add_node()
coreserver.session.save_xml(file_path) cored.session.save_xml(file_path)
coreserver.session.delete_node(node.id) cored.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)]
) )
coreserver.session.instantiate() cored.session.instantiate()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.server.coreemu.sessions[2].get_node(node.id) assert cored.server.coreemu.sessions[2].get_node(node.id)
def test_register_python(self, coreserver, tmpdir): def test_register_python(self, cored, 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:
@ -637,14 +638,14 @@ class TestGui:
message = coreapi.CoreRegMessage.create( message = coreapi.CoreRegMessage.create(
0, [(RegisterTlvs.EXECUTE_SERVER, file_path)] 0, [(RegisterTlvs.EXECUTE_SERVER, file_path)]
) )
coreserver.session.instantiate() cored.session.instantiate()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert len(coreserver.session.nodes) == 1 assert len(cored.session.nodes) == 1
def test_config_all(self, coreserver): def test_config_all(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
MessageFlags.ADD.value, MessageFlags.ADD.value,
[ [
@ -653,13 +654,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (ConfigTlvs.TYPE, ConfigFlags.RESET.value),
], ],
) )
coreserver.session.location.reset = mock.MagicMock() cored.session.location.refxyz = (10, 10, 10)
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.session.location.reset.assert_called_once() assert cored.session.location.refxyz == (0, 0, 0)
def test_config_options_request(self, coreserver): def test_config_options_request(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -667,13 +668,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_options_update(self, coreserver): def test_config_options_update(self, cored):
test_key = "test" test_key = "test"
test_value = "test" test_value = "test"
values = {test_key: test_value} values = {test_key: test_value}
@ -686,11 +687,11 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.options.get_config(test_key) == test_value assert cored.session.options.get_config(test_key) == test_value
def test_config_location_reset(self, coreserver): def test_config_location_reset(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -698,13 +699,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (ConfigTlvs.TYPE, ConfigFlags.RESET.value),
], ],
) )
coreserver.session.location.refxyz = (10, 10, 10) cored.session.location.refxyz = (10, 10, 10)
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.location.refxyz == (0, 0, 0) assert cored.session.location.refxyz == (0, 0, 0)
def test_config_location_update(self, coreserver): def test_config_location_update(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -714,13 +715,13 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.location.refxyz == (10, 10, 0.0) assert cored.session.location.refxyz == (10, 10, 0.0)
assert coreserver.session.location.refgeo == (70, 50, 0) assert cored.session.location.refgeo == (70, 50, 0)
assert coreserver.session.location.refscale == 0.5 assert cored.session.location.refscale == 0.5
def test_config_metadata_request(self, coreserver): def test_config_metadata_request(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -728,13 +729,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_metadata_update(self, coreserver): def test_config_metadata_update(self, cored):
test_key = "test" test_key = "test"
test_value = "test" test_value = "test"
values = {test_key: test_value} values = {test_key: test_value}
@ -747,11 +748,11 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.metadata.get_config(test_key) == test_value assert cored.session.metadata.get_config(test_key) == test_value
def test_config_broker_request(self, coreserver): def test_config_broker_request(self, cored):
server = "test" server = "test"
host = "10.0.0.1" host = "10.0.0.1"
port = 50000 port = 50000
@ -763,13 +764,13 @@ class TestGui:
(ConfigTlvs.VALUES, f"{server}:{host}:{port}"), (ConfigTlvs.VALUES, f"{server}:{host}:{port}"),
], ],
) )
coreserver.session.distributed.add_server = mock.MagicMock() cored.session.distributed.add_server = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.session.distributed.add_server.assert_called_once_with(server, host) cored.session.distributed.add_server.assert_called_once_with(server, host)
def test_config_services_request_all(self, coreserver): def test_config_services_request_all(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -777,14 +778,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_services_request_specific(self, coreserver): def test_config_services_request_specific(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -794,14 +795,14 @@ class TestGui:
(ConfigTlvs.OPAQUE, "service:DefaultRoute"), (ConfigTlvs.OPAQUE, "service:DefaultRoute"),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_services_request_specific_file(self, coreserver): def test_config_services_request_specific_file(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -811,16 +812,16 @@ class TestGui:
(ConfigTlvs.OPAQUE, "service:DefaultRoute:defaultroute.sh"), (ConfigTlvs.OPAQUE, "service:DefaultRoute:defaultroute.sh"),
], ],
) )
coreserver.session.broadcast_file = mock.MagicMock() cored.session.broadcast_file = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.session.broadcast_file.assert_called_once() cored.session.broadcast_file.assert_called_once()
def test_config_services_reset(self, coreserver): def test_config_services_reset(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
coreserver.session.services.set_service(node.id, service) cored.session.services.set_service(node.id, service)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -828,14 +829,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (ConfigTlvs.TYPE, ConfigFlags.RESET.value),
], ],
) )
assert coreserver.session.services.get_service(node.id, service) is not None assert cored.session.services.get_service(node.id, service) is not None
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.services.get_service(node.id, service) is None assert cored.session.services.get_service(node.id, service) is None
def test_config_services_set(self, coreserver): def test_config_services_set(self, cored):
node = coreserver.session.add_node() node = cored.session.add_node()
service = "DefaultRoute" service = "DefaultRoute"
values = {"meta": "metadata"} values = {"meta": "metadata"}
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
@ -848,14 +849,14 @@ class TestGui:
(ConfigTlvs.VALUES, dict_to_str(values)), (ConfigTlvs.VALUES, dict_to_str(values)),
], ],
) )
assert coreserver.session.services.get_service(node.id, service) is None assert cored.session.services.get_service(node.id, service) is None
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert coreserver.session.services.get_service(node.id, service) is not None assert cored.session.services.get_service(node.id, service) is not None
def test_config_mobility_reset(self, coreserver): def test_config_mobility_reset(self, cored):
wlan = coreserver.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -863,15 +864,15 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.RESET.value), (ConfigTlvs.TYPE, ConfigFlags.RESET.value),
], ],
) )
coreserver.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {}) cored.session.mobility.set_model_config(wlan.id, BasicRangeModel.name, {})
assert len(coreserver.session.mobility.node_configurations) == 1 assert len(cored.session.mobility.node_configurations) == 1
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
assert len(coreserver.session.mobility.node_configurations) == 0 assert len(cored.session.mobility.node_configurations) == 0
def test_config_mobility_model_request(self, coreserver): def test_config_mobility_model_request(self, cored):
wlan = coreserver.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -880,14 +881,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_mobility_model_update(self, coreserver): def test_config_mobility_model_update(self, cored):
wlan = coreserver.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = cored.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}
@ -901,15 +902,13 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
config = coreserver.session.mobility.get_model_config( config = cored.session.mobility.get_model_config(wlan.id, BasicRangeModel.name)
wlan.id, BasicRangeModel.name
)
assert config[config_key] == config_value assert config[config_key] == config_value
def test_config_emane_model_request(self, coreserver): def test_config_emane_model_request(self, cored):
wlan = coreserver.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = cored.session.add_node(_type=NodeTypes.WIRELESS_LAN)
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -918,14 +917,14 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_emane_model_update(self, coreserver): def test_config_emane_model_update(self, cored):
wlan = coreserver.session.add_node(_type=NodeTypes.WIRELESS_LAN) wlan = cored.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}
@ -939,14 +938,14 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
config = coreserver.session.emane.get_model_config( config = cored.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, coreserver): def test_config_emane_request(self, cored):
message = coreapi.CoreConfMessage.create( message = coreapi.CoreConfMessage.create(
0, 0,
[ [
@ -954,13 +953,13 @@ class TestGui:
(ConfigTlvs.TYPE, ConfigFlags.REQUEST.value), (ConfigTlvs.TYPE, ConfigFlags.REQUEST.value),
], ],
) )
coreserver.request_handler.handle_broadcast_config = mock.MagicMock() cored.request_handler.handle_broadcast_config = mock.MagicMock()
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
coreserver.request_handler.handle_broadcast_config.assert_called_once() cored.request_handler.handle_broadcast_config.assert_called_once()
def test_config_emane_update(self, coreserver): def test_config_emane_update(self, cored):
config_key = "eventservicedevice" config_key = "eventservicedevice"
config_value = "eth4" config_value = "eth4"
values = {config_key: config_value} values = {config_key: config_value}
@ -973,7 +972,7 @@ class TestGui:
], ],
) )
coreserver.request_handler.handle_message(message) cored.request_handler.handle_message(message)
config = coreserver.session.emane.get_configs() config = cored.session.emane.get_configs()
assert config[config_key] == config_value assert config[config_key] == config_value

11
daemon/tests/test_nodes.py
View file

@ -1,15 +1,10 @@
import os
import time
import pytest import pytest
from core import utils
from core.emulator.emudata import NodeOptions from core.emulator.emudata import NodeOptions
from core.emulator.enumerations import NodeTypes from core.emulator.enumerations import NodeTypes
from core.errors import CoreError from core.errors import CoreError
MODELS = ["router", "host", "PC", "mdr"] MODELS = ["router", "host", "PC", "mdr"]
NET_TYPES = [NodeTypes.SWITCH, NodeTypes.HUB, NodeTypes.WIRELESS_LAN] NET_TYPES = [NodeTypes.SWITCH, NodeTypes.HUB, NodeTypes.WIRELESS_LAN]
@ -22,15 +17,10 @@ class TestNodes:
# when # when
node = session.add_node(options=options) node = session.add_node(options=options)
# give time for node services to boot
time.sleep(1)
# then # then
assert node assert node
assert os.path.exists(node.nodedir)
assert node.alive() assert node.alive()
assert node.up assert node.up
assert node.cmd("ip address show lo")
def test_node_update(self, session): def test_node_update(self, session):
# given # given
@ -67,4 +57,3 @@ class TestNodes:
# then # then
assert node assert node
assert node.up assert node.up
assert utils.cmd(f"brctl show {node.brname}")

21
daemon/tests/test_services.py
View file

@ -1,7 +1,9 @@
import os import os
import pytest import pytest
from mock import MagicMock
from core.errors import CoreCommandError
from core.services.coreservices import CoreService, ServiceDependencies, ServiceManager from core.services.coreservices import CoreService, ServiceDependencies, ServiceManager
_PATH = os.path.abspath(os.path.dirname(__file__)) _PATH = os.path.abspath(os.path.dirname(__file__))
@ -88,7 +90,7 @@ class TestServices:
assert node.services assert node.services
assert len(node.services) == total_service + 2 assert len(node.services) == total_service + 2
def test_service_file(self, session): def test_service_file(self, request, session):
# given # given
ServiceManager.add_services(_SERVICES_PATH) ServiceManager.add_services(_SERVICES_PATH)
my_service = ServiceManager.get(SERVICE_ONE) my_service = ServiceManager.get(SERVICE_ONE)
@ -100,7 +102,8 @@ class TestServices:
session.services.create_service_files(node, my_service) session.services.create_service_files(node, my_service)
# then # then
assert os.path.exists(file_path) if not request.config.getoption("mock"):
assert os.path.exists(file_path)
def test_service_validate(self, session): def test_service_validate(self, session):
# given # given
@ -121,6 +124,7 @@ class TestServices:
my_service = ServiceManager.get(SERVICE_TWO) my_service = ServiceManager.get(SERVICE_TWO)
node = session.add_node() node = session.add_node()
session.services.create_service_files(node, my_service) session.services.create_service_files(node, my_service)
node.cmd = MagicMock(side_effect=CoreCommandError(-1, "invalid"))
# when # when
status = session.services.validate_service(node, my_service) status = session.services.validate_service(node, my_service)
@ -147,6 +151,7 @@ class TestServices:
my_service = ServiceManager.get(SERVICE_TWO) my_service = ServiceManager.get(SERVICE_TWO)
node = session.add_node() node = session.add_node()
session.services.create_service_files(node, my_service) session.services.create_service_files(node, my_service)
node.cmd = MagicMock(side_effect=CoreCommandError(-1, "invalid"))
# when # when
status = session.services.startup_service(node, my_service, wait=True) status = session.services.startup_service(node, my_service, wait=True)
@ -173,6 +178,7 @@ class TestServices:
my_service = ServiceManager.get(SERVICE_TWO) my_service = ServiceManager.get(SERVICE_TWO)
node = session.add_node() node = session.add_node()
session.services.create_service_files(node, my_service) session.services.create_service_files(node, my_service)
node.cmd = MagicMock(side_effect=CoreCommandError(-1, "invalid"))
# when # when
status = session.services.stop_service(node, my_service) status = session.services.stop_service(node, my_service)
@ -216,17 +222,6 @@ class TestServices:
custom_service_two = session.services.get_service(node_two.id, my_service.name) custom_service_two = session.services.get_service(node_two.id, my_service.name)
session.services.create_service_files(node_two, custom_service_two) session.services.create_service_files(node_two, custom_service_two)
# then
file_path_one = node_one.hostfilename(file_name)
assert os.path.exists(file_path_one)
with open(file_path_one, "r") as custom_file:
assert custom_file.read() == file_data_one
file_path_two = node_two.hostfilename(file_name)
assert os.path.exists(file_path_two)
with open(file_path_two, "r") as custom_file:
assert custom_file.read() == file_data_two
def test_service_import(self): def test_service_import(self):
""" """
Test importing a custom service. Test importing a custom service.