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

moved emane dependent tests into subdir, moved pytest config into setup.cfg, added CoreError and made use of in session.py

Browse source
This commit is contained in:
bharnden 2019-09-11 13:12:42 -07:00
parent 4598bddcdb
commit 3776e1dda7
14 changed files with 208 additions and 201 deletions
Download patch file Download diff file Expand all files Collapse all files

138
daemon/tests/emane/test_emane.py Normal file
View file

@ -0,0 +1,138 @@
"""
Unit tests for testing CORE EMANE networks.
"""
import os
from xml.etree import ElementTree
import pytest
from conftest import ping
from core import CoreError
from core.emane.bypass import EmaneBypassModel
from core.emane.commeffect import EmaneCommEffectModel
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emane.rfpipe import EmaneRfPipeModel
from core.emane.tdma import EmaneTdmaModel
from core.emulator.emudata import NodeOptions
_EMANE_MODELS = [
EmaneIeee80211abgModel,
EmaneRfPipeModel,
EmaneBypassModel,
EmaneCommEffectModel,
EmaneTdmaModel,
]
_DIR = os.path.dirname(os.path.abspath(__file__))
class TestEmane:
@pytest.mark.parametrize("model", _EMANE_MODELS)
def test_models(self, session, model, ip_prefixes):
"""
Test emane models within a basic network.
:param core.emulator.coreemu.EmuSession session: session for test
:param model: emane model to test
:param ip_prefixes: generates ip addresses for nodes
"""
# create emane node for networking the core nodes
emane_network = session.create_emane_network(
model, geo_reference=(47.57917, -122.13232, 2.00000)
)
emane_network.setposition(x=80, y=50)
# configure tdma
if model == EmaneTdmaModel:
session.emane.set_model_config(
emane_network.id,
EmaneTdmaModel.name,
{"schedule": os.path.join(_DIR, "../examples/tdma/schedule.xml")},
)
# create nodes
node_options = NodeOptions()
node_options.set_position(150, 150)
node_one = session.create_wireless_node(node_options=node_options)
node_options.set_position(300, 150)
node_two = session.create_wireless_node(node_options=node_options)
for i, node in enumerate([node_one, node_two]):
node.setposition(x=150 * (i + 1), y=150)
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, emane_network.id, interface_one=interface)
# instantiate session
session.instantiate()
# ping n2 from n1 and assert success
status = ping(node_one, node_two, ip_prefixes, count=5)
assert not status
def test_xml_emane(self, session, tmpdir, ip_prefixes):
"""
Test xml client methods for emane.
:param session: session for test
:param tmpdir: tmpdir to create data in
:param ip_prefixes: generates ip addresses for nodes
"""
# create emane node for networking the core nodes
emane_network = session.create_emane_network(
EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000),
config={"test": "1"},
)
emane_network.setposition(x=80, y=50)
# create nodes
node_options = NodeOptions()
node_options.set_position(150, 150)
node_one = session.create_wireless_node(node_options=node_options)
node_options.set_position(300, 150)
node_two = session.create_wireless_node(node_options=node_options)
for i, node in enumerate([node_one, node_two]):
node.setposition(x=150 * (i + 1), y=150)
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, emane_network.id, interface_one=interface)
# instantiate session
session.instantiate()
# get ids for nodes
emane_id = emane_network.id
n1_id = node_one.id
n2_id = node_two.id
# save xml
xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath
session.save_xml(file_path)
# verify xml file was created and can be parsed
assert xml_file.isfile()
assert ElementTree.parse(file_path)
# stop current session, clearing data
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
session.open_xml(file_path, start=True)
# retrieve configuration we set originally
value = str(
session.emane.get_config("test", emane_id, EmaneIeee80211abgModel.name)
)
# verify nodes and configuration were restored
assert session.get_node(n1_id)
assert session.get_node(n2_id)
assert session.get_node(emane_id)
assert value == "1"

2
daemon/tests/pytest.ini
View file

@ -1,2 +0,0 @@
[pytest]
norecursedirs = distributed

68
daemon/tests/test_emane.py
View file

@ -1,68 +0,0 @@
"""
Unit tests for testing CORE EMANE networks.
"""
import os
import pytest
from conftest import ping
from core.emane.bypass import EmaneBypassModel
from core.emane.commeffect import EmaneCommEffectModel
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emane.rfpipe import EmaneRfPipeModel
from core.emane.tdma import EmaneTdmaModel
from core.emulator.emudata import NodeOptions
_EMANE_MODELS = [
EmaneIeee80211abgModel,
EmaneRfPipeModel,
EmaneBypassModel,
EmaneCommEffectModel,
EmaneTdmaModel,
]
_DIR = os.path.dirname(os.path.abspath(__file__))
class TestEmane:
@pytest.mark.parametrize("model", _EMANE_MODELS)
def test_models(self, session, model, ip_prefixes):
"""
Test emane models within a basic network.
:param core.emulator.coreemu.EmuSession session: session for test
:param model: emane model to test
:param ip_prefixes: generates ip addresses for nodes
"""
# create emane node for networking the core nodes
emane_network = session.create_emane_network(
model, geo_reference=(47.57917, -122.13232, 2.00000)
)
emane_network.setposition(x=80, y=50)
# configure tdma
if model == EmaneTdmaModel:
session.emane.set_model_config(
emane_network.id,
EmaneTdmaModel.name,
{"schedule": os.path.join(_DIR, "../examples/tdma/schedule.xml")},
)
# create nodes
node_options = NodeOptions()
node_options.set_position(150, 150)
node_one = session.create_wireless_node(node_options=node_options)
node_options.set_position(300, 150)
node_two = session.create_wireless_node(node_options=node_options)
for i, node in enumerate([node_one, node_two]):
node.setposition(x=150 * (i + 1), y=150)
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, emane_network.id, interface_one=interface)
# instantiate session
session.instantiate()
# ping n2 from n1 and assert success
status = ping(node_one, node_two, ip_prefixes, count=5)
assert not status

3
daemon/tests/test_grpc.py
View file

@ -5,6 +5,7 @@ from queue import Queue
import grpc
import pytest
from core import CoreError
from core.api.grpc import core_pb2
from core.api.grpc.client import CoreGrpcClient
from core.config import ConfigShim
@ -239,7 +240,7 @@ class TestGrpc:
# then
assert response.result is expected
if expected is True:
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert session.get_node(node.id)
def test_node_command(self, grpc_server):

3
daemon/tests/test_gui.py
View file

@ -7,6 +7,7 @@ import time
import mock
import pytest
from core import CoreError
from core.api.tlv import coreapi
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emulator.enumerations import (
@ -92,7 +93,7 @@ class TestGui:
coreserver.request_handler.handle_message(message)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
coreserver.session.get_node(node_id)
def test_link_add_node_to_net(self, coreserver):

4
daemon/tests/test_nodes.py
View file

@ -3,7 +3,7 @@ import time
import pytest
from core import utils
from core import CoreError, utils
from core.emulator.emudata import NodeOptions
from core.emulator.enumerations import NodeTypes
@ -53,7 +53,7 @@ class TestNodes:
session.delete_node(node.id)
# then
with pytest.raises(KeyError):
with pytest.raises(CoreError):
session.get_node(node.id)
@pytest.mark.parametrize("net_type", NET_TYPES)

98
daemon/tests/test_xml.py
View file

@ -2,7 +2,7 @@ from xml.etree import ElementTree
import pytest
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core import CoreError
from core.emulator.emudata import LinkOptions, NodeOptions
from core.emulator.enumerations import NodeTypes
from core.location.mobility import BasicRangeModel
@ -84,9 +84,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -145,9 +145,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -205,9 +205,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -222,74 +222,6 @@ class TestXml:
assert session.get_node(wlan_id)
assert value == "1"
def test_xml_emane(self, session, tmpdir, ip_prefixes):
"""
Test xml client methods for emane.
:param session: session for test
:param tmpdir: tmpdir to create data in
:param ip_prefixes: generates ip addresses for nodes
"""
# create emane node for networking the core nodes
emane_network = session.create_emane_network(
EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000),
config={"test": "1"},
)
emane_network.setposition(x=80, y=50)
# create nodes
node_options = NodeOptions()
node_options.set_position(150, 150)
node_one = session.create_wireless_node(node_options=node_options)
node_options.set_position(300, 150)
node_two = session.create_wireless_node(node_options=node_options)
for i, node in enumerate([node_one, node_two]):
node.setposition(x=150 * (i + 1), y=150)
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, emane_network.id, interface_one=interface)
# instantiate session
session.instantiate()
# get ids for nodes
emane_id = emane_network.id
n1_id = node_one.id
n2_id = node_two.id
# save xml
xml_file = tmpdir.join("session.xml")
file_path = xml_file.strpath
session.save_xml(file_path)
# verify xml file was created and can be parsed
assert xml_file.isfile()
assert ElementTree.parse(file_path)
# stop current session, clearing data
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
assert not session.get_node(n2_id)
# load saved xml
session.open_xml(file_path, start=True)
# retrieve configuration we set originally
value = str(
session.emane.get_config("test", emane_id, EmaneIeee80211abgModel.name)
)
# verify nodes and configuration were restored
assert session.get_node(n1_id)
assert session.get_node(n2_id)
assert session.get_node(emane_id)
assert value == "1"
def test_network_to_network(self, session, tmpdir):
"""
Test xml generation when dealing with network to network nodes.
@ -324,9 +256,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -383,9 +315,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -450,9 +382,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml
@ -532,9 +464,9 @@ class TestXml:
session.shutdown()
# verify nodes have been removed from session
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n1_id)
with pytest.raises(KeyError):
with pytest.raises(CoreError):
assert not session.get_node(n2_id)
# load saved xml