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

moved future core server and handler code to act as the default core-daemon, updated future examples and tests to leverage new api

Browse source
This commit is contained in:
Blake J. Harnden 2018-04-25 16:33:58 -07:00
parent f431895357
commit 8644e9d61e
24 changed files with 618 additions and 2728 deletions
Download patch file Download diff file Expand all files Collapse all files

433
daemon/tests/test_core.py
View file

@ -11,22 +11,22 @@ from xml.etree import ElementTree
import pytest
from mock import MagicMock
from conftest import EMANE_SERVICES
from core.data import ConfigData
from core.enumerations import MessageFlags
from core.enumerations import MessageFlags, NodeTypes
from core.future.futuredata import NodeOptions
from core.mobility import BasicRangeModel
from core.netns import nodes
from core.netns import vnodeclient
from core.netns.vnodeclient 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]
_WIRED = [
NodeTypes.PEER_TO_PEER,
NodeTypes.HUB,
NodeTypes.SWITCH
]
def createclients(sessiondir, clientcls=VnodeClient, cmdchnlfilterfunc=None):
@ -47,8 +47,13 @@ def createclients(sessiondir, clientcls=VnodeClient, cmdchnlfilterfunc=None):
return map(lambda x: clientcls(os.path.basename(x), x), cmdchnls)
def ping(from_node, to_node, ip_prefixes):
address = ip_prefixes.ip4_address(to_node)
return from_node.cmd(["ping", "-c", "3", address])
class TestCore:
def test_import_service(self, core):
def test_import_service(self):
"""
Test importing a custom service.
@ -58,118 +63,115 @@ class TestCore:
assert ServiceManager.get("MyService")
assert ServiceManager.get("MyService2")
@pytest.mark.parametrize("cls", _NODE_CLASSES)
def test_nodes(self, core, cls):
@pytest.mark.parametrize("net_type", _WIRED)
def test_wired_ping(self, session, net_type, ip_prefixes):
"""
Test ptp node network.
:param conftest.Core core: core fixture to test with
:param cls: node classes that work within a simple network
:param session: session for test
:param core.enumerations.NodeTypes net_type: type of net node to create
:param ip_prefixes: generates ip addresses for nodes
"""
# create ptp
network_node = core.session.add_object(cls=cls)
# create net node
net_node = session.add_node(_type=net_type)
# create nodes
core.create_node("n1")
core.create_node("n2")
node_one = session.add_node()
node_two = session.add_node()
# add interfaces
core.add_interface(network_node, "n1")
core.add_interface(network_node, "n2")
# link nodes to net node
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, net_node.objid, interface_one=interface)
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = ping(node_one, node_two, ip_prefixes)
assert not status
@pytest.mark.parametrize("version", _XML_VERSIONS)
def test_xml(self, core, tmpdir, version):
def test_xml(self, session, tmpdir, version, ip_prefixes):
"""
Test xml client methods.
:param conftest.Core core: core fixture to test with
:param session: session for test
:param tmpdir: tmpdir to create data in
:param str version: xml version to write and parse
:param ip_prefixes: generates ip addresses for nodes
"""
# create ptp
ptp_node = core.session.add_object(cls=nodes.PtpNet)
ptp_node = session.add_node(_type=NodeTypes.PEER_TO_PEER)
# create nodes
core.create_node("n1")
core.create_node("n2")
node_one = session.add_node()
node_two = session.add_node()
# add interfaces
core.add_interface(ptp_node, "n1")
core.add_interface(ptp_node, "n2")
# link nodes to ptp net
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, ptp_node.objid, interface_one=interface)
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# get ids for nodes
n1_id = core.get_node("n1").objid
n2_id = core.get_node("n2").objid
n1_id = node_one.objid
n2_id = node_two.objid
# save xml
xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath
xmlsession.save_session_xml(core.session, file_path, version)
session.save_xml(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()
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
assert not core.session.get_object_by_name(n1_id)
assert not session.get_object(n1_id)
with pytest.raises(KeyError):
assert not core.session.get_object(n2_id)
assert not session.get_object(n2_id)
# load saved xml
xmlsession.open_session_xml(core.session, file_path, start=True)
session.open_xml(file_path, start=True)
# verify nodes have been recreated
assert core.session.get_object(n1_id)
assert core.session.get_object(n2_id)
assert session.get_object(n1_id)
assert session.get_object(n2_id)
def test_vnode_client(self, core):
def test_vnode_client(self, session, ip_prefixes):
"""
Test vnode client methods.
:param conftest.Core core: core fixture to test with
:param session: session for test
:param ip_prefixes: generates ip addresses for nodes
"""
# create ptp
ptp_node = core.session.add_object(cls=nodes.PtpNet)
ptp_node = session.add_node(_type=NodeTypes.PEER_TO_PEER)
# create nodes
core.create_node("n1")
core.create_node("n2")
node_one = session.add_node()
node_two = session.add_node()
# add interfaces
core.add_interface(ptp_node, "n1")
core.add_interface(ptp_node, "n2")
# link nodes to ptp net
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, ptp_node.objid, interface_one=interface)
# get node client for testing
n1 = core.get_node("n1")
client = n1.client
client = node_one.client
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# check we are connected
assert client.connected()
@ -195,183 +197,154 @@ class TestCore:
assert not client.shcmd(command[0])
# check module methods
assert createclients(core.session.session_dir)
assert createclients(session.session_dir)
# check convenience methods for interface information
assert client.getaddr("eth0")
assert client.netifstats()
def test_netif(self, core):
def test_netif(self, session, ip_prefixes):
"""
Test netif methods.
:param conftest.Core core: core fixture to test with
:param session: session for test
:param ip_prefixes: generates ip addresses for nodes
"""
# create ptp
ptp_node = core.session.add_object(cls=nodes.PtpNet)
ptp_node = session.add_node(_type=NodeTypes.PEER_TO_PEER)
# create nodes
core.create_node("n1")
core.create_node("n2")
node_one = session.add_node()
node_two = session.add_node()
# 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")
# link nodes to ptp net
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, ptp_node.objid, interface_one=interface)
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# 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)
assert node_one.commonnets(node_two)
assert node_two.commonnets(node_one)
# check we can retrieve netif index
assert n1.getifindex(n1_interface) == 0
assert n2.getifindex(n2_interface) == 0
assert node_one.getifindex(0)
assert node_two.getifindex(0)
# check interface parameters
n1_interface.setparam("test", 1)
assert n1_interface.getparam("test") == 1
assert n1_interface.getparams()
interface = node_one.netif(0)
interface.setparam("test", 1)
assert interface.getparam("test") == 1
assert interface.getparams()
# delete netif and test that if no longer exists
n1.delnetif(0)
assert not n1.netif(0)
node_one.delnetif(0)
assert not node_one.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):
def test_wlan_good(self, session, ip_prefixes):
"""
Test basic wlan network.
:param conftest.Core core: core fixture to test with
:param core.future.coreemu.FutureSession session: session for test
:param ip_prefixes: generates ip addresses for nodes
"""
# create wlan
wlan_node = core.session.add_object(cls=nodes.WlanNode)
values = BasicRangeModel.getdefaultvalues()
wlan_node.setmodel(BasicRangeModel, values)
wlan_node = session.add_node(_type=NodeTypes.WIRELESS_LAN)
session.set_wireless_model(wlan_node, BasicRangeModel)
# 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")
node_options = NodeOptions()
node_options.set_position(0, 0)
node_one = session.create_wireless_node(node_options=node_options)
node_two = session.create_wireless_node(node_options=node_options)
# add interfaces
interface_one = core.add_interface(wlan_node, "n1")
interface_two = core.add_interface(wlan_node, "n2")
# link nodes
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, wlan_node.objid, interface_one=interface)
# link nodes in wlan
core.link(wlan_node, interface_one, interface_two)
session.wireless_link_all(wlan_node, [node_one, node_two])
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = ping(node_one, node_two, ip_prefixes)
assert not status
def test_wlan_basic_range_bad(self, core):
def test_wlan_bad(self, session, ip_prefixes):
"""
Test basic wlan network with leveraging basic range model.
:param conftest.Core core: core fixture to test with
:param core.future.coreemu.FutureSession session: session for test
:param ip_prefixes: generates ip addresses for nodes
"""
# create wlan
wlan_node = core.session.add_object(cls=nodes.WlanNode)
values = BasicRangeModel.getdefaultvalues()
wlan_node.setmodel(BasicRangeModel, values)
wlan_node = session.add_node(_type=NodeTypes.WIRELESS_LAN)
session.set_wireless_model(wlan_node, BasicRangeModel)
# 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")
node_options = NodeOptions()
node_options.set_position(0, 0)
node_one = session.create_wireless_node(node_options=node_options)
node_two = session.create_wireless_node(node_options=node_options)
# add interfaces
interface_one = core.add_interface(wlan_node, "n1")
interface_two = core.add_interface(wlan_node, "n2")
# link nodes
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, wlan_node.objid, interface_one=interface)
# 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)
session.wireless_link_all(wlan_node, [node_one, node_two])
# instantiate session
core.session.instantiate()
session.instantiate()
# assert node directories created
core.assert_nodes()
# move node two out of range, default range check is 275
time.sleep(5)
update_options = NodeOptions()
update_options.set_position(500, 500)
session.update_node(node_two.objid, update_options)
# ping n2 from n1 and assert failure )
time.sleep(3)
status = core.ping("n1", "n2")
# ping n2 from n1 and assert failure
time.sleep(5)
status = ping(node_one, node_two, ip_prefixes)
assert status
def test_mobility(self, core):
def test_mobility(self, session, ip_prefixes):
"""
Test basic wlan network.
:param conftest.Core core: core fixture to test with
:param core.future.coreemu.FutureSession session: session for test
:param ip_prefixes: generates ip addresses for nodes
"""
# create wlan
wlan_node = core.session.add_object(cls=nodes.WlanNode)
values = BasicRangeModel.getdefaultvalues()
wlan_node.setmodel(BasicRangeModel, values)
wlan_node = session.add_node(_type=NodeTypes.WIRELESS_LAN)
session.set_wireless_model(wlan_node, BasicRangeModel)
# 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")
node_options = NodeOptions()
node_options.set_position(0, 0)
node_one = session.create_wireless_node(node_options=node_options)
node_two = session.create_wireless_node(node_options=node_options)
# add interfaces
interface_one = core.add_interface(wlan_node, "n1")
interface_two = core.add_interface(wlan_node, "n2")
# link nodes
for node in [node_one, node_two]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.objid, wlan_node.objid, interface_one=interface)
# link nodes in wlan
core.link(wlan_node, interface_one, interface_two)
session.wireless_link_all(wlan_node, [node_one, node_two])
# configure mobility script for session
config = ConfigData(
@ -382,7 +355,7 @@ class TestCore:
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)
session.config_object(config)
# add handler for receiving node updates
event = threading.Event()
@ -390,138 +363,10 @@ class TestCore:
def node_update(_):
event.set()
core.session.node_handlers.append(node_update)
session.node_handlers.append(node_update)
# instantiate session
core.session.instantiate()
# assert node directories created
core.assert_nodes()
session.instantiate()
# 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