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

added flake8/black, pre-commit integration for flake8/black, and black formatting changes

Browse source
This commit is contained in:
bharnden 2019-09-10 15:10:24 -07:00
parent d5055f85d3
commit 1fc8d647c3
77 changed files with 4452 additions and 1964 deletions
Download patch file Download diff file Expand all files Collapse all files

64
daemon/tests/conftest.py
View file

@ -27,10 +27,9 @@ class CoreServerTest(object):
self.host = "localhost"
self.port = port
address = (self.host, self.port)
self.server = CoreServer(address, CoreHandler, {
"numthreads": 1,
"daemonize": False,
})
self.server = CoreServer(
address, CoreHandler, {"numthreads": 1, "daemonize": False}
)
self.distributed_server = "core2"
self.prefix = ipaddress.Ipv4Prefix("10.83.0.0/16")
@ -62,36 +61,51 @@ class CoreServerTest(object):
# have broker handle a configuration state change
self.session.set_state(EventTypes.DEFINITION_STATE)
message = CoreEventMessage.create(0, [(EventTlvs.TYPE, EventTypes.CONFIGURATION_STATE.value)])
message = CoreEventMessage.create(
0, [(EventTlvs.TYPE, EventTypes.CONFIGURATION_STATE.value)]
)
self.request_handler.handle_message(message)
# add broker server for distributed core
distributed = "%s:%s:%s" % (self.distributed_server, distributed_address, self.port)
message = CoreConfMessage.create(0, [
(ConfigTlvs.OBJECT, "broker"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (10,)),
(ConfigTlvs.VALUES, distributed)
])
distributed = "%s:%s:%s" % (
self.distributed_server,
distributed_address,
self.port,
)
message = CoreConfMessage.create(
0,
[
(ConfigTlvs.OBJECT, "broker"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (10,)),
(ConfigTlvs.VALUES, distributed),
],
)
self.request_handler.handle_message(message)
# set session location
message = CoreConfMessage.create(0, [
(ConfigTlvs.OBJECT, "location"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (9, 9, 9, 9, 9, 9)),
(ConfigTlvs.VALUES, "0|0| 47.5766974863|-122.125920191|0.0|150.0")
])
message = CoreConfMessage.create(
0,
[
(ConfigTlvs.OBJECT, "location"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (9, 9, 9, 9, 9, 9)),
(ConfigTlvs.VALUES, "0|0| 47.5766974863|-122.125920191|0.0|150.0"),
],
)
self.request_handler.handle_message(message)
# set services for host nodes
message = CoreConfMessage.create(0, [
(ConfigTlvs.SESSION, str(self.session.id)),
(ConfigTlvs.OBJECT, "services"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (10, 10, 10)),
(ConfigTlvs.VALUES, "host|DefaultRoute|SSH")
])
message = CoreConfMessage.create(
0,
[
(ConfigTlvs.SESSION, str(self.session.id)),
(ConfigTlvs.OBJECT, "services"),
(ConfigTlvs.TYPE, 0),
(ConfigTlvs.DATA_TYPES, (10, 10, 10)),
(ConfigTlvs.VALUES, "host|DefaultRoute|SSH"),
],
)
self.request_handler.handle_message(message)
def shutdown(self):

155
daemon/tests/distributed/test_distributed.py
View file

@ -25,14 +25,19 @@ from core.nodes.ipaddress import IpAddress, Ipv4Prefix, MacAddress
def set_emane_model(node_id, model):
return CoreConfMessage.create(0, [
(ConfigTlvs.NODE, node_id),
(ConfigTlvs.OBJECT, model),
(ConfigTlvs.TYPE, ConfigFlags.UPDATE.value),
])
return CoreConfMessage.create(
0,
[
(ConfigTlvs.NODE, node_id),
(ConfigTlvs.OBJECT, model),
(ConfigTlvs.TYPE, ConfigFlags.UPDATE.value),
],
)
def node_message(_id, name, emulation_server=None, node_type=NodeTypes.DEFAULT, model=None):
def node_message(
_id, name, emulation_server=None, node_type=NodeTypes.DEFAULT, model=None
):
"""
Convenience method for creating a node TLV messages.
@ -59,7 +64,16 @@ def node_message(_id, name, emulation_server=None, node_type=NodeTypes.DEFAULT,
return CoreNodeMessage.create(MessageFlags.ADD.value, values)
def link_message(n1, n2, intf_one=None, address_one=None, intf_two=None, address_two=None, key=None, mask=24):
def link_message(
n1,
n2,
intf_one=None,
address_one=None,
intf_two=None,
address_two=None,
key=None,
mask=24,
):
"""
Convenience method for creating link TLV messages.
@ -115,11 +129,14 @@ def command_message(node, command):
:rtype: core.api.tlv.coreapi.CoreExecMessage
"""
flags = MessageFlags.STRING.value | MessageFlags.TEXT.value
return CoreExecMessage.create(flags, [
(ExecuteTlvs.NODE, node.id),
(ExecuteTlvs.NUMBER, 1),
(ExecuteTlvs.COMMAND, command)
])
return CoreExecMessage.create(
flags,
[
(ExecuteTlvs.NODE, node.id),
(ExecuteTlvs.NUMBER, 1),
(ExecuteTlvs.COMMAND, command),
],
)
def state_message(state):
@ -130,9 +147,7 @@ def state_message(state):
:return: tlv message
:rtype: core.api.tlv.coreapi.CoreEventMessage
"""
return CoreEventMessage.create(0, [
(EventTlvs.TYPE, state.value)
])
return CoreEventMessage.create(0, [(EventTlvs.TYPE, state.value)])
def validate_response(replies, _):
@ -144,8 +159,8 @@ def validate_response(replies, _):
:return: nothing
"""
response = replies[0]
header = response[:CoreExecMessage.header_len]
tlv_data = response[CoreExecMessage.header_len:]
header = response[: CoreExecMessage.header_len]
tlv_data = response[CoreExecMessage.header_len :]
response = CoreExecMessage(MessageFlags.TEXT, header, tlv_data)
assert not response.get_tlv(ExecuteTlvs.STATUS.value)
@ -162,48 +177,27 @@ class TestDistributed:
cored.setup(distributed_address)
# create local node
message = node_message(
_id=1,
name="n1",
model="host"
)
message = node_message(_id=1, name="n1", model="host")
cored.request_handler.handle_message(message)
# create distributed node and assign to distributed server
message = node_message(
_id=2,
name="n2",
emulation_server=cored.distributed_server,
model="host"
_id=2, name="n2", emulation_server=cored.distributed_server, model="host"
)
cored.request_handler.handle_message(message)
# create distributed switch and assign to distributed server
message = node_message(
_id=3,
name="n3",
node_type=NodeTypes.SWITCH
)
message = node_message(_id=3, name="n3", node_type=NodeTypes.SWITCH)
cored.request_handler.handle_message(message)
# link message one
ip4_address = cored.prefix.addr(1)
message = link_message(
n1=1,
n2=3,
intf_one=0,
address_one=ip4_address
)
message = link_message(n1=1, n2=3, intf_one=0, address_one=ip4_address)
cored.request_handler.handle_message(message)
# link message two
ip4_address = cored.prefix.addr(2)
message = link_message(
n1=3,
n2=2,
intf_two=0,
address_two=ip4_address
)
message = link_message(n1=3, n2=2, intf_two=0, address_two=ip4_address)
cored.request_handler.handle_message(message)
# change session to instantiation state
@ -227,31 +221,22 @@ class TestDistributed:
cored.setup(distributed_address)
# configure required controlnet
cored.session.options.set_config("controlnet", "core1:172.16.1.0/24 core2:172.16.2.0/24")
cored.session.options.set_config(
"controlnet", "core1:172.16.1.0/24 core2:172.16.2.0/24"
)
# create local node
message = node_message(
_id=1,
name="n1",
model="mdr"
)
message = node_message(_id=1, name="n1", model="mdr")
cored.request_handler.handle_message(message)
# create distributed node and assign to distributed server
message = node_message(
_id=2,
name="n2",
emulation_server=cored.distributed_server,
model="mdr"
_id=2, name="n2", emulation_server=cored.distributed_server, model="mdr"
)
cored.request_handler.handle_message(message)
# create distributed switch and assign to distributed server
message = node_message(
_id=3,
name="n3",
node_type=NodeTypes.EMANE
)
message = node_message(_id=3, name="n3", node_type=NodeTypes.EMANE)
cored.request_handler.handle_message(message)
# set emane model
@ -260,24 +245,12 @@ class TestDistributed:
# link message one
ip4_address = cored.prefix.addr(1)
message = link_message(
n1=1,
n2=3,
intf_one=0,
address_one=ip4_address,
mask=32
)
message = link_message(n1=1, n2=3, intf_one=0, address_one=ip4_address, mask=32)
cored.request_handler.handle_message(message)
# link message two
ip4_address = cored.prefix.addr(2)
message = link_message(
n1=2,
n2=3,
intf_one=0,
address_one=ip4_address,
mask=32
)
message = link_message(n1=2, n2=3, intf_one=0, address_one=ip4_address, mask=32)
cored.request_handler.handle_message(message)
# change session to instantiation state
@ -301,11 +274,7 @@ class TestDistributed:
cored.setup(distributed_address)
# create local node
message = node_message(
_id=1,
name="n1",
model="host"
)
message = node_message(_id=1, name="n1", model="host")
cored.request_handler.handle_message(message)
# create distributed node and assign to distributed server
@ -314,36 +283,22 @@ class TestDistributed:
name="n2",
emulation_server=cored.distributed_server,
node_type=NodeTypes.PHYSICAL,
model="prouter"
model="prouter",
)
cored.request_handler.handle_message(message)
# create distributed switch and assign to distributed server
message = node_message(
_id=3,
name="n3",
node_type=NodeTypes.SWITCH
)
message = node_message(_id=3, name="n3", node_type=NodeTypes.SWITCH)
cored.request_handler.handle_message(message)
# link message one
ip4_address = cored.prefix.addr(1)
message = link_message(
n1=1,
n2=3,
intf_one=0,
address_one=ip4_address
)
message = link_message(n1=1, n2=3, intf_one=0, address_one=ip4_address)
cored.request_handler.handle_message(message)
# link message two
ip4_address = cored.prefix.addr(2)
message = link_message(
n1=3,
n2=2,
intf_two=0,
address_two=ip4_address
)
message = link_message(n1=3, n2=2, intf_two=0, address_two=ip4_address)
cored.request_handler.handle_message(message)
# change session to instantiation state
@ -368,11 +323,7 @@ class TestDistributed:
cored.setup(distributed_address)
# create local node
message = node_message(
_id=1,
name="n1",
model="host"
)
message = node_message(_id=1, name="n1", model="host")
cored.request_handler.handle_message(message)
# create distributed node and assign to distributed server
@ -380,7 +331,7 @@ class TestDistributed:
_id=2,
name=distributed_address,
emulation_server=cored.distributed_server,
node_type=NodeTypes.TUNNEL
node_type=NodeTypes.TUNNEL,
)
cored.request_handler.handle_message(message)
@ -394,7 +345,7 @@ class TestDistributed:
address_one=ip4_address,
intf_two=0,
address_two=address_two,
key=1
key=1,
)
cored.request_handler.handle_message(message)

19
daemon/tests/test_conf.py
View file

@ -1,5 +1,10 @@
import pytest
from core.config import ConfigurableManager, ConfigurableOptions, Configuration, ModelManager
from core.config import (
ConfigurableManager,
ConfigurableOptions,
Configuration,
ModelManager,
)
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emulator.enumerations import ConfigDataTypes, NodeTypes
from core.location.mobility import BasicRangeModel
@ -9,16 +14,8 @@ class TestConfigurableOptions(ConfigurableOptions):
name_one = "value1"
name_two = "value2"
options = [
Configuration(
_id=name_one,
_type=ConfigDataTypes.STRING,
label=name_one
),
Configuration(
_id=name_two,
_type=ConfigDataTypes.STRING,
label=name_two
)
Configuration(_id=name_one, _type=ConfigDataTypes.STRING, label=name_one),
Configuration(_id=name_two, _type=ConfigDataTypes.STRING, label=name_two),
]

10
daemon/tests/test_core.py
View file

@ -15,11 +15,7 @@ from core.nodes.client import VnodeClient
_PATH = os.path.abspath(os.path.dirname(__file__))
_MOBILITY_FILE = os.path.join(_PATH, "mobility.scen")
_WIRED = [
NodeTypes.PEER_TO_PEER,
NodeTypes.HUB,
NodeTypes.SWITCH
]
_WIRED = [NodeTypes.PEER_TO_PEER, NodeTypes.HUB, NodeTypes.SWITCH]
def createclients(sessiondir, clientcls=VnodeClient, cmdchnlfilterfunc=None):
@ -112,7 +108,9 @@ class TestCore:
p, stdin, stdout, stderr = client.popen(command)
assert not p.wait()
assert not client.icmd(command)
assert not client.redircmd(subprocess.PIPE, subprocess.PIPE, subprocess.PIPE, command)
assert not client.redircmd(
subprocess.PIPE, subprocess.PIPE, subprocess.PIPE, command
)
assert not client.shcmd(command[0])
# check various command using command line

11
daemon/tests/test_emane.py
View file

@ -35,16 +35,17 @@ class TestEmane:
# create emane node for networking the core nodes
emane_network = session.create_emane_network(
model,
geo_reference=(47.57917, -122.13232, 2.00000)
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")
})
session.emane.set_model_config(
emane_network.id,
EmaneTdmaModel.name,
{"schedule": os.path.join(_DIR, "../examples/tdma/schedule.xml")},
)
# create nodes
node_options = NodeOptions()

133
daemon/tests/test_grpc.py
View file

@ -11,7 +11,12 @@ from core.config import ConfigShim
from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emulator.data import EventData
from core.emulator.emudata import NodeOptions
from core.emulator.enumerations import ConfigFlags, EventTypes, ExceptionLevels, NodeTypes
from core.emulator.enumerations import (
ConfigFlags,
EventTypes,
ExceptionLevels,
NodeTypes,
)
from core.location.mobility import BasicRangeModel, Ns2ScriptedMobility
@ -35,10 +40,7 @@ class TestGrpc:
assert response.session_id == session_id
assert session.id == session_id
@pytest.mark.parametrize("session_id, expected", [
(None, True),
(6013, False)
])
@pytest.mark.parametrize("session_id, expected", [(None, True), (6013, False)])
def test_delete_session(self, grpc_server, session_id, expected):
# given
client = CoreGrpcClient()
@ -130,9 +132,13 @@ class TestGrpc:
with client.context_connect():
response = client.set_session_location(
session.id,
x=xyz[0], y=xyz[1], z=xyz[2],
lat=lat_lon_alt[0], lon=lat_lon_alt[1], alt=lat_lon_alt[2],
scale=scale
x=xyz[0],
y=xyz[1],
z=xyz[2],
lat=lat_lon_alt[0],
lon=lat_lon_alt[1],
alt=lat_lon_alt[2],
scale=scale,
)
# then
@ -165,7 +171,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_session_state(session.id, core_pb2.SessionState.DEFINITION)
response = client.set_session_state(
session.id, core_pb2.SessionState.DEFINITION
)
# then
assert response.result is True
@ -198,10 +206,7 @@ class TestGrpc:
# then
assert response.node.id == node.id
@pytest.mark.parametrize("node_id, expected", [
(1, True),
(2, False)
])
@pytest.mark.parametrize("node_id, expected", [(1, True), (2, False)])
def test_edit_node(self, grpc_server, node_id, expected):
# given
client = CoreGrpcClient()
@ -220,10 +225,7 @@ class TestGrpc:
assert node.position.x == x
assert node.position.y == y
@pytest.mark.parametrize("node_id, expected", [
(1, True),
(2, False)
])
@pytest.mark.parametrize("node_id, expected", [(1, True), (2, False)])
def test_delete_node(self, grpc_server, node_id, expected):
# given
client = CoreGrpcClient()
@ -302,7 +304,9 @@ class TestGrpc:
file_name = "test"
file_data = "echo hello"
with client.context_connect():
response = client.add_hook(session.id, core_pb2.SessionState.RUNTIME, file_name, file_data)
response = client.add_hook(
session.id, core_pb2.SessionState.RUNTIME, file_name, file_data
)
# then
assert response.result is True
@ -408,7 +412,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.edit_link(session.id, node.id, switch.id, options, interface_one_id=interface.id)
response = client.edit_link(
session.id, node.id, switch.id, options, interface_one_id=interface.id
)
# then
assert response.result is True
@ -435,7 +441,8 @@ class TestGrpc:
# then
with client.context_connect():
response = client.delete_link(
session.id, node_one.id, node_two.id, interface_one.id, interface_two.id)
session.id, node_one.id, node_two.id, interface_one.id, interface_two.id
)
# then
assert response.result is True
@ -467,14 +474,18 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_wlan_config(session.id, wlan.id, {
range_key: range_value,
"delay": "0",
"loss": "0",
"bandwidth": "50000",
"error": "0",
"jitter": "0"
})
response = client.set_wlan_config(
session.id,
wlan.id,
{
range_key: range_value,
"delay": "0",
"loss": "0",
"bandwidth": "50000",
"error": "0",
"jitter": "0",
},
)
# then
assert response.result is True
@ -516,12 +527,13 @@ class TestGrpc:
client = CoreGrpcClient()
session = grpc_server.coreemu.create_session()
emane_network = session.create_emane_network(
model=EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000)
model=EmaneIeee80211abgModel, geo_reference=(47.57917, -122.13232, 2.00000)
)
config_key = "platform_id_start"
config_value = "2"
session.emane.set_model_config(emane_network.id, EmaneIeee80211abgModel.name, {config_key: config_value})
session.emane.set_model_config(
emane_network.id, EmaneIeee80211abgModel.name, {config_key: config_value}
)
# then
with client.context_connect():
@ -536,8 +548,7 @@ class TestGrpc:
client = CoreGrpcClient()
session = grpc_server.coreemu.create_session()
emane_network = session.create_emane_network(
model=EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000)
model=EmaneIeee80211abgModel, geo_reference=(47.57917, -122.13232, 2.00000)
)
config_key = "bandwidth"
config_value = "900000"
@ -545,11 +556,17 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_emane_model_config(
session.id, emane_network.id, EmaneIeee80211abgModel.name, {config_key: config_value})
session.id,
emane_network.id,
EmaneIeee80211abgModel.name,
{config_key: config_value},
)
# then
assert response.result is True
config = session.emane.get_model_config(emane_network.id, EmaneIeee80211abgModel.name)
config = session.emane.get_model_config(
emane_network.id, EmaneIeee80211abgModel.name
)
assert config[config_key] == config_value
def test_get_emane_model_config(self, grpc_server):
@ -557,14 +574,14 @@ class TestGrpc:
client = CoreGrpcClient()
session = grpc_server.coreemu.create_session()
emane_network = session.create_emane_network(
model=EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000)
model=EmaneIeee80211abgModel, geo_reference=(47.57917, -122.13232, 2.00000)
)
# then
with client.context_connect():
response = client.get_emane_model_config(
session.id, emane_network.id, EmaneIeee80211abgModel.name)
session.id, emane_network.id, EmaneIeee80211abgModel.name
)
# then
assert len(response.groups) > 0
@ -620,7 +637,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_mobility_config(session.id, wlan.id, {config_key: config_value})
response = client.set_mobility_config(
session.id, wlan.id, {config_key: config_value}
)
# then
assert response.result is True
@ -637,7 +656,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.mobility_action(session.id, wlan.id, core_pb2.MobilityAction.STOP)
response = client.mobility_action(
session.id, wlan.id, core_pb2.MobilityAction.STOP
)
# then
assert response.result is True
@ -701,7 +722,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.get_node_service_file(session.id, node.id, "DefaultRoute", "defaultroute.sh")
response = client.get_node_service_file(
session.id, node.id, "DefaultRoute", "defaultroute.sh"
)
# then
assert response.data is not None
@ -716,11 +739,15 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_node_service(session.id, node.id, service_name, [], validate, [])
response = client.set_node_service(
session.id, node.id, service_name, [], validate, []
)
# then
assert response.result is True
service = session.services.get_service(node.id, service_name, default_service=True)
service = session.services.get_service(
node.id, service_name, default_service=True
)
assert service.validate == tuple(validate)
def test_set_node_service_file(self, grpc_server):
@ -734,7 +761,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.set_node_service_file(session.id, node.id, service_name, file_name, file_data)
response = client.set_node_service_file(
session.id, node.id, service_name, file_name, file_data
)
# then
assert response.result is True
@ -750,7 +779,9 @@ class TestGrpc:
# then
with client.context_connect():
response = client.service_action(session.id, node.id, service_name, core_pb2.ServiceAction.STOP)
response = client.service_action(
session.id, node.id, service_name, core_pb2.ServiceAction.STOP
)
# then
assert response.result is True
@ -831,7 +862,9 @@ class TestGrpc:
with client.context_connect():
client.events(session.id, handle_event)
time.sleep(0.1)
event = EventData(event_type=EventTypes.RUNTIME_STATE.value, time="%s" % time.time())
event = EventData(
event_type=EventTypes.RUNTIME_STATE.value, time="%s" % time.time()
)
session.broadcast_event(event)
# then
@ -852,7 +885,9 @@ class TestGrpc:
client.events(session.id, handle_event)
time.sleep(0.1)
session_config = session.options.get_configs()
config_data = ConfigShim.config_data(0, None, ConfigFlags.UPDATE.value, session.options, session_config)
config_data = ConfigShim.config_data(
0, None, ConfigFlags.UPDATE.value, session.options, session_config
)
session.broadcast_config(config_data)
# then
@ -892,7 +927,9 @@ class TestGrpc:
with client.context_connect():
client.events(session.id, handle_event)
time.sleep(0.1)
file_data = session.services.get_service_file(node, "DefaultRoute", "defaultroute.sh")
file_data = session.services.get_service_file(
node, "DefaultRoute", "defaultroute.sh"
)
session.broadcast_file(file_data)
# then

751
daemon/tests/test_gui.py
View file

File diff suppressed because it is too large Load diff

20
daemon/tests/test_links.py
View file

@ -105,8 +105,12 @@ class TestLinks:
link_options.bandwidth = 5000000
link_options.per = 25
link_options.dup = 25
session.update_link(node_one.id, node_two.id,
interface_one_id=interface_one.id, link_options=link_options)
session.update_link(
node_one.id,
node_two.id,
interface_one_id=interface_one.id,
link_options=link_options,
)
# then
output = utils.check_cmd(["tc", "qdisc", "show", "dev", interface.localname])
@ -126,7 +130,9 @@ class TestLinks:
assert node_two.netif(interface_two.id)
# when
session.delete_link(node_one.id, node_two.id, interface_one.id, interface_two.id)
session.delete_link(
node_one.id, node_two.id, interface_one.id, interface_two.id
)
# then
assert not node_one.netif(interface_one.id)
@ -149,7 +155,7 @@ class TestLinks:
# run iperf, validate normal bandwidth
stdout = iperf(node_one, node_two, ip_prefixes)
assert stdout
value = int(stdout.split(',')[bandwidth_index])
value = int(stdout.split(",")[bandwidth_index])
assert 900000 <= value <= 1100000
# change bandwidth in bits per second
@ -160,7 +166,7 @@ class TestLinks:
# run iperf again
stdout = iperf(node_one, node_two, ip_prefixes)
assert stdout
value = int(stdout.split(',')[bandwidth_index])
value = int(stdout.split(",")[bandwidth_index])
assert 400000 <= value <= 600000
def test_link_loss(self, session, ip_prefixes):
@ -180,7 +186,7 @@ class TestLinks:
# run iperf, validate normal bandwidth
stdout = iperf(node_one, node_two, ip_prefixes)
assert stdout
value = float(stdout.split(',')[loss_index])
value = float(stdout.split(",")[loss_index])
assert 0 <= value <= 0.5
# change bandwidth in bits per second
@ -191,7 +197,7 @@ class TestLinks:
# run iperf again
stdout = iperf(node_one, node_two, ip_prefixes)
assert stdout
value = float(stdout.split(',')[loss_index])
value = float(stdout.split(",")[loss_index])
assert 40 <= value <= 60
def test_link_delay(self, session, ip_prefixes):

13
daemon/tests/test_nodes.py
View file

@ -6,18 +6,9 @@ from core import utils
from core.emulator.emudata import NodeOptions
from core.emulator.enumerations import NodeTypes
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]
class TestNodes:

34
daemon/tests/test_services.py
View file

@ -202,8 +202,12 @@ class TestServices:
file_name = my_service.configs[0]
file_data_one = "# custom file one"
file_data_two = "# custom file two"
session.services.set_service_file(node_one.id, my_service.name, file_name, file_data_one)
session.services.set_service_file(node_two.id, my_service.name, file_name, file_data_two)
session.services.set_service_file(
node_one.id, my_service.name, file_name, file_data_one
)
session.services.set_service_file(
node_two.id, my_service.name, file_name, file_data_two
)
# when
custom_service_one = session.services.get_service(node_one.id, my_service.name)
@ -238,9 +242,13 @@ class TestServices:
# when
no_service = session.services.get_service(node.id, SERVICE_ONE)
default_service = session.services.get_service(node.id, SERVICE_ONE, default_service=True)
default_service = session.services.get_service(
node.id, SERVICE_ONE, default_service=True
)
session.services.set_service(node.id, SERVICE_ONE)
custom_service = session.services.get_service(node.id, SERVICE_ONE, default_service=True)
custom_service = session.services.get_service(
node.id, SERVICE_ONE, default_service=True
)
# then
assert no_service is None
@ -249,13 +257,7 @@ class TestServices:
def test_services_dependencies(self):
# given
services = [
ServiceA,
ServiceB,
ServiceC,
ServiceD,
ServiceF
]
services = [ServiceA, ServiceB, ServiceC, ServiceD, ServiceF]
# when
boot_paths = ServiceDependencies(services).boot_paths()
@ -271,7 +273,7 @@ class TestServices:
ServiceC,
ServiceD,
ServiceF,
ServiceBadDependency
ServiceBadDependency,
]
# when, then
@ -282,13 +284,7 @@ class TestServices:
# given
service_d = ServiceD()
service_d.dependencies = ("C",)
services = [
ServiceA,
ServiceB,
ServiceC,
service_d,
ServiceF
]
services = [ServiceA, ServiceB, ServiceC, service_d, ServiceF]
# when, then
with pytest.raises(ValueError):

30
daemon/tests/test_xml.py
View file

@ -120,7 +120,9 @@ class TestXml:
session.services.set_service(node_one.id, SshService.name)
service_file = SshService.configs[0]
file_data = "# test"
session.services.set_service_file(node_one.id, SshService.name, service_file, file_data)
session.services.set_service_file(
node_one.id, SshService.name, service_file, file_data
)
# instantiate session
session.instantiate()
@ -231,7 +233,7 @@ class TestXml:
emane_network = session.create_emane_network(
EmaneIeee80211abgModel,
geo_reference=(47.57917, -122.13232, 2.00000),
config={"test": "1"}
config={"test": "1"},
)
emane_network.setposition(x=80, y=50)
@ -277,7 +279,9 @@ class TestXml:
session.open_xml(file_path, start=True)
# retrieve configuration we set originally
value = str(session.emane.get_config("test", emane_id, EmaneIeee80211abgModel.name))
value = str(
session.emane.get_config("test", emane_id, EmaneIeee80211abgModel.name)
)
# verify nodes and configuration were restored
assert session.get_node(n1_id)
@ -354,7 +358,9 @@ class TestXml:
link_options.jitter = 10
link_options.delay = 30
link_options.dup = 5
session.add_link(node_one.id, switch.id, interface_one, link_options=link_options)
session.add_link(
node_one.id, switch.id, interface_one, link_options=link_options
)
# instantiate session
session.instantiate()
@ -419,7 +425,9 @@ class TestXml:
link_options.jitter = 10
link_options.delay = 30
link_options.dup = 5
session.add_link(node_one.id, node_two.id, interface_one, interface_two, link_options)
session.add_link(
node_one.id, node_two.id, interface_one, interface_two, link_options
)
# instantiate session
session.instantiate()
@ -485,7 +493,9 @@ class TestXml:
link_options_one.per = 10.5
link_options_one.dup = 5
link_options_one.jitter = 5
session.add_link(node_one.id, node_two.id, interface_one, interface_two, link_options_one)
session.add_link(
node_one.id, node_two.id, interface_one, interface_two, link_options_one
)
link_options_two = LinkOptions()
link_options_two.unidirectional = 1
link_options_two.bandwidth = 10000
@ -493,7 +503,13 @@ class TestXml:
link_options_two.per = 10
link_options_two.dup = 10
link_options_two.jitter = 10
session.update_link(node_two.id, node_one.id, interface_two.id, interface_one.id, link_options_two)
session.update_link(
node_two.id,
node_one.id,
interface_two.id,
interface_one.id,
link_options_two,
)
# instantiate session
session.instantiate()