""" Unit tests for testing basic CORE networks. """ import os import threading import time import pytest from xml.etree import ElementTree from mock import MagicMock from conftest import EMANE_SERVICES from core.data import ConfigData from core.enumerations import MessageFlags from core.mobility import BasicRangeModel from core.netns import nodes from core.netns import vnodeclient from core.phys.pnodes import PhysicalNode from core.service import ServiceManager from core.xml import xmlsession _PATH = os.path.abspath(os.path.dirname(__file__)) _SERVICES_PATH = os.path.join(_PATH, "myservices") _MOBILITY_FILE = os.path.join(_PATH, "mobility.scen") _XML_VERSIONS = ["0.0", "1.0"] _NODE_CLASSES = [nodes.PtpNet, nodes.HubNode, nodes.SwitchNode] class TestCore: def test_import_service(self, core): """ Test importing a custom service. :param conftest.Core core: core fixture to test with """ core.session.services.importcustom(_SERVICES_PATH) assert ServiceManager.get("MyService") assert ServiceManager.get("MyService2") @pytest.mark.parametrize("cls", _NODE_CLASSES) def test_nodes(self, core, cls): """ Test ptp node network. :param conftest.Core core: core fixture to test with :param cls: node classes that work within a simple network """ # create ptp network_node = core.session.add_object(cls=cls) # create nodes core.create_node("n1") core.create_node("n2") # add interfaces core.add_interface(network_node, "n1") core.add_interface(network_node, "n2") # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # ping n2 from n1 and assert success status = core.ping("n1", "n2") assert not status @pytest.mark.parametrize("version", _XML_VERSIONS) def test_xml(self, core, tmpdir, version): """ Test xml client methods. :param conftest.Core core: core fixture to test with :param str version: xml version to write and parse """ # create ptp ptp_node = core.session.add_object(cls=nodes.PtpNet) # create nodes core.create_node("n1") core.create_node("n2") # add interfaces core.add_interface(ptp_node, "n1") core.add_interface(ptp_node, "n2") # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # get ids for nodes n1_id = core.get_node("n1").objid n2_id = core.get_node("n2").objid # save xml xml_file = tmpdir.join("session.xml") file_path = xml_file.strpath xmlsession.save_session_xml(core.session, file_path, version) # verify xml file was created and can be parsed assert xml_file.isfile() assert ElementTree.parse(file_path) # stop current session, clearing data core.session.shutdown() # verify nodes have been removed from session with pytest.raises(KeyError): assert not core.session.get_object_by_name(n1_id) with pytest.raises(KeyError): assert not core.session.get_object(n2_id) # load saved xml xmlsession.open_session_xml(core.session, file_path, start=True) # verify nodes have been recreated assert core.session.get_object(n1_id) assert core.session.get_object(n2_id) def test_vnode_client(self, core): """ Test vnode client methods. :param conftest.Core core: core fixture to test with """ # create ptp ptp_node = core.session.add_object(cls=nodes.PtpNet) # create nodes core.create_node("n1") core.create_node("n2") # add interfaces core.add_interface(ptp_node, "n1") core.add_interface(ptp_node, "n2") # get node client for testing n1 = core.get_node("n1") client = n1.vnodeclient # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # check we are connected assert client.connected() # check various command using vcmd module command = ["ls"] assert not client.cmd(command) status, output = client.cmdresult(command) assert not status p, stdin, stdout, stderr = client.popen(command) assert not p.status() assert not client.icmd(command) assert not client.redircmd(MagicMock(), MagicMock(), MagicMock(), command) assert not client.shcmd(command[0]) # check various command using command line vnodeclient.USE_VCMD_MODULE = False assert not client.cmd(command) status, output = client.cmdresult(command) assert not status p, stdin, stdout, stderr = client.popen(command) assert not p.wait() assert not client.icmd(command) assert not client.shcmd(command[0]) # check module methods assert vnodeclient.createclients(core.session.session_dir) # check convenience methods for interface information assert client.getaddr("eth0") assert client.netifstats() def test_netif(self, core): """ Test netif methods. :param conftest.Core core: core fixture to test with """ # create ptp ptp_node = core.session.add_object(cls=nodes.PtpNet) # create nodes core.create_node("n1") core.create_node("n2") # add interfaces n1_interface = core.add_interface(ptp_node, "n1") n2_interface = core.add_interface(ptp_node, "n2") # get nodes n1 = core.get_node("n1") n2 = core.get_node("n2") # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # check link data gets generated assert ptp_node.all_link_data(MessageFlags.ADD.value) # check common nets exist between linked nodes assert n1.commonnets(n2) assert n2.commonnets(n1) # check we can retrieve netif index assert n1.getifindex(n1_interface) == 0 assert n2.getifindex(n2_interface) == 0 # check interface parameters n1_interface.setparam("test", 1) assert n1_interface.getparam("test") == 1 assert n1_interface.getparams() # delete netif and test that if no longer exists n1.delnetif(0) assert not n1.netif(0) def test_physical(self, core): """ Test physical node network. :param conftest.Core core: core fixture to test with """ # create switch node switch_node = core.session.add_object(cls=nodes.SwitchNode) # create a physical node core.create_node(cls=PhysicalNode, name="p1") # mock method that will not work physical_node = core.get_node("p1") physical_node.newnetif = MagicMock(return_value=0) # create regular node core.create_node("n1") # add interface core.add_interface(switch_node, "n1") core.add_interface(switch_node, "p1") # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() def test_wlan_basic_range_good(self, core): """ Test basic wlan network. :param conftest.Core core: core fixture to test with """ # create wlan wlan_node = core.session.add_object(cls=nodes.WlanNode) values = BasicRangeModel.getdefaultvalues() wlan_node.setmodel(BasicRangeModel, values) # create nodes core.create_node("n1", position=(0, 0), services=EMANE_SERVICES, model="mdr") core.create_node("n2", position=(0, 0), services=EMANE_SERVICES, model="mdr") # add interfaces interface_one = core.add_interface(wlan_node, "n1") interface_two = core.add_interface(wlan_node, "n2") # link nodes in wlan core.link(wlan_node, interface_one, interface_two) # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # ping n2 from n1 and assert success status = core.ping("n1", "n2") assert not status def test_wlan_basic_range_bad(self, core): """ Test basic wlan network with leveraging basic range model. :param conftest.Core core: core fixture to test with """ # create wlan wlan_node = core.session.add_object(cls=nodes.WlanNode) values = BasicRangeModel.getdefaultvalues() wlan_node.setmodel(BasicRangeModel, values) # create nodes core.create_node("n1", position=(0, 0), services=EMANE_SERVICES, model="mdr") core.create_node("n2", position=(0, 0), services=EMANE_SERVICES, model="mdr") # add interfaces interface_one = core.add_interface(wlan_node, "n1") interface_two = core.add_interface(wlan_node, "n2") # link nodes in wlan core.link(wlan_node, interface_one, interface_two) # move nodes out of range, default range check is 275 core.get_node("n1").setposition(0, 0) core.get_node("n2").setposition(500, 500) # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # ping n2 from n1 and assert failure ) time.sleep(3) status = core.ping("n1", "n2") assert status def test_mobility(self, core): """ Test basic wlan network. :param conftest.Core core: core fixture to test with """ # create wlan wlan_node = core.session.add_object(cls=nodes.WlanNode) values = BasicRangeModel.getdefaultvalues() wlan_node.setmodel(BasicRangeModel, values) # create nodes core.create_node("n1", objid=1, position=(0, 0), services=EMANE_SERVICES, model="mdr") core.create_node("n2", objid=2, position=(0, 0), services=EMANE_SERVICES, model="mdr") # add interfaces interface_one = core.add_interface(wlan_node, "n1") interface_two = core.add_interface(wlan_node, "n2") # link nodes in wlan core.link(wlan_node, interface_one, interface_two) # configure mobility script for session config = ConfigData( node=wlan_node.objid, object="ns2script", type=0, data_types=(10, 3, 11, 10, 10, 10, 10, 10, 0), data_values="file=%s|refresh_ms=50|loop=1|autostart=0.0|" "map=|script_start=|script_pause=|script_stop=" % _MOBILITY_FILE ) core.session.config_object(config) # add handler for receiving node updates event = threading.Event() def node_update(_): event.set() core.session.node_handlers.append(node_update) # instantiate session core.session.instantiate() # assert node directories created core.assert_nodes() # validate we receive a node message for updating its location assert event.wait(5) def test_link_bandwidth(self, core): """ Test ptp node network with modifying link bandwidth. :param conftest.Core core: core fixture to test with """ # create link network ptp_node, interface_one, interface_two = core.create_link_network() # output csv index bandwidth_index = 8 # run iperf, validate normal bandwidth stdout = core.iperf("n1", "n2") assert stdout value = int(stdout.split(',')[bandwidth_index]) assert 900000 <= value <= 1100000 # change bandwidth in bits per second bandwidth = 500000 core.configure_link(ptp_node, interface_one, interface_two, { "bw": bandwidth }) # run iperf again stdout = core.iperf("n1", "n2") assert stdout value = int(stdout.split(',')[bandwidth_index]) assert 400000 <= value <= 600000 def test_link_loss(self, core): """ Test ptp node network with modifying link packet loss. :param conftest.Core core: core fixture to test with """ # create link network ptp_node, interface_one, interface_two = core.create_link_network() # output csv index loss_index = -2 # run iperf, validate normal bandwidth stdout = core.iperf("n1", "n2") assert stdout value = float(stdout.split(',')[loss_index]) assert 0 <= value <= 0.5 # change bandwidth in bits per second loss = 50 core.configure_link(ptp_node, interface_one, interface_two, { "loss": loss }) # run iperf again stdout = core.iperf("n1", "n2") assert stdout value = float(stdout.split(',')[loss_index]) assert 40 <= value <= 60 def test_link_delay(self, core): """ Test ptp node network with modifying link packet delay. :param conftest.Core core: core fixture to test with """ # create link network ptp_node, interface_one, interface_two = core.create_link_network() # run ping for delay information stdout = core.ping_output("n1", "n2") assert stdout rtt_line = stdout.split("\n")[-1] rtt_values = rtt_line.split("=")[1].split("ms")[0].strip() rtt_avg = float(rtt_values.split("/")[2]) assert 0 <= rtt_avg <= 0.2 # change delay in microseconds delay = 1000000 core.configure_link(ptp_node, interface_one, interface_two, { "delay": delay }) # run ping for delay information again stdout = core.ping_output("n1", "n2") assert stdout rtt_line = stdout.split("\n")[-1] rtt_values = rtt_line.split("=")[1].split("ms")[0].strip() rtt_avg = float(rtt_values.split("/")[2]) assert 1800 <= rtt_avg <= 2200 def test_link_jitter(self, core): """ Test ptp node network with modifying link packet jitter. :param conftest.Core core: core fixture to test with """ # create link network ptp_node, interface_one, interface_two = core.create_link_network() # output csv index jitter_index = 9 # run iperf stdout = core.iperf("n1", "n2") assert stdout value = float(stdout.split(",")[jitter_index]) assert -0.5 <= value <= 0.05 # change jitter in microseconds jitter = 1000000 core.configure_link(ptp_node, interface_one, interface_two, { "jitter": jitter }) # run iperf again stdout = core.iperf("n1", "n2") assert stdout value = float(stdout.split(",")[jitter_index]) assert 200 <= value <= 500