core-extra/daemon/tests/test_links.py

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)