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

variable/grpc cleanup to rename everything using spelt out numbers instead of actual numbers

Browse source
This commit is contained in:
Blake Harnden 2020-06-12 16:52:41 -07:00
parent b28ef76d65
commit 876699e8ef
45 changed files with 932 additions and 966 deletions
Download patch file Download diff file Expand all files Collapse all files

37
daemon/tests/emane/test_emane.py
View file

@ -3,6 +3,7 @@ Unit tests for testing CORE EMANE networks.
"""
import os
from tempfile import TemporaryFile
from typing import Type
from xml.etree import ElementTree
import pytest
@ -43,7 +44,9 @@ def ping(
class TestEmane:
@pytest.mark.parametrize("model", _EMANE_MODELS)
def test_models(self, session: Session, model: EmaneModel, ip_prefixes: IpPrefixes):
def test_models(
self, session: Session, model: Type[EmaneModel], ip_prefixes: IpPrefixes
):
"""
Test emane models within a basic network.
@ -70,20 +73,20 @@ class TestEmane:
# create nodes
options = NodeOptions(model="mdr")
options.set_position(150, 150)
node_one = session.add_node(CoreNode, options=options)
node1 = session.add_node(CoreNode, options=options)
options.set_position(300, 150)
node_two = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode, options=options)
for i, node in enumerate([node_one, node_two]):
for i, node in enumerate([node1, node2]):
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)
session.add_link(node.id, emane_network.id, interface1_data=interface)
# instantiate session
session.instantiate()
# ping n2 from n1 and assert success
status = ping(node_one, node_two, ip_prefixes, count=5)
# ping node2 from node1 and assert success
status = ping(node1, node2, ip_prefixes, count=5)
assert not status
def test_xml_emane(
@ -110,22 +113,22 @@ class TestEmane:
# create nodes
options = NodeOptions(model="mdr")
options.set_position(150, 150)
node_one = session.add_node(CoreNode, options=options)
node1 = session.add_node(CoreNode, options=options)
options.set_position(300, 150)
node_two = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode, options=options)
for i, node in enumerate([node_one, node_two]):
for i, node in enumerate([node1, node2]):
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)
session.add_link(node.id, emane_network.id, interface1_data=interface)
# instantiate session
session.instantiate()
# get ids for nodes
emane_id = emane_network.id
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -141,9 +144,9 @@ class TestEmane:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
@ -154,7 +157,7 @@ class TestEmane:
)
# verify nodes and configuration were restored
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
assert session.get_node(emane_id, EmaneNet)
assert value == config_value

16
daemon/tests/test_conf.py
View file

@ -14,11 +14,11 @@ from core.nodes.network import WlanNode
class TestConfigurableOptions(ConfigurableOptions):
name_one = "value1"
name_two = "value2"
name1 = "value1"
name2 = "value2"
options = [
Configuration(_id=name_one, _type=ConfigDataTypes.STRING, label=name_one),
Configuration(_id=name_two, _type=ConfigDataTypes.STRING, label=name_two),
Configuration(_id=name1, _type=ConfigDataTypes.STRING, label=name1),
Configuration(_id=name2, _type=ConfigDataTypes.STRING, label=name2),
]
@ -33,11 +33,11 @@ class TestConf:
# then
assert len(default_values) == 2
assert TestConfigurableOptions.name_one in default_values
assert TestConfigurableOptions.name_two in default_values
assert TestConfigurableOptions.name1 in default_values
assert TestConfigurableOptions.name2 in default_values
assert len(instance_default_values) == 2
assert TestConfigurableOptions.name_one in instance_default_values
assert TestConfigurableOptions.name_two in instance_default_values
assert TestConfigurableOptions.name1 in instance_default_values
assert TestConfigurableOptions.name2 in instance_default_values
def test_nodes(self):
# given

64
daemon/tests/test_core.py
View file

@ -48,19 +48,19 @@ class TestCore:
net_node = session.add_node(net_type)
# create nodes
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
# link nodes to net node
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, net_node.id, interface_one=interface)
session.add_link(node.id, net_node.id, interface1_data=interface)
# instantiate session
session.instantiate()
# ping n2 from n1 and assert success
status = ping(node_one, node_two, ip_prefixes)
# ping node2 from node1 and assert success
status = ping(node1, node2, ip_prefixes)
assert not status
def test_vnode_client(self, request, session: Session, ip_prefixes: IpPrefixes):
@ -75,16 +75,16 @@ class TestCore:
ptp_node = session.add_node(PtpNet)
# create nodes
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
# link nodes to ptp net
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, ptp_node.id, interface_one=interface)
session.add_link(node.id, ptp_node.id, interface1_data=interface)
# get node client for testing
client = node_one.client
client = node1.client
# instantiate session
session.instantiate()
@ -108,13 +108,13 @@ class TestCore:
ptp_node = session.add_node(PtpNet)
# create nodes
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
# link nodes to ptp net
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, ptp_node.id, interface_one=interface)
session.add_link(node.id, ptp_node.id, interface1_data=interface)
# instantiate session
session.instantiate()
@ -123,22 +123,22 @@ class TestCore:
assert ptp_node.all_link_data(MessageFlags.ADD)
# check common nets exist between linked nodes
assert node_one.commonnets(node_two)
assert node_two.commonnets(node_one)
assert node1.commonnets(node2)
assert node2.commonnets(node1)
# check we can retrieve netif index
assert node_one.ifname(0)
assert node_two.ifname(0)
assert node1.ifname(0)
assert node2.ifname(0)
# check interface parameters
interface = node_one.netif(0)
interface = node1.netif(0)
interface.setparam("test", 1)
assert interface.getparam("test") == 1
assert interface.getparams()
# delete netif and test that if no longer exists
node_one.delnetif(0)
assert not node_one.netif(0)
node1.delnetif(0)
assert not node1.netif(0)
def test_wlan_ping(self, session: Session, ip_prefixes: IpPrefixes):
"""
@ -155,19 +155,19 @@ class TestCore:
# create nodes
options = NodeOptions(model="mdr")
options.set_position(0, 0)
node_one = session.add_node(CoreNode, options=options)
node_two = session.add_node(CoreNode, options=options)
node1 = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode, options=options)
# link nodes
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, wlan_node.id, interface_one=interface)
session.add_link(node.id, wlan_node.id, interface1_data=interface)
# instantiate session
session.instantiate()
# ping n2 from n1 and assert success
status = ping(node_one, node_two, ip_prefixes)
# ping node2 from node1 and assert success
status = ping(node1, node2, ip_prefixes)
assert not status
def test_mobility(self, session: Session, ip_prefixes: IpPrefixes):
@ -185,13 +185,13 @@ class TestCore:
# create nodes
options = NodeOptions(model="mdr")
options.set_position(0, 0)
node_one = session.add_node(CoreNode, options=options)
node_two = session.add_node(CoreNode, options=options)
node1 = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode, options=options)
# link nodes
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, wlan_node.id, interface_one=interface)
session.add_link(node.id, wlan_node.id, interface1_data=interface)
# configure mobility script for session
config = {

54
daemon/tests/test_grpc.py
View file

@ -34,23 +34,23 @@ class TestGrpc:
client = CoreGrpcClient()
session = grpc_server.coreemu.create_session()
position = core_pb2.Position(x=50, y=100)
node_one = core_pb2.Node(id=1, position=position, model="PC")
node1 = core_pb2.Node(id=1, position=position, model="PC")
position = core_pb2.Position(x=100, y=100)
node_two = core_pb2.Node(id=2, position=position, model="PC")
node2 = core_pb2.Node(id=2, position=position, model="PC")
position = core_pb2.Position(x=200, y=200)
wlan_node = core_pb2.Node(
id=3, type=NodeTypes.WIRELESS_LAN.value, position=position
)
nodes = [node_one, node_two, wlan_node]
nodes = [node1, node2, wlan_node]
interface_helper = InterfaceHelper(ip4_prefix="10.83.0.0/16")
interface_one = interface_helper.create_interface(node_one.id, 0)
interface_two = interface_helper.create_interface(node_two.id, 0)
interface1 = interface_helper.create_interface(node1.id, 0)
interface2 = interface_helper.create_interface(node2.id, 0)
link = core_pb2.Link(
type=core_pb2.LinkType.WIRED,
node_one_id=node_one.id,
node_two_id=node_two.id,
interface_one=interface_one,
interface_two=interface_two,
node1_id=node1.id,
node2_id=node2.id,
interface1=interface1,
interface2=interface2,
)
links = [link]
hook = core_pb2.Hook(
@ -99,11 +99,11 @@ class TestGrpc:
)
mobility_configs = [mobility_config]
service_config = ServiceConfig(
node_id=node_one.id, service="DefaultRoute", validate=["echo hello"]
node_id=node1.id, service="DefaultRoute", validate=["echo hello"]
)
service_configs = [service_config]
service_file_config = ServiceFileConfig(
node_id=node_one.id,
node_id=node1.id,
service="DefaultRoute",
file="defaultroute.sh",
data="echo hello",
@ -128,11 +128,11 @@ class TestGrpc:
)
# then
assert node_one.id in session.nodes
assert node_two.id in session.nodes
assert node1.id in session.nodes
assert node2.id in session.nodes
assert wlan_node.id in session.nodes
assert session.nodes[node_one.id].netif(0) is not None
assert session.nodes[node_two.id].netif(0) is not None
assert session.nodes[node1.id].netif(0) is not None
assert session.nodes[node2.id].netif(0) is not None
hook_file, hook_data = session._hooks[EventTypes.RUNTIME_STATE][0]
assert hook_file == hook.file
assert hook_data == hook.data
@ -153,11 +153,11 @@ class TestGrpc:
)
assert set_model_config[model_config_key] == model_config_value
service = session.services.get_service(
node_one.id, service_config.service, default_service=True
node1.id, service_config.service, default_service=True
)
assert service.validate == tuple(service_config.validate)
service_file = session.services.get_service_file(
node_one, service_file_config.service, service_file_config.file
node1, service_file_config.service, service_file_config.file
)
assert service_file.data == service_file_config.data
@ -596,7 +596,7 @@ class TestGrpc:
# then
with client.context_connect():
response = client.edit_link(
session.id, node.id, switch.id, options, interface_one_id=interface.id
session.id, node.id, switch.id, options, interface1_id=interface.id
)
# then
@ -608,28 +608,28 @@ class TestGrpc:
# given
client = CoreGrpcClient()
session = grpc_server.coreemu.create_session()
node_one = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
node_two = session.add_node(CoreNode)
interface_two = ip_prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_one, interface_two)
node1 = session.add_node(CoreNode)
interface1 = ip_prefixes.create_interface(node1)
node2 = session.add_node(CoreNode)
interface2 = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface1, interface2)
link_node = None
for node_id in session.nodes:
node = session.nodes[node_id]
if node.id not in {node_one.id, node_two.id}:
if node.id not in {node1.id, node2.id}:
link_node = node
break
assert len(link_node.all_link_data(0)) == 1
assert len(link_node.all_link_data()) == 1
# 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, node1.id, node2.id, interface1.id, interface2.id
)
# then
assert response.result is True
assert len(link_node.all_link_data(0)) == 0
assert len(link_node.all_link_data()) == 0
def test_get_wlan_config(self, grpc_server: CoreGrpcServer):
# given

159
daemon/tests/test_gui.py
View file

@ -50,12 +50,13 @@ class TestGui:
self, coretlv: CoreHandler, node_type: NodeTypes, model: Optional[str]
):
node_id = 1
name = "node1"
message = coreapi.CoreNodeMessage.create(
MessageFlags.ADD.value,
[
(NodeTlvs.NUMBER, node_id),
(NodeTlvs.TYPE, node_type.value),
(NodeTlvs.NAME, "n1"),
(NodeTlvs.NAME, name),
(NodeTlvs.X_POSITION, 0),
(NodeTlvs.Y_POSITION, 0),
(NodeTlvs.MODEL, model),
@ -63,7 +64,9 @@ class TestGui:
)
coretlv.handle_message(message)
assert coretlv.session.get_node(node_id, NodeBase) is not None
node = coretlv.session.get_node(node_id, NodeBase)
assert node
assert node.name == name
def test_node_update(self, coretlv: CoreHandler):
node_id = 1
@ -99,71 +102,71 @@ class TestGui:
coretlv.session.get_node(node_id, NodeBase)
def test_link_add_node_to_net(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
switch = 2
coretlv.session.add_node(SwitchNode, _id=switch)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
switch_id = 2
coretlv.session.add_node(SwitchNode, _id=switch_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface1_ip4 = str(ip_prefix[node1_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
def test_link_add_net_to_node(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
switch = 2
coretlv.session.add_node(SwitchNode, _id=switch)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
switch_id = 2
coretlv.session.add_node(SwitchNode, _id=switch_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface2_ip4 = str(ip_prefix[node1_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, switch),
(LinkTlvs.N2_NUMBER, node_one),
(LinkTlvs.N1_NUMBER, switch_id),
(LinkTlvs.N2_NUMBER, node1_id),
(LinkTlvs.INTERFACE2_NUMBER, 0),
(LinkTlvs.INTERFACE2_IP4, interface_one),
(LinkTlvs.INTERFACE2_IP4, interface2_ip4),
(LinkTlvs.INTERFACE2_IP4_MASK, 24),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
def test_link_add_node_to_node(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
node_two = 2
coretlv.session.add_node(CoreNode, _id=node_two)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
node2_id = 2
coretlv.session.add_node(CoreNode, _id=node2_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface_two = str(ip_prefix[node_two])
interface1_ip4 = str(ip_prefix[node1_id])
interface2_ip4 = str(ip_prefix[node2_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, node_two),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, node2_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
(LinkTlvs.INTERFACE2_NUMBER, 0),
(LinkTlvs.INTERFACE2_IP4, interface_two),
(LinkTlvs.INTERFACE2_IP4, interface2_ip4),
(LinkTlvs.INTERFACE2_IP4_MASK, 24),
],
)
@ -177,24 +180,24 @@ class TestGui:
assert len(all_links) == 1
def test_link_update(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
switch = 2
coretlv.session.add_node(SwitchNode, _id=switch)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
switch_id = 2
coretlv.session.add_node(SwitchNode, _id=switch_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface1_ip4 = str(ip_prefix[node1_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
link = all_links[0]
@ -204,37 +207,37 @@ class TestGui:
message = coreapi.CoreLinkMessage.create(
0,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.BANDWIDTH, bandwidth),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
link = all_links[0]
assert link.bandwidth == bandwidth
def test_link_delete_node_to_node(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
node_two = 2
coretlv.session.add_node(CoreNode, _id=node_two)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
node2_id = 2
coretlv.session.add_node(CoreNode, _id=node2_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface_two = str(ip_prefix[node_two])
interface1_ip4 = str(ip_prefix[node1_id])
interface2_ip4 = str(ip_prefix[node2_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, node_two),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, node2_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
(LinkTlvs.INTERFACE2_IP4, interface_two),
(LinkTlvs.INTERFACE2_IP4, interface2_ip4),
(LinkTlvs.INTERFACE2_IP4_MASK, 24),
],
)
@ -248,8 +251,8 @@ class TestGui:
message = coreapi.CoreLinkMessage.create(
MessageFlags.DELETE.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, node_two),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, node2_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE2_NUMBER, 0),
],
@ -263,74 +266,74 @@ class TestGui:
assert len(all_links) == 0
def test_link_delete_node_to_net(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
switch = 2
coretlv.session.add_node(SwitchNode, _id=switch)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
switch_id = 2
coretlv.session.add_node(SwitchNode, _id=switch_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface1_ip4 = str(ip_prefix[node1_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
message = coreapi.CoreLinkMessage.create(
MessageFlags.DELETE.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 0
def test_link_delete_net_to_node(self, coretlv: CoreHandler):
node_one = 1
coretlv.session.add_node(CoreNode, _id=node_one)
switch = 2
coretlv.session.add_node(SwitchNode, _id=switch)
node1_id = 1
coretlv.session.add_node(CoreNode, _id=node1_id)
switch_id = 2
coretlv.session.add_node(SwitchNode, _id=switch_id)
ip_prefix = netaddr.IPNetwork("10.0.0.0/24")
interface_one = str(ip_prefix[node_one])
interface1_ip4 = str(ip_prefix[node1_id])
message = coreapi.CoreLinkMessage.create(
MessageFlags.ADD.value,
[
(LinkTlvs.N1_NUMBER, node_one),
(LinkTlvs.N2_NUMBER, switch),
(LinkTlvs.N1_NUMBER, node1_id),
(LinkTlvs.N2_NUMBER, switch_id),
(LinkTlvs.INTERFACE1_NUMBER, 0),
(LinkTlvs.INTERFACE1_IP4, interface_one),
(LinkTlvs.INTERFACE1_IP4, interface1_ip4),
(LinkTlvs.INTERFACE1_IP4_MASK, 24),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 1
message = coreapi.CoreLinkMessage.create(
MessageFlags.DELETE.value,
[
(LinkTlvs.N1_NUMBER, switch),
(LinkTlvs.N2_NUMBER, node_one),
(LinkTlvs.N1_NUMBER, switch_id),
(LinkTlvs.N2_NUMBER, node1_id),
(LinkTlvs.INTERFACE2_NUMBER, 0),
],
)
coretlv.handle_message(message)
switch_node = coretlv.session.get_node(switch, SwitchNode)
switch_node = coretlv.session.get_node(switch_id, SwitchNode)
all_links = switch_node.all_link_data()
assert len(all_links) == 0

246
daemon/tests/test_links.py
View file

@ -10,71 +10,71 @@ def create_ptp_network(
session: Session, ip_prefixes: IpPrefixes
) -> Tuple[CoreNode, CoreNode]:
# create nodes
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
# link nodes to net node
interface_one = ip_prefixes.create_interface(node_one)
interface_two = ip_prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_one, interface_two)
interface1_data = ip_prefixes.create_interface(node1)
interface2_data = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface1_data, interface2_data)
# instantiate session
session.instantiate()
return node_one, node_two
return node1, node2
class TestLinks:
def test_add_ptp(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
interface_two = ip_prefixes.create_interface(node_two)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
interface2_data = ip_prefixes.create_interface(node2)
# when
session.add_link(node_one.id, node_two.id, interface_one, interface_two)
session.add_link(node1.id, node2.id, interface1_data, interface2_data)
# then
assert node_one.netif(interface_one.id)
assert node_two.netif(interface_two.id)
assert node1.netif(interface1_data.id)
assert node2.netif(interface2_data.id)
def test_add_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(CoreNode)
node_two = session.add_node(SwitchNode)
interface_one = ip_prefixes.create_interface(node_one)
node1 = session.add_node(CoreNode)
node2 = session.add_node(SwitchNode)
interface1_data = ip_prefixes.create_interface(node1)
# when
session.add_link(node_one.id, node_two.id, interface_one=interface_one)
session.add_link(node1.id, node2.id, interface1_data=interface1_data)
# then
assert node_two.all_link_data()
assert node_one.netif(interface_one.id)
assert node2.all_link_data()
assert node1.netif(interface1_data.id)
def test_add_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(SwitchNode)
node_two = session.add_node(CoreNode)
interface_two = ip_prefixes.create_interface(node_two)
node1 = session.add_node(SwitchNode)
node2 = session.add_node(CoreNode)
interface2_data = ip_prefixes.create_interface(node2)
# when
session.add_link(node_one.id, node_two.id, interface_two=interface_two)
session.add_link(node1.id, node2.id, interface2_data=interface2_data)
# then
assert node_one.all_link_data()
assert node_two.netif(interface_two.id)
assert node1.all_link_data()
assert node2.netif(interface2_data.id)
def test_add_net_to_net(self, session):
# given
node_one = session.add_node(SwitchNode)
node_two = session.add_node(SwitchNode)
node1 = session.add_node(SwitchNode)
node2 = session.add_node(SwitchNode)
# when
session.add_link(node_one.id, node_two.id)
session.add_link(node1.id, node2.id)
# then
assert node_one.all_link_data()
assert node1.all_link_data()
def test_update_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
# given
@ -83,34 +83,31 @@ class TestLinks:
per = 25
dup = 25
jitter = 10
node_one = session.add_node(CoreNode)
node_two = session.add_node(SwitchNode)
interface_one_data = ip_prefixes.create_interface(node_one)
session.add_link(node_one.id, node_two.id, interface_one_data)
interface_one = node_one.netif(interface_one_data.id)
assert interface_one.getparam("delay") != delay
assert interface_one.getparam("bw") != bandwidth
assert interface_one.getparam("loss") != per
assert interface_one.getparam("duplicate") != dup
assert interface_one.getparam("jitter") != jitter
node1 = session.add_node(CoreNode)
node2 = session.add_node(SwitchNode)
interface1_data = ip_prefixes.create_interface(node1)
session.add_link(node1.id, node2.id, interface1_data)
interface1 = node1.netif(interface1_data.id)
assert interface1.getparam("delay") != delay
assert interface1.getparam("bw") != bandwidth
assert interface1.getparam("loss") != per
assert interface1.getparam("duplicate") != dup
assert interface1.getparam("jitter") != jitter
# when
link_options = LinkOptions(
options = LinkOptions(
delay=delay, bandwidth=bandwidth, per=per, dup=dup, jitter=jitter
)
session.update_link(
node_one.id,
node_two.id,
interface_one_id=interface_one_data.id,
options=link_options,
node1.id, node2.id, interface1_id=interface1_data.id, options=options
)
# then
assert interface_one.getparam("delay") == delay
assert interface_one.getparam("bw") == bandwidth
assert interface_one.getparam("loss") == per
assert interface_one.getparam("duplicate") == dup
assert interface_one.getparam("jitter") == jitter
assert interface1.getparam("delay") == delay
assert interface1.getparam("bw") == bandwidth
assert interface1.getparam("loss") == per
assert interface1.getparam("duplicate") == dup
assert interface1.getparam("jitter") == jitter
def test_update_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
# given
@ -119,34 +116,31 @@ class TestLinks:
per = 25
dup = 25
jitter = 10
node_one = session.add_node(SwitchNode)
node_two = session.add_node(CoreNode)
interface_two_data = ip_prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_two=interface_two_data)
interface_two = node_two.netif(interface_two_data.id)
assert interface_two.getparam("delay") != delay
assert interface_two.getparam("bw") != bandwidth
assert interface_two.getparam("loss") != per
assert interface_two.getparam("duplicate") != dup
assert interface_two.getparam("jitter") != jitter
node1 = session.add_node(SwitchNode)
node2 = session.add_node(CoreNode)
interface2_data = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface2_data=interface2_data)
interface2 = node2.netif(interface2_data.id)
assert interface2.getparam("delay") != delay
assert interface2.getparam("bw") != bandwidth
assert interface2.getparam("loss") != per
assert interface2.getparam("duplicate") != dup
assert interface2.getparam("jitter") != jitter
# when
link_options = LinkOptions(
options = LinkOptions(
delay=delay, bandwidth=bandwidth, per=per, dup=dup, jitter=jitter
)
session.update_link(
node_one.id,
node_two.id,
interface_two_id=interface_two_data.id,
options=link_options,
node1.id, node2.id, interface2_id=interface2_data.id, options=options
)
# then
assert interface_two.getparam("delay") == delay
assert interface_two.getparam("bw") == bandwidth
assert interface_two.getparam("loss") == per
assert interface_two.getparam("duplicate") == dup
assert interface_two.getparam("jitter") == jitter
assert interface2.getparam("delay") == delay
assert interface2.getparam("bw") == bandwidth
assert interface2.getparam("loss") == per
assert interface2.getparam("duplicate") == dup
assert interface2.getparam("jitter") == jitter
def test_update_ptp(self, session: Session, ip_prefixes: IpPrefixes):
# given
@ -155,93 +149,85 @@ class TestLinks:
per = 25
dup = 25
jitter = 10
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
interface_one_data = ip_prefixes.create_interface(node_one)
interface_two_data = ip_prefixes.create_interface(node_two)
session.add_link(
node_one.id, node_two.id, interface_one_data, interface_two_data
)
interface_one = node_one.netif(interface_one_data.id)
interface_two = node_two.netif(interface_two_data.id)
assert interface_one.getparam("delay") != delay
assert interface_one.getparam("bw") != bandwidth
assert interface_one.getparam("loss") != per
assert interface_one.getparam("duplicate") != dup
assert interface_one.getparam("jitter") != jitter
assert interface_two.getparam("delay") != delay
assert interface_two.getparam("bw") != bandwidth
assert interface_two.getparam("loss") != per
assert interface_two.getparam("duplicate") != dup
assert interface_two.getparam("jitter") != jitter
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
interface2_data = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface1_data, interface2_data)
interface1 = node1.netif(interface1_data.id)
interface2 = node2.netif(interface2_data.id)
assert interface1.getparam("delay") != delay
assert interface1.getparam("bw") != bandwidth
assert interface1.getparam("loss") != per
assert interface1.getparam("duplicate") != dup
assert interface1.getparam("jitter") != jitter
assert interface2.getparam("delay") != delay
assert interface2.getparam("bw") != bandwidth
assert interface2.getparam("loss") != per
assert interface2.getparam("duplicate") != dup
assert interface2.getparam("jitter") != jitter
# when
link_options = LinkOptions(
options = LinkOptions(
delay=delay, bandwidth=bandwidth, per=per, dup=dup, jitter=jitter
)
session.update_link(
node_one.id,
node_two.id,
interface_one_data.id,
interface_two_data.id,
link_options,
node1.id, node2.id, interface1_data.id, interface2_data.id, options
)
# then
assert interface_one.getparam("delay") == delay
assert interface_one.getparam("bw") == bandwidth
assert interface_one.getparam("loss") == per
assert interface_one.getparam("duplicate") == dup
assert interface_one.getparam("jitter") == jitter
assert interface_two.getparam("delay") == delay
assert interface_two.getparam("bw") == bandwidth
assert interface_two.getparam("loss") == per
assert interface_two.getparam("duplicate") == dup
assert interface_two.getparam("jitter") == jitter
assert interface1.getparam("delay") == delay
assert interface1.getparam("bw") == bandwidth
assert interface1.getparam("loss") == per
assert interface1.getparam("duplicate") == dup
assert interface1.getparam("jitter") == jitter
assert interface2.getparam("delay") == delay
assert interface2.getparam("bw") == bandwidth
assert interface2.getparam("loss") == per
assert interface2.getparam("duplicate") == dup
assert interface2.getparam("jitter") == jitter
def test_delete_ptp(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
interface_two = ip_prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_one, interface_two)
assert node_one.netif(interface_one.id)
assert node_two.netif(interface_two.id)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
interface2_data = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface1_data, interface2_data)
assert node1.netif(interface1_data.id)
assert node2.netif(interface2_data.id)
# when
session.delete_link(
node_one.id, node_two.id, interface_one.id, interface_two.id
)
session.delete_link(node1.id, node2.id, interface1_data.id, interface2_data.id)
# then
assert not node_one.netif(interface_one.id)
assert not node_two.netif(interface_two.id)
assert not node1.netif(interface1_data.id)
assert not node2.netif(interface2_data.id)
def test_delete_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(CoreNode)
node_two = session.add_node(SwitchNode)
interface_one = ip_prefixes.create_interface(node_one)
session.add_link(node_one.id, node_two.id, interface_one)
assert node_one.netif(interface_one.id)
node1 = session.add_node(CoreNode)
node2 = session.add_node(SwitchNode)
interface1_data = ip_prefixes.create_interface(node1)
session.add_link(node1.id, node2.id, interface1_data)
assert node1.netif(interface1_data.id)
# when
session.delete_link(node_one.id, node_two.id, interface_one_id=interface_one.id)
session.delete_link(node1.id, node2.id, interface1_id=interface1_data.id)
# then
assert not node_one.netif(interface_one.id)
assert not node1.netif(interface1_data.id)
def test_delete_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
# given
node_one = session.add_node(SwitchNode)
node_two = session.add_node(CoreNode)
interface_two = ip_prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_two=interface_two)
assert node_two.netif(interface_two.id)
node1 = session.add_node(SwitchNode)
node2 = session.add_node(CoreNode)
interface2_data = ip_prefixes.create_interface(node2)
session.add_link(node1.id, node2.id, interface2_data=interface2_data)
assert node2.netif(interface2_data.id)
# when
session.delete_link(node_one.id, node_two.id, interface_two_id=interface_two.id)
session.delete_link(node1.id, node2.id, interface2_id=interface2_data.id)
# then
assert not node_two.netif(interface_two.id)
assert not node2.netif(interface2_data.id)

20
daemon/tests/test_services.py
View file

@ -206,23 +206,23 @@ class TestServices:
# given
ServiceManager.add_services(_SERVICES_PATH)
my_service = ServiceManager.get(SERVICE_ONE)
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
file_name = my_service.configs[0]
file_data_one = "# custom file one"
file_data_two = "# custom file two"
file_data1 = "# custom file one"
file_data2 = "# custom file two"
session.services.set_service_file(
node_one.id, my_service.name, file_name, file_data_one
node1.id, my_service.name, file_name, file_data1
)
session.services.set_service_file(
node_two.id, my_service.name, file_name, file_data_two
node2.id, my_service.name, file_name, file_data2
)
# when
custom_service_one = session.services.get_service(node_one.id, my_service.name)
session.services.create_service_files(node_one, custom_service_one)
custom_service_two = session.services.get_service(node_two.id, my_service.name)
session.services.create_service_files(node_two, custom_service_two)
custom_service1 = session.services.get_service(node1.id, my_service.name)
session.services.create_service_files(node1, custom_service1)
custom_service2 = session.services.get_service(node2.id, my_service.name)
session.services.create_service_files(node2, custom_service2)
def test_service_import(self):
"""

258
daemon/tests/test_xml.py
View file

@ -68,20 +68,20 @@ class TestXml:
ptp_node = session.add_node(PtpNet)
# create nodes
node_one = session.add_node(CoreNode)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode)
node2 = session.add_node(CoreNode)
# link nodes to ptp net
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, ptp_node.id, interface_one=interface)
session.add_link(node.id, ptp_node.id, interface1_data=interface)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -97,16 +97,16 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
# verify nodes have been recreated
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
def test_xml_ptp_services(
self, session: Session, tmpdir: TemporaryFile, ip_prefixes: IpPrefixes
@ -123,28 +123,28 @@ class TestXml:
# create nodes
options = NodeOptions(model="host")
node_one = session.add_node(CoreNode, options=options)
node_two = session.add_node(CoreNode)
node1 = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode)
# link nodes to ptp net
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, ptp_node.id, interface_one=interface)
session.add_link(node.id, ptp_node.id, interface1_data=interface)
# set custom values for node service
session.services.set_service(node_one.id, SshService.name)
session.services.set_service(node1.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
node1.id, SshService.name, service_file, file_data
)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -160,19 +160,19 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
# retrieve custom service
service = session.services.get_service(node_one.id, SshService.name)
service = session.services.get_service(node1.id, SshService.name)
# verify nodes have been recreated
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
assert service.config_data.get(service_file) == file_data
def test_xml_mobility(
@ -192,21 +192,21 @@ class TestXml:
# create nodes
options = NodeOptions(model="mdr")
options.set_position(0, 0)
node_one = session.add_node(CoreNode, options=options)
node_two = session.add_node(CoreNode, options=options)
node1 = session.add_node(CoreNode, options=options)
node2 = session.add_node(CoreNode, options=options)
# link nodes
for node in [node_one, node_two]:
for node in [node1, node2]:
interface = ip_prefixes.create_interface(node)
session.add_link(node.id, wlan_node.id, interface_one=interface)
session.add_link(node.id, wlan_node.id, interface1_data=interface)
# instantiate session
session.instantiate()
# get ids for nodes
wlan_id = wlan_node.id
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -222,9 +222,9 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
@ -233,8 +233,8 @@ class TestXml:
value = str(session.mobility.get_config("test", wlan_id, BasicRangeModel.name))
# verify nodes and configuration were restored
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
assert session.get_node(wlan_id, WlanNode)
assert value == "1"
@ -246,18 +246,18 @@ class TestXml:
:param tmpdir: tmpdir to create data in
"""
# create nodes
switch_one = session.add_node(SwitchNode)
switch_two = session.add_node(SwitchNode)
switch1 = session.add_node(SwitchNode)
switch2 = session.add_node(SwitchNode)
# link nodes
session.add_link(switch_one.id, switch_two.id)
session.add_link(switch1.id, switch2.id)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = switch_one.id
n2_id = switch_two.id
node1_id = switch1.id
node2_id = switch2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -273,19 +273,19 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, SwitchNode)
assert not session.get_node(node1_id, SwitchNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, SwitchNode)
assert not session.get_node(node2_id, SwitchNode)
# load saved xml
session.open_xml(file_path, start=True)
# verify nodes have been recreated
switch_one = session.get_node(n1_id, SwitchNode)
switch_two = session.get_node(n2_id, SwitchNode)
assert switch_one
assert switch_two
assert len(switch_one.all_link_data() + switch_two.all_link_data()) == 1
switch1 = session.get_node(node1_id, SwitchNode)
switch2 = session.get_node(node2_id, SwitchNode)
assert switch1
assert switch2
assert len(switch1.all_link_data() + switch2.all_link_data()) == 1
def test_link_options(
self, session: Session, tmpdir: TemporaryFile, ip_prefixes: IpPrefixes
@ -298,25 +298,25 @@ class TestXml:
:param ip_prefixes: generates ip addresses for nodes
"""
# create nodes
node_one = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
node1 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
switch = session.add_node(SwitchNode)
# create link
link_options = LinkOptions()
link_options.per = 10.5
link_options.bandwidth = 50000
link_options.jitter = 10
link_options.delay = 30
link_options.dup = 5
session.add_link(node_one.id, switch.id, interface_one, options=link_options)
options = LinkOptions()
options.per = 10.5
options.bandwidth = 50000
options.jitter = 10
options.delay = 30
options.dup = 5
session.add_link(node1.id, switch.id, interface1_data, options=options)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = node_one.id
n2_id = switch.id
node1_id = node1.id
node2_id = switch.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -332,26 +332,26 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, SwitchNode)
assert not session.get_node(node2_id, SwitchNode)
# load saved xml
session.open_xml(file_path, start=True)
# verify nodes have been recreated
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, SwitchNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, SwitchNode)
links = []
for node_id in session.nodes:
node = session.nodes[node_id]
links += node.all_link_data()
link = links[0]
assert link_options.per == link.per
assert link_options.bandwidth == link.bandwidth
assert link_options.jitter == link.jitter
assert link_options.delay == link.delay
assert link_options.dup == link.dup
assert options.per == link.per
assert options.bandwidth == link.bandwidth
assert options.jitter == link.jitter
assert options.delay == link.delay
assert options.dup == link.dup
def test_link_options_ptp(
self, session: Session, tmpdir: TemporaryFile, ip_prefixes: IpPrefixes
@ -364,28 +364,26 @@ class TestXml:
:param ip_prefixes: generates ip addresses for nodes
"""
# create nodes
node_one = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
node_two = session.add_node(CoreNode)
interface_two = ip_prefixes.create_interface(node_two)
node1 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
node2 = session.add_node(CoreNode)
interface2_data = ip_prefixes.create_interface(node2)
# create link
link_options = LinkOptions()
link_options.per = 10.5
link_options.bandwidth = 50000
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
)
options = LinkOptions()
options.per = 10.5
options.bandwidth = 50000
options.jitter = 10
options.delay = 30
options.dup = 5
session.add_link(node1.id, node2.id, interface1_data, interface2_data, options)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -401,26 +399,26 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
# verify nodes have been recreated
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
links = []
for node_id in session.nodes:
node = session.nodes[node_id]
links += node.all_link_data()
link = links[0]
assert link_options.per == link.per
assert link_options.bandwidth == link.bandwidth
assert link_options.jitter == link.jitter
assert link_options.delay == link.delay
assert link_options.dup == link.dup
assert options.per == link.per
assert options.bandwidth == link.bandwidth
assert options.jitter == link.jitter
assert options.delay == link.delay
assert options.dup == link.dup
def test_link_options_bidirectional(
self, session: Session, tmpdir: TemporaryFile, ip_prefixes: IpPrefixes
@ -433,43 +431,37 @@ class TestXml:
:param ip_prefixes: generates ip addresses for nodes
"""
# create nodes
node_one = session.add_node(CoreNode)
interface_one = ip_prefixes.create_interface(node_one)
node_two = session.add_node(CoreNode)
interface_two = ip_prefixes.create_interface(node_two)
node1 = session.add_node(CoreNode)
interface1_data = ip_prefixes.create_interface(node1)
node2 = session.add_node(CoreNode)
interface2_data = ip_prefixes.create_interface(node2)
# create link
link_options_one = LinkOptions()
link_options_one.unidirectional = 1
link_options_one.bandwidth = 5000
link_options_one.delay = 10
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
)
link_options_two = LinkOptions()
link_options_two.unidirectional = 1
link_options_two.bandwidth = 10000
link_options_two.delay = 20
link_options_two.per = 10
link_options_two.dup = 10
link_options_two.jitter = 10
options1 = LinkOptions()
options1.unidirectional = 1
options1.bandwidth = 5000
options1.delay = 10
options1.per = 10.5
options1.dup = 5
options1.jitter = 5
session.add_link(node1.id, node2.id, interface1_data, interface2_data, options1)
options2 = LinkOptions()
options2.unidirectional = 1
options2.bandwidth = 10000
options2.delay = 20
options2.per = 10
options2.dup = 10
options2.jitter = 10
session.update_link(
node_two.id,
node_one.id,
interface_two.id,
interface_one.id,
link_options_two,
node2.id, node1.id, interface2_data.id, interface1_data.id, options2
)
# instantiate session
session.instantiate()
# get ids for nodes
n1_id = node_one.id
n2_id = node_two.id
node1_id = node1.id
node2_id = node2.id
# save xml
xml_file = tmpdir.join("session.xml")
@ -485,30 +477,30 @@ class TestXml:
# verify nodes have been removed from session
with pytest.raises(CoreError):
assert not session.get_node(n1_id, CoreNode)
assert not session.get_node(node1_id, CoreNode)
with pytest.raises(CoreError):
assert not session.get_node(n2_id, CoreNode)
assert not session.get_node(node2_id, CoreNode)
# load saved xml
session.open_xml(file_path, start=True)
# verify nodes have been recreated
assert session.get_node(n1_id, CoreNode)
assert session.get_node(n2_id, CoreNode)
assert session.get_node(node1_id, CoreNode)
assert session.get_node(node2_id, CoreNode)
links = []
for node_id in session.nodes:
node = session.nodes[node_id]
links += node.all_link_data()
assert len(links) == 2
link_one = links[0]
link_two = links[1]
assert link_options_one.bandwidth == link_one.bandwidth
assert link_options_one.delay == link_one.delay
assert link_options_one.per == link_one.per
assert link_options_one.dup == link_one.dup
assert link_options_one.jitter == link_one.jitter
assert link_options_two.bandwidth == link_two.bandwidth
assert link_options_two.delay == link_two.delay
assert link_options_two.per == link_two.per
assert link_options_two.dup == link_two.dup
assert link_options_two.jitter == link_two.jitter
link1 = links[0]
link2 = links[1]
assert options1.bandwidth == link1.bandwidth
assert options1.delay == link1.delay
assert options1.per == link1.per
assert options1.dup == link1.dup
assert options1.jitter == link1.jitter
assert options2.bandwidth == link2.bandwidth
assert options2.delay == link2.delay
assert options2.per == link2.per
assert options2.dup == link2.dup
assert options2.jitter == link2.jitter