test_int_manager.TestINTManager.test__install_int_flows() - Code Metrics - Inspection of "Merge pull request #82 from kytos-ng/chore/inc_tes..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6f27fa...aec56c )

by Vinicius

created 2024-04-05 12:34 UTC

TestINTManager.test__install_int_flows() A

↳ Parent: test_int_manager

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	3
dl	0
loc	11
ccs	10
cts	10
cp	1
crap	1
rs	9.9
c	0
b	0
f	0

"""Test INTManager"""

import pytest

from unittest.mock import AsyncMock, MagicMock
from napps.kytos.telemetry_int.exceptions import ProxyPortSameSourceIntraEVC
from napps.kytos.telemetry_int.managers.int import INTManager
from napps.kytos.telemetry_int import exceptions
from kytos.core.common import EntityStatus

from kytos.lib.helpers import (
    get_interface_mock,
    get_controller_mock,
    get_switch_mock,
)


class TestINTManager:
    """TestINTManager."""

    def test_get_proxy_port_or_raise(self) -> None:
        """Test proxy_port_or_raise."""
        dpid_a = "00:00:00:00:00:00:00:01"
        mock_switch_a = get_switch_mock(dpid_a, 0x04)
        mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
        mock_interface_a.metadata = {}
        intf_id = f"{dpid_a}:1"
        controller = get_controller_mock()
        evc_id = "3766c105686749"
        int_manager = INTManager(controller)

        # Initially the mocked interface and switch hasn't been associated in the ctrllr
        with pytest.raises(exceptions.ProxyPortNotFound) as exc:
            int_manager.get_proxy_port_or_raise(intf_id, evc_id)
        assert f"interface {intf_id} not found" in str(exc)

        # Now, proxy_port still hasn't been set yet
        controller.get_interface_by_id = lambda x: mock_interface_a
        with pytest.raises(exceptions.ProxyPortNotFound) as exc:
            int_manager.get_proxy_port_or_raise(intf_id, evc_id)
        assert f"proxy_port metadata not found in {intf_id}" in str(exc)

        # Now, destination interface hasn't been mocked yet
        mock_interface_a.metadata = {"proxy_port": 5}
        with pytest.raises(exceptions.ProxyPortDestNotFound) as exc:
            int_manager.get_proxy_port_or_raise(intf_id, evc_id)
        assert "destination interface not found" in str(exc)

        mock_interface_b = get_interface_mock("s1-eth5", 5, mock_switch_a)
        mock_interface_b.metadata = {"looped": {"port_numbers": [5, 6]}}
        mock_interface_a.switch.get_interface_by_port_no = lambda x: mock_interface_b
        # Now all dependencies have been mocked and it should get the ProxyPort
        pp = int_manager.get_proxy_port_or_raise(intf_id, evc_id)
        assert pp.source == mock_interface_b

    def test_load_uni_src_proxy_port(self) -> None:
        """Test test_load_uni_src_proxy_port."""
        dpid_a = "00:00:00:00:00:00:00:01"
        mock_switch_a = get_switch_mock(dpid_a, 0x04)
        mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
        mock_interface_a.metadata = {"proxy_port": 3}
        mock_interface_z = get_interface_mock("s1-eth2", 2, mock_switch_a)
        mock_interface_z.metadata = {"proxy_port": 5}
        intf_id_a = f"{dpid_a}:1"
        intf_id_z = f"{dpid_a}:2"
        intf_id_a_1 = f"{dpid_a}:3"
        intf_id_z_1 = f"{dpid_a}:5"

        mock_interface_a_1 = get_interface_mock("s1-eth3", 3, mock_switch_a)
        mock_interface_a_1.metadata = {"looped": {"port_numbers": [3, 4]}}
        mock_interface_a_2 = get_interface_mock("s1-eth4", 4, mock_switch_a)
        mock_interface_z_1 = get_interface_mock("s1-eth5", 5, mock_switch_a)
        mock_interface_z_1.metadata = {"looped": {"port_numbers": [5, 6]}}
        mock_interface_z_2 = get_interface_mock("s1-eth6", 6, mock_switch_a)

        def get_interface_by_port_no(port_no):
            data = {
                1: mock_interface_a,
                2: mock_interface_z,
                3: mock_interface_a_1,
                4: mock_interface_a_2,
                5: mock_interface_z_1,
                6: mock_interface_z_2,
            }
            return data[port_no]

        def get_interface_by_id(intf_id):
            data = {
                intf_id_a: mock_interface_a,
                intf_id_z: mock_interface_z,
            }
            return data[intf_id]

        controller = get_controller_mock()
        mock_switch_a.get_interface_by_port_no = get_interface_by_port_no
        controller.get_interface_by_id = get_interface_by_id

        evcs = {
            "3766c105686749": {
                "metadata": {"telemetry": {"enabled": True}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            },
            "3766c105686748": {
                "metadata": {"telemetry": {"enabled": True}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            },
            "3766c105686747": {
                "metadata": {"telemetry": {"enabled": False}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            },
        }
        int_manager = INTManager(controller)
        int_manager.load_uni_src_proxy_ports(evcs)
        assert len(int_manager.unis_src) == 2
        assert int_manager.unis_src[intf_id_a] == intf_id_a_1
        assert int_manager.unis_src[intf_id_z] == intf_id_z_1

        assert len(int_manager.srcs_pp) == 2
        assert int_manager.srcs_pp[intf_id_a_1].source == mock_interface_a_1
        assert int_manager.srcs_pp[intf_id_a_1].destination == mock_interface_a_2
        assert int_manager.srcs_pp[intf_id_z_1].source == mock_interface_z_1
        assert int_manager.srcs_pp[intf_id_z_1].destination == mock_interface_z_2

        assert int_manager.srcs_pp[intf_id_a_1].evc_ids == {
            "3766c105686749",
            "3766c105686748",
        }
        assert int_manager.srcs_pp[intf_id_z_1].evc_ids == {
            "3766c105686749",
            "3766c105686748",
        }

    async def test_handle_pp_link_down(self, monkeypatch):
        """Test test_handle_pp_link_down."""
        int_manager = INTManager(MagicMock())
        api_mock, link_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        link_mock.endpoint_a.id = "some_intf_id"
        evc_id = "3766c105686748"
        int_manager.srcs_pp[link_mock.endpoint_a.id] = pp_mock
        pp_mock.evc_ids = {evc_id}

        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        api_mock.get_evcs.return_value = {evc_id: {}}
        int_manager.remove_int_flows = AsyncMock()

        await int_manager.handle_pp_link_down(link_mock)
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_args[1] == {
            "metadata.telemetry.enabled": "true",
            "metadata.telemetry.status": "UP",
        }
        assert int_manager.remove_int_flows.call_count == 1
        args = int_manager.remove_int_flows.call_args[0]
        assert evc_id in args[0]
        assert "telemetry" in args[1]
        telemetry = args[1]["telemetry"]
        assert telemetry["enabled"]
        assert telemetry["status"] == "DOWN"
        assert telemetry["status_reason"] == ["proxy_port_down"]
        assert "status_updated_at" in telemetry

    async def test_handle_pp_link_up(self, monkeypatch):
        """Test handle_pp_link_up."""
        int_manager = INTManager(MagicMock())
        api_mock, link_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        link_mock.endpoint_a.id = "3"
        pp_mock.status = EntityStatus.UP
        link_mock.status = EntityStatus.UP
        link_mock.status_reason = []
        evc_id = "3766c105686748"
        uni_a_id, uni_z_id = "1", "2"
        src_a_id, src_z_id = "3", "5"
        int_manager.srcs_pp[src_a_id] = pp_mock
        int_manager.srcs_pp[src_z_id] = pp_mock
        int_manager.unis_src[uni_a_id] = src_a_id
        int_manager.unis_src[uni_z_id] = src_z_id
        pp_mock.evc_ids = {evc_id}

        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        api_mock.get_evcs.return_value = {
            evc_id: {
                "active": True,
                "archived": False,
                "uni_a": {"interface_id": uni_a_id},
                "uni_z": {"interface_id": uni_z_id},
            }
        }
        int_manager.install_int_flows = AsyncMock()
        int_manager._validate_map_enable_evcs = MagicMock()

        await int_manager.handle_pp_link_up(link_mock)
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_args[1] == {
            "metadata.telemetry.enabled": "true",
            "metadata.telemetry.status": "DOWN",
        }
        assert int_manager.install_int_flows.call_count == 1
        args = int_manager.install_int_flows.call_args[0]
        assert "telemetry" in args[1]
        telemetry_dict = args[1]["telemetry"]
        expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
        assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)
        assert telemetry_dict["enabled"]
        assert telemetry_dict["status"] == "UP"
        assert not telemetry_dict["status_reason"]

    async def test_handle_pp_metadata_removed(self, monkeypatch):
        """Test handle_pp_metadata_removed."""
        int_manager = INTManager(MagicMock())
        api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        intf_mock.id = "some_intf_id"
        source_id = "some_source_id"
        evc_id = "3766c105686748"
        int_manager.unis_src[intf_mock.id] = source_id
        int_manager.srcs_pp[source_id] = pp_mock
        pp_mock.evc_ids = {evc_id}

        assert "proxy_port" not in intf_mock.metadata
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        api_mock.get_evcs.return_value = {evc_id: {}}
        int_manager.remove_int_flows = AsyncMock()

        await int_manager.handle_pp_metadata_removed(intf_mock)
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_args[1] == {
            "metadata.telemetry.enabled": "true",
            "metadata.telemetry.status": "UP",
        }
        assert int_manager.remove_int_flows.call_count == 1
        args = int_manager.remove_int_flows.call_args[0]
        assert evc_id in args[0]
        assert "telemetry" in args[1]
        telemetry = args[1]["telemetry"]
        assert telemetry["enabled"]
        assert telemetry["status"] == "DOWN"
        assert telemetry["status_reason"] == ["proxy_port_metadata_removed"]
        assert "status_updated_at" in telemetry

    async def test_handle_pp_metadata_added(self, monkeypatch):
        """Test handle_pp_metadata_added."""
        int_manager = INTManager(MagicMock())
        api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        intf_mock.id = "some_intf_id"
        source_id, source_port = "some_source_id", 2
        intf_mock.metadata = {"proxy_port": source_port}
        evc_id = "3766c105686748"
        int_manager.unis_src[intf_mock.id] = source_id
        int_manager.srcs_pp[source_id] = pp_mock
        pp_mock.evc_ids = {evc_id}

        assert "proxy_port" in intf_mock.metadata
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        api_mock.get_evcs.return_value = {evc_id: {}}
        int_manager.disable_int = AsyncMock()
        int_manager.enable_int = AsyncMock()

        await int_manager.handle_pp_metadata_added(intf_mock)
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_args[1] == {"metadata.telemetry.enabled": "true"}
        assert int_manager.disable_int.call_count == 1
        assert int_manager.enable_int.call_count == 1

    async def test_handle_pp_metadata_added_no_change(self, monkeypatch):
        """Test handle_pp_metadata_added no change."""
        int_manager = INTManager(MagicMock())
        api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        intf_mock.id = "some_intf_id"
        source_id, source_port = "some_source_id", 2
        source_intf = MagicMock()
        intf_mock.metadata = {"proxy_port": source_port}
        evc_id = "3766c105686748"
        int_manager.unis_src[intf_mock.id] = source_id
        int_manager.srcs_pp[source_id] = pp_mock
        pp_mock.evc_ids = {evc_id}

        # Simulating that the current and new proxy_port source are the same
        pp_mock.source = source_intf
        intf_mock.switch.get_interface_by_port_no.return_value = source_intf

        assert "proxy_port" in intf_mock.metadata
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        api_mock.get_evcs.return_value = {evc_id: {}}
        int_manager.disable_int = AsyncMock()
        int_manager.enable_int = AsyncMock()

        await int_manager.handle_pp_metadata_added(intf_mock)
        assert not api_mock.get_evcs.call_count
        assert not int_manager.disable_int.call_count
        assert not int_manager.enable_int.call_count

    async def test_handle_pp_metadata_added_no_affected(self, monkeypatch):
        """Test handle_pp_metadata_added no affected evcs."""
        int_manager = INTManager(MagicMock())
        api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
        intf_mock.id = "some_intf_id"
        source_id, source_port = "some_source_id", 2
        intf_mock.metadata = {"proxy_port": source_port}
        evc_id = "3766c105686748"
        int_manager.unis_src[intf_mock.id] = source_id
        int_manager.srcs_pp[source_id] = pp_mock
        pp_mock.evc_ids = {evc_id}

        assert "proxy_port" in intf_mock.metadata
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)

        # Simulating returning no EVCs that were enabled and UP
        api_mock.get_evcs.return_value = {}
        int_manager.disable_int = AsyncMock()
        int_manager.enable_int = AsyncMock()

        await int_manager.handle_pp_metadata_added(intf_mock)
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_count == 1
        assert api_mock.get_evcs.call_args[1] == {
            "metadata.telemetry.enabled": "true",
        }
        assert not int_manager.disable_int.call_count
        assert not int_manager.enable_int.call_count

    async def test_disable_int_metadata(self, monkeypatch) -> None:
        """Test disable INT metadata args."""
        controller = MagicMock()
        api_mock = AsyncMock()
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)

        int_manager = INTManager(controller)
        int_manager._remove_int_flows = AsyncMock()
        await int_manager.disable_int({}, False)

        assert api_mock.add_evcs_metadata.call_count == 1
        args = api_mock.add_evcs_metadata.call_args[0]
        assert args[0] == {}
        assert "telemetry" in args[1]
        telemetry_dict = args[1]["telemetry"]
        expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
        assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)

        assert not telemetry_dict["enabled"]
        assert telemetry_dict["status"] == "DOWN"
        assert telemetry_dict["status_reason"] == ["disabled"]

        assert args[2] is False

    async def test_enable_int_metadata(self, monkeypatch) -> None:
        """Test enable INT metadata args."""
        controller = MagicMock()
        api_mock = AsyncMock()
        stored_flows_mock = AsyncMock()
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.utils.get_found_stored_flows", stored_flows_mock
        )

        int_manager = INTManager(controller)
        int_manager.remove_int_flows = AsyncMock()
        await int_manager.enable_int({}, False)

        assert stored_flows_mock.call_count == 1
        assert api_mock.add_evcs_metadata.call_count == 1
        args = api_mock.add_evcs_metadata.call_args[0]
        assert args[0] == {}
        assert "telemetry" in args[1]
        telemetry_dict = args[1]["telemetry"]
        expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
        assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)

        assert telemetry_dict["enabled"] is True
        assert telemetry_dict["status"] == "UP"
        assert telemetry_dict["status_reason"] == []

    async def test_redeploy_int(self, monkeypatch) -> None:
        """Test redeploy int."""
        controller = MagicMock()
        api_mock = AsyncMock()
        stored_flows_mock = AsyncMock()
        monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.utils.get_found_stored_flows", stored_flows_mock
        )

        int_manager = INTManager(controller)
        int_manager._remove_int_flows = AsyncMock()
        int_manager._install_int_flows = AsyncMock()

        dpid_a = "00:00:00:00:00:00:00:01"
        intf_id_a = f"{dpid_a}:1"
        intf_id_z = f"{dpid_a}:2"
        evc_id = "3766c105686749"
        evcs = {
            evc_id: {
                "metadata": {"telemetry": {"enabled": True}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            }
        }
        int_manager._validate_map_enable_evcs = MagicMock()
        await int_manager.redeploy_int(evcs)

        assert stored_flows_mock.call_count == 1
        assert int_manager._remove_int_flows.call_count == 1
        assert api_mock.get_stored_flows.call_count == 1
        assert int_manager._install_int_flows.call_count == 1

    def test_validate_intra_evc_different_proxy_ports(self) -> None:
        """Test _validate_intra_evc_different_proxy_ports."""
        pp_a, pp_z, controller = MagicMock(), MagicMock(), MagicMock()
        evc = {
            "id": "some_id",
            "uni_a": {"proxy_port": pp_a, "interface_id": "00:00:00:00:00:00:00:01:1"},
            "uni_z": {"proxy_port": pp_z, "interface_id": "00:00:00:00:00:00:00:01:2"},
        }

        int_manager = INTManager(controller)
        int_manager._validate_intra_evc_different_proxy_ports(evc)

        source = MagicMock()
        pp_a.source, pp_z.source = source, source
        with pytest.raises(ProxyPortSameSourceIntraEVC):
            int_manager._validate_intra_evc_different_proxy_ports(evc)

    async def test__remove_int_flows(self, inter_evc_evpl_flows_data) -> None:
        """test _remove_int_flows."""
        controller = get_controller_mock()
        controller._buffers.app.aput = AsyncMock()
        int_manager = INTManager(controller)
        assert len(inter_evc_evpl_flows_data) == 3
        await int_manager._remove_int_flows(inter_evc_evpl_flows_data)
        assert controller._buffers.app.aput.call_count == 3

    async def test__install_int_flows(self, inter_evc_evpl_flows_data, monkeypatch):
        """test__install_int_flows."""
        sleep_mock = AsyncMock()
        monkeypatch.setattr("asyncio.sleep", sleep_mock)
        controller = get_controller_mock()
        controller._buffers.app.aput = AsyncMock()
        int_manager = INTManager(controller)
        assert len(inter_evc_evpl_flows_data) == 3
        await int_manager._install_int_flows(inter_evc_evpl_flows_data)
        assert controller._buffers.app.aput.call_count == 3
        assert sleep_mock.call_count == 0

    def test__add_pps_evc_ids(self):

        """test_add_pps_evc_ids."""
        dpid_a = "00:00:00:00:00:00:00:01"
        intf_id_a = f"{dpid_a}:1"
        intf_id_z = f"{dpid_a}:2"
        evc_id = "3766c105686749"
        evcs = {
            evc_id: {
                "metadata": {"telemetry": {"enabled": True}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            }
        }
        controller = get_controller_mock()
        int_manager = INTManager(controller)
        pp = MagicMock()
        mock = MagicMock()
        int_manager.get_proxy_port_or_raise = mock
        mock.return_value = pp
        int_manager._add_pps_evc_ids(evcs)
        assert int_manager.get_proxy_port_or_raise.call_count == 2
        assert pp.evc_ids.add.call_count == 2
        pp.evc_ids.add.assert_called_with(evc_id)

    def test__discard_pps_evc_ids(self):

        """test_discard_pps_evc_ids."""
        dpid_a = "00:00:00:00:00:00:00:01"
        intf_id_a = f"{dpid_a}:1"
        intf_id_z = f"{dpid_a}:2"
        evc_id = "3766c105686749"
        evcs = {
            evc_id: {
                "metadata": {"telemetry": {"enabled": True}},
                "uni_a": {"interface_id": intf_id_a},
                "uni_z": {"interface_id": intf_id_z},
            }
        }
        controller = get_controller_mock()
        int_manager = INTManager(controller)
        pp = MagicMock()
        mock = MagicMock()
        int_manager.get_proxy_port_or_raise = mock
        mock.return_value = pp
        int_manager._discard_pps_evc_ids(evcs)
        assert int_manager.get_proxy_port_or_raise.call_count == 2
        assert pp.evc_ids.discard.call_count == 2
        pp.evc_ids.discard.assert_called_with(evc_id)


1			"""Test INTManager"""
2
3	1		import pytest
4
5	1		from unittest.mock import AsyncMock, MagicMock
6	1		from napps.kytos.telemetry_int.exceptions import ProxyPortSameSourceIntraEVC
7	1		from napps.kytos.telemetry_int.managers.int import INTManager
8	1		from napps.kytos.telemetry_int import exceptions
9	1		from kytos.core.common import EntityStatus
10
11	1		from kytos.lib.helpers import (
12			get_interface_mock,
13			get_controller_mock,
14			get_switch_mock,
15			)
16
17
18	1		class TestINTManager:
19			"""TestINTManager."""
20
21	1		def test_get_proxy_port_or_raise(self) -> None:
22			"""Test proxy_port_or_raise."""
23	1		dpid_a = "00:00:00:00:00:00:00:01"
24	1		mock_switch_a = get_switch_mock(dpid_a, 0x04)
25	1		mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
26	1		mock_interface_a.metadata = {}
27	1		intf_id = f"{dpid_a}:1"
28	1		controller = get_controller_mock()
29	1		evc_id = "3766c105686749"
30	1		int_manager = INTManager(controller)
31
32			# Initially the mocked interface and switch hasn't been associated in the ctrllr
33	1		with pytest.raises(exceptions.ProxyPortNotFound) as exc:
34	1		int_manager.get_proxy_port_or_raise(intf_id, evc_id)
35	1		assert f"interface {intf_id} not found" in str(exc)
36
37			# Now, proxy_port still hasn't been set yet
38	1		controller.get_interface_by_id = lambda x: mock_interface_a
39	1		with pytest.raises(exceptions.ProxyPortNotFound) as exc:
40	1		int_manager.get_proxy_port_or_raise(intf_id, evc_id)
41	1		assert f"proxy_port metadata not found in {intf_id}" in str(exc)
42
43			# Now, destination interface hasn't been mocked yet
44	1		mock_interface_a.metadata = {"proxy_port": 5}
45	1		with pytest.raises(exceptions.ProxyPortDestNotFound) as exc:
46	1		int_manager.get_proxy_port_or_raise(intf_id, evc_id)
47	1		assert "destination interface not found" in str(exc)
48
49	1		mock_interface_b = get_interface_mock("s1-eth5", 5, mock_switch_a)
50	1		mock_interface_b.metadata = {"looped": {"port_numbers": [5, 6]}}
51	1		mock_interface_a.switch.get_interface_by_port_no = lambda x: mock_interface_b
52			# Now all dependencies have been mocked and it should get the ProxyPort
53	1		pp = int_manager.get_proxy_port_or_raise(intf_id, evc_id)
54	1		assert pp.source == mock_interface_b
55
56	1		def test_load_uni_src_proxy_port(self) -> None:
57			"""Test test_load_uni_src_proxy_port."""
58	1		dpid_a = "00:00:00:00:00:00:00:01"
59	1		mock_switch_a = get_switch_mock(dpid_a, 0x04)
60	1		mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
61	1		mock_interface_a.metadata = {"proxy_port": 3}
62	1		mock_interface_z = get_interface_mock("s1-eth2", 2, mock_switch_a)
63	1		mock_interface_z.metadata = {"proxy_port": 5}
64	1		intf_id_a = f"{dpid_a}:1"
65	1		intf_id_z = f"{dpid_a}:2"
66	1		intf_id_a_1 = f"{dpid_a}:3"
67	1		intf_id_z_1 = f"{dpid_a}:5"
68
69	1		mock_interface_a_1 = get_interface_mock("s1-eth3", 3, mock_switch_a)
70	1		mock_interface_a_1.metadata = {"looped": {"port_numbers": [3, 4]}}
71	1		mock_interface_a_2 = get_interface_mock("s1-eth4", 4, mock_switch_a)
72	1		mock_interface_z_1 = get_interface_mock("s1-eth5", 5, mock_switch_a)
73	1		mock_interface_z_1.metadata = {"looped": {"port_numbers": [5, 6]}}
74	1		mock_interface_z_2 = get_interface_mock("s1-eth6", 6, mock_switch_a)
75
76	1		def get_interface_by_port_no(port_no):
77	1		data = {
78			1: mock_interface_a,
79			2: mock_interface_z,
80			3: mock_interface_a_1,
81			4: mock_interface_a_2,
82			5: mock_interface_z_1,
83			6: mock_interface_z_2,
84			}
85	1		return data[port_no]
86
87	1		def get_interface_by_id(intf_id):
88	1		data = {
89			intf_id_a: mock_interface_a,
90			intf_id_z: mock_interface_z,
91			}
92	1		return data[intf_id]
93
94	1		controller = get_controller_mock()
95	1		mock_switch_a.get_interface_by_port_no = get_interface_by_port_no
96	1		controller.get_interface_by_id = get_interface_by_id
97
98	1		evcs = {
99			"3766c105686749": {
100			"metadata": {"telemetry": {"enabled": True}},
101			"uni_a": {"interface_id": intf_id_a},
102			"uni_z": {"interface_id": intf_id_z},
103			},
104			"3766c105686748": {
105			"metadata": {"telemetry": {"enabled": True}},
106			"uni_a": {"interface_id": intf_id_a},
107			"uni_z": {"interface_id": intf_id_z},
108			},
109			"3766c105686747": {
110			"metadata": {"telemetry": {"enabled": False}},
111			"uni_a": {"interface_id": intf_id_a},
112			"uni_z": {"interface_id": intf_id_z},
113			},
114			}
115	1		int_manager = INTManager(controller)
116	1		int_manager.load_uni_src_proxy_ports(evcs)
117	1		assert len(int_manager.unis_src) == 2
118	1		assert int_manager.unis_src[intf_id_a] == intf_id_a_1
119	1		assert int_manager.unis_src[intf_id_z] == intf_id_z_1
120
121	1		assert len(int_manager.srcs_pp) == 2
122	1		assert int_manager.srcs_pp[intf_id_a_1].source == mock_interface_a_1
123	1		assert int_manager.srcs_pp[intf_id_a_1].destination == mock_interface_a_2
124	1		assert int_manager.srcs_pp[intf_id_z_1].source == mock_interface_z_1
125	1		assert int_manager.srcs_pp[intf_id_z_1].destination == mock_interface_z_2
126
127	1		assert int_manager.srcs_pp[intf_id_a_1].evc_ids == {
128			"3766c105686749",
129			"3766c105686748",
130			}
131	1		assert int_manager.srcs_pp[intf_id_z_1].evc_ids == {
132			"3766c105686749",
133			"3766c105686748",
134			}
135
136	1		async def test_handle_pp_link_down(self, monkeypatch):
137			"""Test test_handle_pp_link_down."""
138	1		int_manager = INTManager(MagicMock())
139	1		api_mock, link_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
140	1		link_mock.endpoint_a.id = "some_intf_id"
141	1		evc_id = "3766c105686748"
142	1		int_manager.srcs_pp[link_mock.endpoint_a.id] = pp_mock
143	1		pp_mock.evc_ids = {evc_id}
144
145	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
146	1		api_mock.get_evcs.return_value = {evc_id: {}}
147	1		int_manager.remove_int_flows = AsyncMock()
148
149	1		await int_manager.handle_pp_link_down(link_mock)
150	1		assert api_mock.get_evcs.call_count == 1
151	1		assert api_mock.get_evcs.call_count == 1
152	1		assert api_mock.get_evcs.call_args[1] == {
153			"metadata.telemetry.enabled": "true",
154			"metadata.telemetry.status": "UP",
155			}
156	1		assert int_manager.remove_int_flows.call_count == 1
157	1		args = int_manager.remove_int_flows.call_args[0]
158	1		assert evc_id in args[0]
159	1		assert "telemetry" in args[1]
160	1		telemetry = args[1]["telemetry"]
161	1		assert telemetry["enabled"]
162	1		assert telemetry["status"] == "DOWN"
163	1		assert telemetry["status_reason"] == ["proxy_port_down"]
164	1		assert "status_updated_at" in telemetry
165
166	1		async def test_handle_pp_link_up(self, monkeypatch):
167			"""Test handle_pp_link_up."""
168	1		int_manager = INTManager(MagicMock())
169	1		api_mock, link_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
170	1		link_mock.endpoint_a.id = "3"
171	1		pp_mock.status = EntityStatus.UP
172	1		link_mock.status = EntityStatus.UP
173	1		link_mock.status_reason = []
174	1		evc_id = "3766c105686748"
175	1		uni_a_id, uni_z_id = "1", "2"
176	1		src_a_id, src_z_id = "3", "5"
177	1		int_manager.srcs_pp[src_a_id] = pp_mock
178	1		int_manager.srcs_pp[src_z_id] = pp_mock
179	1		int_manager.unis_src[uni_a_id] = src_a_id
180	1		int_manager.unis_src[uni_z_id] = src_z_id
181	1		pp_mock.evc_ids = {evc_id}
182
183	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
184	1		api_mock.get_evcs.return_value = {
185			evc_id: {
186			"active": True,
187			"archived": False,
188			"uni_a": {"interface_id": uni_a_id},
189			"uni_z": {"interface_id": uni_z_id},
190			}
191			}
192	1		int_manager.install_int_flows = AsyncMock()
193	1		int_manager._validate_map_enable_evcs = MagicMock()
194
195	1		await int_manager.handle_pp_link_up(link_mock)
196	1		assert api_mock.get_evcs.call_count == 1
197	1		assert api_mock.get_evcs.call_args[1] == {
198			"metadata.telemetry.enabled": "true",
199			"metadata.telemetry.status": "DOWN",
200			}
201	1		assert int_manager.install_int_flows.call_count == 1
202	1		args = int_manager.install_int_flows.call_args[0]
203	1		assert "telemetry" in args[1]
204	1		telemetry_dict = args[1]["telemetry"]
205	1		expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
206	1		assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)
207	1		assert telemetry_dict["enabled"]
208	1		assert telemetry_dict["status"] == "UP"
209	1		assert not telemetry_dict["status_reason"]
210
211	1		async def test_handle_pp_metadata_removed(self, monkeypatch):
212			"""Test handle_pp_metadata_removed."""
213	1		int_manager = INTManager(MagicMock())
214	1		api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
215	1		intf_mock.id = "some_intf_id"
216	1		source_id = "some_source_id"
217	1		evc_id = "3766c105686748"
218	1		int_manager.unis_src[intf_mock.id] = source_id
219	1		int_manager.srcs_pp[source_id] = pp_mock
220	1		pp_mock.evc_ids = {evc_id}
221
222	1		assert "proxy_port" not in intf_mock.metadata
223	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
224	1		api_mock.get_evcs.return_value = {evc_id: {}}
225	1		int_manager.remove_int_flows = AsyncMock()
226
227	1		await int_manager.handle_pp_metadata_removed(intf_mock)
228	1		assert api_mock.get_evcs.call_count == 1
229	1		assert api_mock.get_evcs.call_count == 1
230	1		assert api_mock.get_evcs.call_args[1] == {
231			"metadata.telemetry.enabled": "true",
232			"metadata.telemetry.status": "UP",
233			}
234	1		assert int_manager.remove_int_flows.call_count == 1
235	1		args = int_manager.remove_int_flows.call_args[0]
236	1		assert evc_id in args[0]
237	1		assert "telemetry" in args[1]
238	1		telemetry = args[1]["telemetry"]
239	1		assert telemetry["enabled"]
240	1		assert telemetry["status"] == "DOWN"
241	1		assert telemetry["status_reason"] == ["proxy_port_metadata_removed"]
242	1		assert "status_updated_at" in telemetry
243
244	1		async def test_handle_pp_metadata_added(self, monkeypatch):
245			"""Test handle_pp_metadata_added."""
246	1		int_manager = INTManager(MagicMock())
247	1		api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
248	1		intf_mock.id = "some_intf_id"
249	1		source_id, source_port = "some_source_id", 2
250	1		intf_mock.metadata = {"proxy_port": source_port}
251	1		evc_id = "3766c105686748"
252	1		int_manager.unis_src[intf_mock.id] = source_id
253	1		int_manager.srcs_pp[source_id] = pp_mock
254	1		pp_mock.evc_ids = {evc_id}
255
256	1		assert "proxy_port" in intf_mock.metadata
257	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
258	1		api_mock.get_evcs.return_value = {evc_id: {}}
259	1		int_manager.disable_int = AsyncMock()
260	1		int_manager.enable_int = AsyncMock()
261
262	1		await int_manager.handle_pp_metadata_added(intf_mock)
263	1		assert api_mock.get_evcs.call_count == 1
264	1		assert api_mock.get_evcs.call_count == 1
265	1		assert api_mock.get_evcs.call_args[1] == {"metadata.telemetry.enabled": "true"}
266	1		assert int_manager.disable_int.call_count == 1
267	1		assert int_manager.enable_int.call_count == 1
268
269	1		async def test_handle_pp_metadata_added_no_change(self, monkeypatch):
270			"""Test handle_pp_metadata_added no change."""
271	1		int_manager = INTManager(MagicMock())
272	1		api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
273	1		intf_mock.id = "some_intf_id"
274	1		source_id, source_port = "some_source_id", 2
275	1		source_intf = MagicMock()
276	1		intf_mock.metadata = {"proxy_port": source_port}
277	1		evc_id = "3766c105686748"
278	1		int_manager.unis_src[intf_mock.id] = source_id
279	1		int_manager.srcs_pp[source_id] = pp_mock
280	1		pp_mock.evc_ids = {evc_id}
281
282			# Simulating that the current and new proxy_port source are the same
283	1		pp_mock.source = source_intf
284	1		intf_mock.switch.get_interface_by_port_no.return_value = source_intf
285
286	1		assert "proxy_port" in intf_mock.metadata
287	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
288	1		api_mock.get_evcs.return_value = {evc_id: {}}
289	1		int_manager.disable_int = AsyncMock()
290	1		int_manager.enable_int = AsyncMock()
291
292	1		await int_manager.handle_pp_metadata_added(intf_mock)
293	1		assert not api_mock.get_evcs.call_count
294	1		assert not int_manager.disable_int.call_count
295	1		assert not int_manager.enable_int.call_count
296
297	1		async def test_handle_pp_metadata_added_no_affected(self, monkeypatch):
298			"""Test handle_pp_metadata_added no affected evcs."""
299	1		int_manager = INTManager(MagicMock())
300	1		api_mock, intf_mock, pp_mock = AsyncMock(), MagicMock(), MagicMock()
301	1		intf_mock.id = "some_intf_id"
302	1		source_id, source_port = "some_source_id", 2
303	1		intf_mock.metadata = {"proxy_port": source_port}
304	1		evc_id = "3766c105686748"
305	1		int_manager.unis_src[intf_mock.id] = source_id
306	1		int_manager.srcs_pp[source_id] = pp_mock
307	1		pp_mock.evc_ids = {evc_id}
308
309	1		assert "proxy_port" in intf_mock.metadata
310	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
311
312			# Simulating returning no EVCs that were enabled and UP
313	1		api_mock.get_evcs.return_value = {}
314	1		int_manager.disable_int = AsyncMock()
315	1		int_manager.enable_int = AsyncMock()
316
317	1		await int_manager.handle_pp_metadata_added(intf_mock)
318	1		assert api_mock.get_evcs.call_count == 1
319	1		assert api_mock.get_evcs.call_count == 1
320	1		assert api_mock.get_evcs.call_args[1] == {
321			"metadata.telemetry.enabled": "true",
322			}
323	1		assert not int_manager.disable_int.call_count
324	1		assert not int_manager.enable_int.call_count
325
326	1		async def test_disable_int_metadata(self, monkeypatch) -> None:
327			"""Test disable INT metadata args."""
328	1		controller = MagicMock()
329	1		api_mock = AsyncMock()
330	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
331
332	1		int_manager = INTManager(controller)
333	1		int_manager._remove_int_flows = AsyncMock()
334	1		await int_manager.disable_int({}, False)
335
336	1		assert api_mock.add_evcs_metadata.call_count == 1
337	1		args = api_mock.add_evcs_metadata.call_args[0]
338	1		assert args[0] == {}
339	1		assert "telemetry" in args[1]
340	1		telemetry_dict = args[1]["telemetry"]
341	1		expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
342	1		assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)
343
344	1		assert not telemetry_dict["enabled"]
345	1		assert telemetry_dict["status"] == "DOWN"
346	1		assert telemetry_dict["status_reason"] == ["disabled"]
347
348	1		assert args[2] is False
349
350	1		async def test_enable_int_metadata(self, monkeypatch) -> None:
351			"""Test enable INT metadata args."""
352	1		controller = MagicMock()
353	1		api_mock = AsyncMock()
354	1		stored_flows_mock = AsyncMock()
355	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
356	1		monkeypatch.setattr(
357			"napps.kytos.telemetry_int.utils.get_found_stored_flows", stored_flows_mock
358			)
359
360	1		int_manager = INTManager(controller)
361	1		int_manager.remove_int_flows = AsyncMock()
362	1		await int_manager.enable_int({}, False)
363
364	1		assert stored_flows_mock.call_count == 1
365	1		assert api_mock.add_evcs_metadata.call_count == 1
366	1		args = api_mock.add_evcs_metadata.call_args[0]
367	1		assert args[0] == {}
368	1		assert "telemetry" in args[1]
369	1		telemetry_dict = args[1]["telemetry"]
370	1		expected_keys = ["enabled", "status", "status_reason", "status_updated_at"]
371	1		assert sorted(list(telemetry_dict.keys())) == sorted(expected_keys)
372
373	1		assert telemetry_dict["enabled"] is True
374	1		assert telemetry_dict["status"] == "UP"
375	1		assert telemetry_dict["status_reason"] == []
376
377	1		async def test_redeploy_int(self, monkeypatch) -> None:
378			"""Test redeploy int."""
379	1		controller = MagicMock()
380	1		api_mock = AsyncMock()
381	1		stored_flows_mock = AsyncMock()
382	1		monkeypatch.setattr("napps.kytos.telemetry_int.managers.int.api", api_mock)
383	1		monkeypatch.setattr(
384			"napps.kytos.telemetry_int.utils.get_found_stored_flows", stored_flows_mock
385			)
386
387	1		int_manager = INTManager(controller)
388	1		int_manager._remove_int_flows = AsyncMock()
389	1		int_manager._install_int_flows = AsyncMock()
390
391	1		dpid_a = "00:00:00:00:00:00:00:01"
392	1		intf_id_a = f"{dpid_a}:1"
393	1		intf_id_z = f"{dpid_a}:2"
394	1		evc_id = "3766c105686749"
395	1		evcs = {
396			evc_id: {
397			"metadata": {"telemetry": {"enabled": True}},
398			"uni_a": {"interface_id": intf_id_a},
399			"uni_z": {"interface_id": intf_id_z},
400			}
401			}
402	1		int_manager._validate_map_enable_evcs = MagicMock()
403	1		await int_manager.redeploy_int(evcs)
404
405	1		assert stored_flows_mock.call_count == 1
406	1		assert int_manager._remove_int_flows.call_count == 1
407	1		assert api_mock.get_stored_flows.call_count == 1
408	1		assert int_manager._install_int_flows.call_count == 1
409
410	1		def test_validate_intra_evc_different_proxy_ports(self) -> None:
411			"""Test _validate_intra_evc_different_proxy_ports."""
412	1		pp_a, pp_z, controller = MagicMock(), MagicMock(), MagicMock()
413	1		evc = {
414			"id": "some_id",
415			"uni_a": {"proxy_port": pp_a, "interface_id": "00:00:00:00:00:00:00:01:1"},
416			"uni_z": {"proxy_port": pp_z, "interface_id": "00:00:00:00:00:00:00:01:2"},
417			}
418
419	1		int_manager = INTManager(controller)
420	1		int_manager._validate_intra_evc_different_proxy_ports(evc)
421
422	1		source = MagicMock()
423	1		pp_a.source, pp_z.source = source, source
424	1		with pytest.raises(ProxyPortSameSourceIntraEVC):
425	1		int_manager._validate_intra_evc_different_proxy_ports(evc)
426
427	1		async def test__remove_int_flows(self, inter_evc_evpl_flows_data) -> None:
428			"""test _remove_int_flows."""
429	1		controller = get_controller_mock()
430	1		controller._buffers.app.aput = AsyncMock()
431	1		int_manager = INTManager(controller)
432	1		assert len(inter_evc_evpl_flows_data) == 3
433	1		await int_manager._remove_int_flows(inter_evc_evpl_flows_data)
434	1		assert controller._buffers.app.aput.call_count == 3
435
436	1		async def test__install_int_flows(self, inter_evc_evpl_flows_data, monkeypatch):
437			"""test__install_int_flows."""
438	1		sleep_mock = AsyncMock()
439	1		monkeypatch.setattr("asyncio.sleep", sleep_mock)
440	1		controller = get_controller_mock()
441	1		controller._buffers.app.aput = AsyncMock()
442	1		int_manager = INTManager(controller)
443	1		assert len(inter_evc_evpl_flows_data) == 3
444	1		await int_manager._install_int_flows(inter_evc_evpl_flows_data)
445	1		assert controller._buffers.app.aput.call_count == 3
446	1		assert sleep_mock.call_count == 0
447
448	1	View Code Duplication	def test__add_pps_evc_ids(self):
			0 ignored issues – show Duplication introduced 2024-04-03 14:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
449			"""test_add_pps_evc_ids."""
450	1		dpid_a = "00:00:00:00:00:00:00:01"
451	1		intf_id_a = f"{dpid_a}:1"
452	1		intf_id_z = f"{dpid_a}:2"
453	1		evc_id = "3766c105686749"
454	1		evcs = {
455			evc_id: {
456			"metadata": {"telemetry": {"enabled": True}},
457			"uni_a": {"interface_id": intf_id_a},
458			"uni_z": {"interface_id": intf_id_z},
459			}
460			}
461	1		controller = get_controller_mock()
462	1		int_manager = INTManager(controller)
463	1		pp = MagicMock()
464	1		mock = MagicMock()
465	1		int_manager.get_proxy_port_or_raise = mock
466	1		mock.return_value = pp
467	1		int_manager._add_pps_evc_ids(evcs)
468	1		assert int_manager.get_proxy_port_or_raise.call_count == 2
469	1		assert pp.evc_ids.add.call_count == 2
470	1		pp.evc_ids.add.assert_called_with(evc_id)
471
472	1	View Code Duplication	def test__discard_pps_evc_ids(self):
			0 ignored issues – show Duplication introduced 2024-04-03 14:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
473			"""test_discard_pps_evc_ids."""
474	1		dpid_a = "00:00:00:00:00:00:00:01"
475	1		intf_id_a = f"{dpid_a}:1"
476	1		intf_id_z = f"{dpid_a}:2"
477	1		evc_id = "3766c105686749"
478	1		evcs = {
479			evc_id: {
480			"metadata": {"telemetry": {"enabled": True}},
481			"uni_a": {"interface_id": intf_id_a},
482			"uni_z": {"interface_id": intf_id_z},
483			}
484			}
485	1		controller = get_controller_mock()
486	1		int_manager = INTManager(controller)
487	1		pp = MagicMock()
488	1		mock = MagicMock()
489	1		int_manager.get_proxy_port_or_raise = mock
490	1		mock.return_value = pp
491	1		int_manager._discard_pps_evc_ids(evcs)
492	1		assert int_manager.get_proxy_port_or_raise.call_count == 2
493	1		assert pp.evc_ids.discard.call_count == 2
494			pp.evc_ids.discard.assert_called_with(evc_id)
495

kytos-ng / telemetry_int

Push — master ( 6f27fa...aec56c )

TestINTManager.test__install_int_flows() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like