afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity
core-extra/daemon/tests/test_links.py

418 lines
15 KiB
Python
Raw Blame History

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 len(session.link_manager.links()) == 1
assert node1.get_iface(iface1_data.id)
assert node2.get_iface(iface2_data.id)
assert iface1 is not None
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert node1.get_iface(iface1_data.id)
assert iface2 is not None
assert iface2.options == LINK_OPTIONS
assert iface2.has_netem
assert node1.get_iface(iface1_data.id)
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
iface1, iface2 = session.add_link(
node1.id, node2.id, iface1_data=iface1_data, options=LINK_OPTIONS
)
# then
assert len(session.link_manager.links()) == 1
assert iface1 is not None
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert node1.get_iface(iface1_data.id)
assert iface2 is not None
assert iface2.options == LINK_OPTIONS
assert iface2.has_netem
assert node2.get_iface(iface1_data.id)
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
iface1, iface2 = session.add_link(
node1.id, node2.id, iface2_data=iface2_data, options=LINK_OPTIONS
)
# then
assert len(session.link_manager.links()) == 1
assert iface1 is not None
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert node1.get_iface(iface1.id)
assert iface2 is not None
assert iface2.options == LINK_OPTIONS
assert iface2.has_netem
assert node2.get_iface(iface2.id)
def test_add_net_to_net(self, session: Session):
# given
node1 = session.add_node(SwitchNode)
node2 = session.add_node(SwitchNode)
# when
iface1, iface2 = session.add_link(node1.id, node2.id, options=LINK_OPTIONS)
# then
assert len(session.link_manager.links()) == 1
assert iface1 is not None
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2 is not None
assert iface2.options == LINK_OPTIONS
assert iface2.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 len(session.link_manager.links()) == 1
assert node1.get_iface(iface1_data.id)
assert node2.get_iface(iface2_data.id)
assert iface1 is not None
assert iface1.options == link_options1
assert iface1.has_netem
assert iface2 is not None
assert iface2.options == link_options2
assert iface2.has_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, iface2 = session.add_link(node1.id, node2.id, iface1_data)
assert len(session.link_manager.links()) == 1
assert iface1.options != LINK_OPTIONS
assert iface2.options != LINK_OPTIONS
# when
session.update_link(node1.id, node2.id, iface1.id, iface2.id, LINK_OPTIONS)
# then
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2.options == LINK_OPTIONS
assert iface2.has_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)
iface1, iface2 = session.add_link(node1.id, node2.id, iface2_data=iface2_data)
assert iface1.options != LINK_OPTIONS
assert iface2.options != LINK_OPTIONS
# when
session.update_link(node1.id, node2.id, iface1.id, iface2.id, LINK_OPTIONS)
# then
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2.options == LINK_OPTIONS
assert iface2.has_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.options != LINK_OPTIONS
assert iface2.options != LINK_OPTIONS
# when
session.update_link(node1.id, node2.id, iface1.id, iface2.id, LINK_OPTIONS)
# then
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2.options == LINK_OPTIONS
assert iface2.has_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, iface2 = session.add_link(node1.id, node2.id)
assert iface1.options != LINK_OPTIONS
assert iface2.options != LINK_OPTIONS
# when
session.update_link(node1.id, node2.id, iface1.id, iface2.id, LINK_OPTIONS)
# then
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2.options == LINK_OPTIONS
assert iface2.has_netem
def test_update_error(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.options != LINK_OPTIONS
assert iface2.options != LINK_OPTIONS
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, INVALID_ID, iface1.id, iface2.id)
def test_clear_net_to_net(self, session: Session, ip_prefixes: IpPrefixes):
# given
node1 = session.add_node(SwitchNode)
node2 = session.add_node(SwitchNode)
iface1, iface2 = session.add_link(node1.id, node2.id, options=LINK_OPTIONS)
assert iface1.options == LINK_OPTIONS
assert iface1.has_netem
assert iface2.options == LINK_OPTIONS
assert iface2.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, iface1.id, iface2.id, options)
# then
assert iface1.options.is_clear()
assert not iface1.has_netem
assert iface2.options.is_clear()
assert not iface2.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)
iface1, iface2 = session.add_link(node1.id, node2.id, iface1_data, iface2_data)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
session.delete_link(node1.id, node2.id, iface1.id, iface2.id)
# then
assert len(session.link_manager.links()) == 0
assert iface1.id not in node1.ifaces
assert iface2.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)
iface1, iface2 = session.add_link(node1.id, node2.id, iface1_data)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
session.delete_link(node1.id, node2.id, iface1.id, iface2.id)
# then
assert len(session.link_manager.links()) == 0
assert iface1.id not in node1.ifaces
assert iface2.id not in node2.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)
iface1, iface2 = session.add_link(node1.id, node2.id, iface2_data=iface2_data)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
session.delete_link(node1.id, node2.id, iface1.id, iface2.id)
# then
assert len(session.link_manager.links()) == 0
assert iface1.id not in node1.ifaces
assert iface2.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)
iface1, iface2 = session.add_link(node1.id, node2.id)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
session.delete_link(node1.id, node2.id, iface1.id, iface2.id)
# then
assert len(session.link_manager.links()) == 0
assert iface1.id not in node1.ifaces
assert iface2.id not in node2.ifaces
def test_delete_node_error(self, session: Session, ip_prefixes: IpPrefixes):
# given
node1 = session.add_node(SwitchNode)
node2 = session.add_node(SwitchNode)
iface1, iface2 = session.add_link(node1.id, node2.id)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, INVALID_ID, iface1.id, iface2.id)
with pytest.raises(CoreError):
session.delete_link(INVALID_ID, node2.id, iface1.id, iface2.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)
iface1, iface2 = session.add_link(node1.id, node2.id)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, node3.id, iface1.id, iface2.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, iface2 = session.add_link(node1.id, node2.id, iface1_data)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, node3.id, iface1.id, iface2.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)
iface1, iface2 = session.add_link(node1.id, node2.id, iface2_data=iface2_data)
assert len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, node3.id, iface1.id, iface2.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 len(session.link_manager.links()) == 1
assert node1.get_iface(iface1.id)
assert node2.get_iface(iface2.id)
# when
with pytest.raises(CoreError):
session.delete_link(node1.id, node3.id, iface1.id, iface2.id)
Reference in a new issue View git blame Copy permalink