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

removed unused imports in emane files, refactored emane testing to simplify test cases and reduce repeated code

Browse source
This commit is contained in:
Blake J. Harnden 2017-07-10 10:44:10 -07:00
parent dced47b588
commit 673fdc94b5
9 changed files with 99 additions and 173 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
daemon/core/emane/commeffect.py
View file

@ -9,11 +9,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
except ImportError:
logger.error("error importing emanesh")
try:
import emaneeventservice
import emaneeventcommeffect
@ -99,8 +94,7 @@ class EmaneCommEffectModel(EmaneModel):
link parameters.
"""
if emane.VERSION >= emane.EMANE091:
raise NotImplementedError, \
"CommEffect linkconfig() not implemented for EMANE 0.9.1+"
raise NotImplementedError("CommEffect linkconfig() not implemented for EMANE 0.9.1+")
def z(x):
"""

7
daemon/core/emane/ieee80211abg.py
View file

@ -10,11 +10,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
except:
logger.error("error importing emanesh")
class EmaneIeee80211abgModel(EmaneModel):
def __init__(self, session, object_id=None):
@ -33,7 +28,7 @@ class EmaneIeee80211abgModel(EmaneModel):
# MAC parameters
_confmatrix_mac_base = [
("mode", ConfigDataTypes.UINT8.value, '0',
'0 802.11b (DSSS only),1 802.11b (DSSS only),' + \
'0 802.11b (DSSS only),1 802.11b (DSSS only),' +
'2 802.11a or g (OFDM),3 802.11b/g (DSSS and OFDM)', 'mode'),
("enablepromiscuousmode", ConfigDataTypes.BOOL.value, '0',
'On,Off', 'enable promiscuous mode'),

1
daemon/core/emane/nodes.py
View file

@ -16,7 +16,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
from emanesh.events import LocationEvent
except ImportError:
logger.error("error loading emanesh")

5
daemon/core/emane/rfpipe.py
View file

@ -10,11 +10,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
except ImportError:
logger.error("error importing emanesh")
class EmaneRfPipeModel(EmaneModel):
def __init__(self, session, object_id=None):

5
daemon/core/emane/tdma.py
View file

@ -10,11 +10,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
except:
logger.error("error importing emanesh")
class EmaneTdmaModel(EmaneModel):
def __init__(self, session, object_id=None):

5
daemon/core/emane/universal.py
View file

@ -10,11 +10,6 @@ from core.misc import log
logger = log.get_logger(__name__)
try:
from emanesh.events import EventService
except ImportError:
logger.error("error importing emanesh")
class EmaneUniversalModel(EmaneModel):
"""

16
daemon/core/session.py
View file

@ -1054,16 +1054,14 @@ class Session(object):
updown_script = None
if net_index == 0:
try:
if self.config["controlnet_updown_script"]:
updown_script = self.config["controlnet_updown_script"]
except KeyError:
logger.exception("error retreiving controlnet updown script")
updown_script = self.config.get("controlnet_updown_script")
if not updown_script:
logger.warning("controlnet updown script not configured")
# Check if session option set, overwrite if so
new_updown_script = getattr(self.options, "controlnet_updown_script", None)
if new_updown_script:
updown_script = new_updown_script
# check if session option set, overwrite if so
options_updown_script = getattr(self.options, "controlnet_updown_script", None)
if options_updown_script:
updown_script = options_updown_script
prefixes = prefix_spec.split()
if len(prefixes) > 1:

49
daemon/tests/conftest.py
View file

@ -9,6 +9,8 @@ from core.misc import ipaddress
from core.misc import nodemaps
from core.misc import nodeutils
from core.netns import nodes
from core.services import quagga
from core.services import utility
class Core(object):
@ -19,8 +21,12 @@ class Core(object):
self.nodes = {}
self.node_ips = {}
def create_node(self, name, cls=nodes.CoreNode, objid=None):
def create_node(self, name, cls=nodes.CoreNode, objid=None, position=None, services=None):
node = self.session.add_object(cls=cls, name=name, objid=objid)
if position:
node.setposition(*position)
if services:
self.session.services.addservicestonode(node, "", services)
self.nodes[name] = node
def add_interface(self, network, name):
@ -105,6 +111,11 @@ class Core(object):
return ptp_node, interface_one, interface_two
def set_emane_model(self, emane_node, emane_model):
# set the emane model
values = emane_model.getdefaultvalues()
self.session.emane.setconfig(emane_node.objid, emane_model.name, values)
@pytest.fixture()
def session():
@ -123,6 +134,37 @@ def session():
assert not os.path.exists(session_fixture.session_dir)
@pytest.fixture()
def session_emane():
# configure default nodes
node_map = nodemaps.CLASSIC_NODES
nodeutils.set_node_map(node_map)
# create and return session
session_fixture = Session(1, persistent=True)
assert os.path.exists(session_fixture.session_dir)
# load emane services
quagga.load_services()
utility.load_services()
# set location
session_fixture.master = True
session_fixture.location.setrefgeo(47.57917, -122.13232, 2.00000)
session_fixture.location.refscale = 150.0
# load emane models
session_fixture.emane.loadmodels()
# return session fixture
yield session_fixture
# cleanup
print "shutting down session"
session_fixture.shutdown()
assert not os.path.exists(session_fixture.session_dir)
@pytest.fixture(scope="module")
def ip_prefix():
return ipaddress.Ipv4Prefix("10.83.0.0/16")
@ -131,3 +173,8 @@ def ip_prefix():
@pytest.fixture()
def core(session, ip_prefix):
return Core(session, ip_prefix)
@pytest.fixture()
def core_emane(session_emane, ip_prefix):
return Core(session_emane, ip_prefix)

176
daemon/tests/test_emane.py
View file

@ -7,227 +7,135 @@ from core.emane.commeffect import EmaneCommEffectModel
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emane.nodes import EmaneNode
from core.emane.rfpipe import EmaneRfPipeModel
from core.services import quagga
from core.services import utility
_EMANE_SERVICES = "zebra|OSPFv3MDR|IPForward"
class TestGui:
def test_80211(self, core):
def test_80211(self, core_emane):
"""
Test emane 80211 model.
:param conftest.Core core: core fixture to test with
"""
# load services
quagga.load_services()
utility.load_services()
# set and load emane models
core.session.master = True
core.session.location.setrefgeo(47.57917, -122.13232, 2.00000)
core.session.location.refscale = 150.0
core.session.emane.loadmodels()
# create emane node for networking the core nodes
emane_node = core.session.add_object(name="emane", cls=EmaneNode)
emane_node = core_emane.session.add_object(name="emane", cls=EmaneNode)
emane_node.setposition(x=80, y=50)
# set the emane model
emane_model = EmaneIeee80211abgModel
values = emane_model.getdefaultvalues()
core.session.emane.setconfig(emane_node.objid, emane_model.name, values)
core_emane.set_emane_model(emane_node, EmaneIeee80211abgModel)
# create nodes
core.create_node("n1", objid=1)
core.create_node("n2", objid=2)
node_one = core.get_node("n1")
node_two = core.get_node("n2")
# set node positions
node_one.setposition(x=150, y=150)
node_two.setposition(x=300, y=150)
# add services
services = "zebra|OSPFv3MDR|IPForward"
core.session.services.addservicestonode(node_one, "", services)
core.session.services.addservicestonode(node_two, "", services)
core_emane.create_node("n1", objid=1, position=(150, 150), services=_EMANE_SERVICES)
core_emane.create_node("n2", objid=2, position=(300, 150), services=_EMANE_SERVICES)
# add interfaces to nodes
core.add_interface(emane_node, "n1")
core.add_interface(emane_node, "n2")
core_emane.add_interface(emane_node, "n1")
core_emane.add_interface(emane_node, "n2")
# instantiate session
core.session.instantiate()
core_emane.session.instantiate()
# assert node directories created
core.assert_nodes()
core_emane.assert_nodes()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = core_emane.ping("n1", "n2")
assert not status
def test_rfpipe(self, core):
def test_rfpipe(self, core_emane):
"""
Test emane 80211 model.
:param conftest.Core core: core fixture to test with
"""
# load services
quagga.load_services()
utility.load_services()
# set and load emane models
core.session.master = True
core.session.location.setrefgeo(47.57917, -122.13232, 2.00000)
core.session.location.refscale = 150.0
core.session.emane.loadmodels()
# create emane node for networking the core nodes
emane_node = core.session.add_object(name="emane", cls=EmaneNode)
emane_node = core_emane.session.add_object(name="emane", cls=EmaneNode)
emane_node.setposition(x=80, y=50)
# set the emane model
emane_model = EmaneRfPipeModel
values = emane_model.getdefaultvalues()
core.session.emane.setconfig(emane_node.objid, emane_model.name, values)
core_emane.set_emane_model(emane_node, EmaneRfPipeModel)
# create nodes
core.create_node("n1", objid=1)
core.create_node("n2", objid=2)
node_one = core.get_node("n1")
node_two = core.get_node("n2")
# set node positions
node_one.setposition(x=150, y=150)
node_two.setposition(x=300, y=150)
# add services
services = "zebra|OSPFv3MDR|IPForward"
core.session.services.addservicestonode(node_one, "", services)
core.session.services.addservicestonode(node_two, "", services)
core_emane.create_node("n1", objid=1, position=(150, 150), services=_EMANE_SERVICES)
core_emane.create_node("n2", objid=2, position=(300, 150), services=_EMANE_SERVICES)
# add interfaces to nodes
core.add_interface(emane_node, "n1")
core.add_interface(emane_node, "n2")
core_emane.add_interface(emane_node, "n1")
core_emane.add_interface(emane_node, "n2")
# instantiate session
core.session.instantiate()
core_emane.session.instantiate()
# assert node directories created
core.assert_nodes()
core_emane.assert_nodes()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = core_emane.ping("n1", "n2")
assert not status
def test_commeffect(self, core):
def test_commeffect(self, core_emane):
"""
Test emane 80211 model.
:param conftest.Core core: core fixture to test with
"""
# load services
quagga.load_services()
utility.load_services()
# set and load emane models
core.session.master = True
core.session.location.setrefgeo(47.57917, -122.13232, 2.00000)
core.session.location.refscale = 150.0
core.session.emane.loadmodels()
# create emane node for networking the core nodes
emane_node = core.session.add_object(name="emane", cls=EmaneNode)
emane_node = core_emane.session.add_object(name="emane", cls=EmaneNode)
emane_node.setposition(x=80, y=50)
# set the emane model
emane_model = EmaneCommEffectModel
values = emane_model.getdefaultvalues()
core.session.emane.setconfig(emane_node.objid, emane_model.name, values)
core_emane.set_emane_model(emane_node, EmaneCommEffectModel)
# create nodes
core.create_node("n1", objid=1)
core.create_node("n2", objid=2)
node_one = core.get_node("n1")
node_two = core.get_node("n2")
# set node positions
node_one.setposition(x=150, y=150)
node_two.setposition(x=300, y=150)
# add services
services = "zebra|OSPFv3MDR|IPForward"
core.session.services.addservicestonode(node_one, "", services)
core.session.services.addservicestonode(node_two, "", services)
core_emane.create_node("n1", objid=1, position=(150, 150), services=_EMANE_SERVICES)
core_emane.create_node("n2", objid=2, position=(300, 150), services=_EMANE_SERVICES)
# add interfaces to nodes
core.add_interface(emane_node, "n1")
core.add_interface(emane_node, "n2")
core_emane.add_interface(emane_node, "n1")
core_emane.add_interface(emane_node, "n2")
# instantiate session
core.session.instantiate()
core_emane.session.instantiate()
# assert node directories created
core.assert_nodes()
core_emane.assert_nodes()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = core_emane.ping("n1", "n2")
assert not status
def test_bypass(self, core):
def test_bypass(self, core_emane):
"""
Test emane 80211 model.
:param conftest.Core core: core fixture to test with
"""
# load services
quagga.load_services()
utility.load_services()
# set and load emane models
core.session.master = True
core.session.location.setrefgeo(47.57917, -122.13232, 2.00000)
core.session.location.refscale = 150.0
core.session.emane.loadmodels()
# create emane node for networking the core nodes
emane_node = core.session.add_object(name="emane", cls=EmaneNode)
emane_node = core_emane.session.add_object(name="emane", cls=EmaneNode)
emane_node.setposition(x=80, y=50)
# set the emane model
emane_model = EmaneBypassModel
values = emane_model.getdefaultvalues()
core.session.emane.setconfig(emane_node.objid, emane_model.name, values)
core_emane.set_emane_model(emane_node, EmaneBypassModel)
# create nodes
core.create_node("n1", objid=1)
core.create_node("n2", objid=2)
node_one = core.get_node("n1")
node_two = core.get_node("n2")
# set node positions
node_one.setposition(x=150, y=150)
node_two.setposition(x=300, y=150)
# add services
services = "zebra|OSPFv3MDR|IPForward"
core.session.services.addservicestonode(node_one, "", services)
core.session.services.addservicestonode(node_two, "", services)
core_emane.create_node("n1", objid=1, position=(150, 150), services=_EMANE_SERVICES)
core_emane.create_node("n2", objid=2, position=(300, 150), services=_EMANE_SERVICES)
# add interfaces to nodes
core.add_interface(emane_node, "n1")
core.add_interface(emane_node, "n2")
core_emane.add_interface(emane_node, "n1")
core_emane.add_interface(emane_node, "n2")
# instantiate session
core.session.instantiate()
core_emane.session.instantiate()
# assert node directories created
core.assert_nodes()
core_emane.assert_nodes()
# ping n2 from n1 and assert success
status = core.ping("n1", "n2")
status = core_emane.ping("n1", "n2")
assert not status