366 lines
12 KiB
Python
366 lines
12 KiB
Python
from typing import Tuple
|
|
|
|
import pytest
|
|
|
|
from core.emulator.data import IpPrefixes, LinkOptions
|
|
from core.emulator.session import Session
|
|
from core.errors import CoreError
|
|
from core.nodes.base import CoreNode
|
|
from core.nodes.network import SwitchNode
|
|
|
|
INVALID_ID: int = 100
|
|
LINK_OPTIONS: LinkOptions = LinkOptions(
|
|
delay=50, bandwidth=5000000, loss=25, dup=25, jitter=10, buffer=100
|
|
)
|
|
|
|
|
|
def create_ptp_network(
|
|
session: Session, ip_prefixes: IpPrefixes
|
|
) -> Tuple[CoreNode, CoreNode]:
|
|
# create nodes
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
|
|
# link nodes to net node
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
session.add_link(node1.id, node2.id, iface1_data, iface2_data)
|
|
|
|
# instantiate session
|
|
session.instantiate()
|
|
|
|
return node1, node2
|
|
|
|
|
|
class TestLinks:
|
|
def test_add_node_to_node(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
|
|
# when
|
|
iface1, iface2 = session.add_link(
|
|
node1.id, node2.id, iface1_data, iface2_data, options=LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert node1.get_iface(iface1_data.id)
|
|
assert node2.get_iface(iface2_data.id)
|
|
assert iface1 is not None
|
|
assert iface2 is not None
|
|
assert iface1.local_options == LINK_OPTIONS
|
|
assert iface1.has_local_netem
|
|
assert iface2.local_options == LINK_OPTIONS
|
|
assert iface2.has_local_netem
|
|
|
|
def test_add_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
|
|
# when
|
|
iface, _ = session.add_link(
|
|
node1.id, node2.id, iface1_data=iface1_data, options=LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert node2.links()
|
|
assert node1.get_iface(iface1_data.id)
|
|
assert iface is not None
|
|
assert iface.local_options == LINK_OPTIONS
|
|
assert iface.has_local_netem
|
|
|
|
def test_add_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
|
|
# when
|
|
_, iface = session.add_link(
|
|
node1.id, node2.id, iface2_data=iface2_data, options=LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert node1.links()
|
|
assert node2.get_iface(iface2_data.id)
|
|
assert iface is not None
|
|
assert iface.local_options == LINK_OPTIONS
|
|
assert iface.has_local_netem
|
|
|
|
def test_add_net_to_net(self, session):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
|
|
# when
|
|
iface, _ = session.add_link(node1.id, node2.id, options=LINK_OPTIONS)
|
|
|
|
# then
|
|
assert node1.links()
|
|
assert iface is not None
|
|
assert iface.local_options == LINK_OPTIONS
|
|
assert iface.options == LINK_OPTIONS
|
|
assert iface.has_local_netem
|
|
assert iface.has_netem
|
|
|
|
def test_add_node_to_node_uni(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
link_options1 = LinkOptions(
|
|
delay=50,
|
|
bandwidth=5000000,
|
|
loss=25,
|
|
dup=25,
|
|
jitter=10,
|
|
buffer=100,
|
|
unidirectional=True,
|
|
)
|
|
link_options2 = LinkOptions(
|
|
delay=51,
|
|
bandwidth=5000001,
|
|
loss=26,
|
|
dup=26,
|
|
jitter=11,
|
|
buffer=101,
|
|
unidirectional=True,
|
|
)
|
|
|
|
# when
|
|
iface1, iface2 = session.add_link(
|
|
node1.id, node2.id, iface1_data, iface2_data, link_options1
|
|
)
|
|
session.update_link(
|
|
node2.id, node1.id, iface2_data.id, iface1_data.id, link_options2
|
|
)
|
|
|
|
# then
|
|
assert node1.get_iface(iface1_data.id)
|
|
assert node2.get_iface(iface2_data.id)
|
|
assert iface1 is not None
|
|
assert iface2 is not None
|
|
assert iface1.local_options == link_options1
|
|
assert iface1.has_local_netem
|
|
assert iface2.local_options == link_options2
|
|
assert iface2.has_local_netem
|
|
|
|
def test_update_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface1, _ = session.add_link(node1.id, node2.id, iface1_data)
|
|
assert iface1.local_options != LINK_OPTIONS
|
|
|
|
# when
|
|
session.update_link(
|
|
node1.id, node2.id, iface1_id=iface1_data.id, options=LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert iface1.local_options == LINK_OPTIONS
|
|
assert iface1.has_local_netem
|
|
|
|
def test_update_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
_, iface2 = session.add_link(node1.id, node2.id, iface2_data=iface2_data)
|
|
assert iface2.local_options != LINK_OPTIONS
|
|
|
|
# when
|
|
session.update_link(
|
|
node1.id, node2.id, iface2_id=iface2_data.id, options=LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert iface2.local_options == LINK_OPTIONS
|
|
assert iface2.has_local_netem
|
|
|
|
def test_update_ptp(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
iface1, iface2 = session.add_link(node1.id, node2.id, iface1_data, iface2_data)
|
|
assert iface1.local_options != LINK_OPTIONS
|
|
assert iface2.local_options != LINK_OPTIONS
|
|
|
|
# when
|
|
session.update_link(
|
|
node1.id, node2.id, iface1_data.id, iface2_data.id, LINK_OPTIONS
|
|
)
|
|
|
|
# then
|
|
assert iface1.local_options == LINK_OPTIONS
|
|
assert iface1.has_local_netem
|
|
assert iface2.local_options == LINK_OPTIONS
|
|
assert iface2.has_local_netem
|
|
|
|
def test_update_net_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
iface1, _ = session.add_link(node1.id, node2.id)
|
|
assert iface1.local_options != LINK_OPTIONS
|
|
|
|
# when
|
|
session.update_link(node1.id, node2.id, options=LINK_OPTIONS)
|
|
|
|
# then
|
|
assert iface1.local_options == LINK_OPTIONS
|
|
assert iface1.has_local_netem
|
|
assert iface1.options == LINK_OPTIONS
|
|
assert iface1.has_netem
|
|
|
|
def test_clear_net_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
iface1, _ = session.add_link(node1.id, node2.id, options=LINK_OPTIONS)
|
|
assert iface1.local_options == LINK_OPTIONS
|
|
assert iface1.has_local_netem
|
|
assert iface1.options == LINK_OPTIONS
|
|
assert iface1.has_netem
|
|
|
|
# when
|
|
options = LinkOptions(delay=0, bandwidth=0, loss=0.0, dup=0, jitter=0, buffer=0)
|
|
session.update_link(node1.id, node2.id, options=options)
|
|
|
|
# then
|
|
assert iface1.local_options.is_clear()
|
|
assert not iface1.has_local_netem
|
|
assert iface1.options.is_clear()
|
|
assert not iface1.has_netem
|
|
|
|
def test_delete_node_to_node(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
session.add_link(node1.id, node2.id, iface1_data, iface2_data)
|
|
assert node1.get_iface(iface1_data.id)
|
|
assert node2.get_iface(iface2_data.id)
|
|
|
|
# when
|
|
session.delete_link(node1.id, node2.id, iface1_data.id, iface2_data.id)
|
|
|
|
# then
|
|
assert iface1_data.id not in node1.ifaces
|
|
assert iface2_data.id not in node2.ifaces
|
|
|
|
def test_delete_node_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
session.add_link(node1.id, node2.id, iface1_data)
|
|
assert node1.get_iface(iface1_data.id)
|
|
|
|
# when
|
|
session.delete_link(node1.id, node2.id, iface1_id=iface1_data.id)
|
|
|
|
# then
|
|
assert iface1_data.id not in node1.ifaces
|
|
|
|
def test_delete_net_to_node(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(CoreNode)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
session.add_link(node1.id, node2.id, iface2_data=iface2_data)
|
|
assert node2.get_iface(iface2_data.id)
|
|
|
|
# when
|
|
session.delete_link(node1.id, node2.id, iface2_id=iface2_data.id)
|
|
|
|
# then
|
|
assert iface2_data.id not in node2.ifaces
|
|
|
|
def test_delete_net_to_net(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
session.add_link(node1.id, node2.id)
|
|
assert node1.get_linked_iface(node2)
|
|
|
|
# when
|
|
session.delete_link(node1.id, node2.id)
|
|
|
|
# then
|
|
assert not node1.get_linked_iface(node2)
|
|
|
|
def test_delete_node_error(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
session.add_link(node1.id, node2.id)
|
|
assert node1.get_linked_iface(node2)
|
|
|
|
# when
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(node1.id, INVALID_ID)
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(INVALID_ID, node2.id)
|
|
|
|
def test_delete_net_to_net_error(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
node3 = session.add_node(SwitchNode)
|
|
session.add_link(node1.id, node2.id)
|
|
assert node1.get_linked_iface(node2)
|
|
|
|
# when
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(node1.id, node3.id)
|
|
|
|
def test_delete_node_to_net_error(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(SwitchNode)
|
|
node3 = session.add_node(SwitchNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface1, _ = session.add_link(node1.id, node2.id, iface1_data)
|
|
assert iface1
|
|
|
|
# when
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(node1.id, node3.id)
|
|
|
|
def test_delete_net_to_node_error(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(SwitchNode)
|
|
node2 = session.add_node(CoreNode)
|
|
node3 = session.add_node(SwitchNode)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
_, iface2 = session.add_link(node1.id, node2.id, iface2_data=iface2_data)
|
|
assert iface2
|
|
|
|
# when
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(node1.id, node3.id)
|
|
|
|
def test_delete_node_to_node_error(self, session: Session, ip_prefixes: IpPrefixes):
|
|
# given
|
|
node1 = session.add_node(CoreNode)
|
|
node2 = session.add_node(CoreNode)
|
|
node3 = session.add_node(SwitchNode)
|
|
iface1_data = ip_prefixes.create_iface(node1)
|
|
iface2_data = ip_prefixes.create_iface(node2)
|
|
iface1, iface2 = session.add_link(node1.id, node2.id, iface1_data, iface2_data)
|
|
assert iface1
|
|
assert iface2
|
|
|
|
# when
|
|
with pytest.raises(CoreError):
|
|
session.delete_link(node1.id, node3.id)
|