test_main.TestMain.test_on_link_down() - Code Metrics - Inspection of "Merge pull request #91 from kytos-ng/feat/event_un..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( fb285f...124bde )

by Vinicius

created 2024-05-01 13:42 UTC

test_main.TestMain.test_on_link_down() A

↳ Parent: test_main

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	5
ccs	4
cts	4
cp	1
crap	1
rs	10
c	0
b	0
f	0

"""Test Main methods."""

import pytest

from unittest.mock import AsyncMock, MagicMock, patch
from napps.kytos.telemetry_int.main import Main
from napps.kytos.telemetry_int import utils
from kytos.lib.helpers import get_controller_mock, get_test_client
from kytos.core.events import KytosEvent


class TestMain:
    """Tests for the Main class."""

    def setup_method(self):
        """Setup."""
        patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
        # pylint: disable=import-outside-toplevel
        controller = get_controller_mock()
        self.napp = Main(controller)
        self.api_client = get_test_client(controller, self.napp)
        self.base_endpoint = "kytos/telemetry_int/v1"

    async def test_enable_telemetry(self, monkeypatch) -> None:

        """Test enable telemetry."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {"metadata": {"telemetry": {"enabled": False}}}
        }

        self.napp.int_manager = AsyncMock()

        endpoint = f"{self.base_endpoint}/evc/enable"
        response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]})
        assert self.napp.int_manager.enable_int.call_count == 1
        assert self.napp.int_manager._remove_int_flows.call_count == 1
        assert response.status_code == 201
        assert response.json() == [evc_id]

    async def test_redeploy_telemetry(self, monkeypatch) -> None:

        """Test redeploy telemetry."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {"metadata": {"telemetry": {"enabled": False}}}
        }

        self.napp.int_manager = AsyncMock()

        endpoint = f"{self.base_endpoint}/evc/redeploy"
        response = await self.api_client.patch(endpoint, json={"evc_ids": [evc_id]})
        assert self.napp.int_manager.redeploy_int.call_count == 1
        assert response.status_code == 201
        assert response.json() == [evc_id]

    @pytest.mark.parametrize("route", ["/evc/enable", "/evc/disable"])
    async def test_en_dis_openapi_validation(self, route: str) -> None:
        """Test OpenAPI enable/disable basic validation."""
        endpoint = f"{self.base_endpoint}{route}"
        # wrong evc_ids payload data type
        response = await self.api_client.post(endpoint, json={"evc_ids": 1})
        assert response.status_code == 400
        assert "evc_ids" in response.json()["description"]

    async def test_disable_telemetry(self, monkeypatch) -> None:

        """Test disable telemetry."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {"metadata": {"telemetry": {"enabled": True}}}
        }

        self.napp.int_manager = AsyncMock()

        endpoint = f"{self.base_endpoint}/evc/disable"
        response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]})
        assert self.napp.int_manager.disable_int.call_count == 1
        assert response.status_code == 200
        assert response.json() == [evc_id]

    async def test_get_enabled_evcs(self, monkeypatch) -> None:
        """Test get enabled evcs."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {"metadata": {"telemetry": {"enabled": True}}},
        }

        endpoint = f"{self.base_endpoint}/evc"
        response = await self.api_client.get(endpoint)
        assert api_mock.get_evcs.call_args[1] == {"metadata.telemetry.enabled": "true"}
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert evc_id in data

    async def test_get_evc_compare(self, monkeypatch) -> None:
        """Test get evc compre ok case."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        evc_id = "1"
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {
                "id": evc_id,
                "name": "evc",
                "metadata": {"telemetry": {"enabled": True}},
            },
        }
        api_mock.get_stored_flows.side_effect = [
            {flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
            {flow.cookie: [{"id": "some_2", "match": {"in_port": 1}}]},
        ]

        endpoint = f"{self.base_endpoint}/evc/compare"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 0

    async def test_get_evc_compare_wrong_metadata(self, monkeypatch) -> None:
        """Test get evc compre wrong_metadata_has_int_flows case."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        evc_id = "1"
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {"id": evc_id, "name": "evc", "metadata": {}},
        }
        api_mock.get_stored_flows.side_effect = [
            {flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
            {flow.cookie: [{"id": "some_2", "match": {"in_port": 1}}]},
        ]

        endpoint = f"{self.base_endpoint}/evc/compare"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == evc_id
        assert data[0]["compare_reason"] == ["wrong_metadata_has_int_flows"]
        assert data[0]["name"] == "evc"

    async def test_get_evc_compare_missing_some_int_flows(self, monkeypatch) -> None:
        """Test get evc compre missing_some_int_flows case."""
        api_mock, flow = AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        evc_id = "1"
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )

        evc_id = utils.get_id_from_cookie(flow.cookie)
        api_mock.get_evcs.return_value = {
            evc_id: {
                "id": evc_id,
                "name": "evc",
                "metadata": {"telemetry": {"enabled": True}},
            },
        }
        api_mock.get_stored_flows.side_effect = [
            {flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
            {
                flow.cookie: [
                    {"id": "some_2", "match": {"in_port": 1}},
                    {"id": "some_3", "match": {"in_port": 1}},
                ]
            },
        ]

        endpoint = f"{self.base_endpoint}/evc/compare"
        response = await self.api_client.get(endpoint)
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == evc_id
        assert data[0]["compare_reason"] == ["missing_some_int_flows"]
        assert data[0]["name"] == "evc"

    async def test_on_table_enabled(self) -> None:
        """Test on_table_enabled."""
        assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
        await self.napp.on_table_enabled(
            KytosEvent(content={"telemetry_int": {"evpl": 22, "epl": 33}})
        )
        assert self.napp.int_manager.flow_builder.table_group == {"evpl": 22, "epl": 33}
        assert self.napp.controller.buffers.app.aput.call_count == 1

    async def test_on_table_enabled_no_group(self) -> None:
        """Test on_table_enabled no group."""
        await self.napp.on_table_enabled(
            KytosEvent(content={"mef_eline": {"evpl": 22, "epl": 33}})
        )
        assert not self.napp.controller.buffers.app.aput.call_count

    async def test_on_evc_deleted(self) -> None:
        """Test on_evc_deleted."""
        content = {"metadata": {"telemetry": {"enabled": True}}, "evc_id": "some_id"}
        self.napp.int_manager.disable_int = AsyncMock()
        await self.napp.on_evc_deleted(KytosEvent(content=content))
        assert self.napp.int_manager.disable_int.call_count == 1

    async def test_on_uni_active_updated(self, monkeypatch) -> None:
        """Test on UNI active updated."""
        api_mock = AsyncMock()
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock,
        )
        content = {
            "metadata": {"telemetry": {"enabled": True}},
            "evc_id": "some_id",
            "active": True,
        }
        await self.napp.on_uni_active_updated(KytosEvent(content=content))
        assert api_mock.add_evcs_metadata.call_count == 1
        args = api_mock.add_evcs_metadata.call_args[0][1]
        assert args["telemetry"]["status"] == "UP"

        content["active"] = False
        await self.napp.on_uni_active_updated(KytosEvent(content=content))
        assert api_mock.add_evcs_metadata.call_count == 2
        args = api_mock.add_evcs_metadata.call_args[0][1]
        assert args["telemetry"]["status"] == "DOWN"

    async def test_on_evc_undeployed(self) -> None:
        """Test on_evc_undeployed."""
        content = {
            "enabled": False,
            "metadata": {"telemetry": {"enabled": False}},
            "evc_id": "some_id",
        }
        self.napp.int_manager.remove_int_flows = AsyncMock()
        await self.napp.on_evc_undeployed(KytosEvent(content=content))
        assert self.napp.int_manager.remove_int_flows.call_count == 0

        content["metadata"]["telemetry"]["enabled"] = True
        await self.napp.on_evc_undeployed(KytosEvent(content=content))
        assert self.napp.int_manager.remove_int_flows.call_count == 1

    async def test_on_evc_redeployed_link(self) -> None:
        """Test on redeployed_link_down|redeployed_link_up."""
        content = {
            "enabled": True,
            "metadata": {"telemetry": {"enabled": False}},
            "evc_id": "some_id",
        }
        self.napp.int_manager.redeploy_int = AsyncMock()
        await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
        assert self.napp.int_manager.redeploy_int.call_count == 0

        content["metadata"]["telemetry"]["enabled"] = True
        await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
        assert self.napp.int_manager.redeploy_int.call_count == 1

    async def test_on_evc_error_redeployed_link_down(self) -> None:
        """Test error_redeployed_link_down."""
        content = {
            "enabled": True,
            "metadata": {"telemetry": {"enabled": False}},
            "evc_id": "some_id",
        }
        self.napp.int_manager.remove_int_flows = AsyncMock()
        await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
        assert self.napp.int_manager.remove_int_flows.call_count == 0

        content["metadata"]["telemetry"]["enabled"] = True
        await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
        assert self.napp.int_manager.remove_int_flows.call_count == 1

    async def test_on_link_down(self) -> None:
        """Test on link_down."""
        self.napp.int_manager.handle_pp_link_down = AsyncMock()
        await self.napp.on_link_down(KytosEvent(content={"link": MagicMock()}))
        assert self.napp.int_manager.handle_pp_link_down.call_count == 1

    async def test_on_link_up(self) -> None:
        """Test on link_up."""
        self.napp.int_manager.handle_pp_link_up = AsyncMock()
        await self.napp.on_link_up(KytosEvent(content={"link": MagicMock()}))
        assert self.napp.int_manager.handle_pp_link_up.call_count == 1

    async def test_on_table_enabled_error(self, monkeypatch) -> None:
        """Test on_table_enabled error case."""
        assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
        log_mock = MagicMock()
        monkeypatch.setattr("napps.kytos.telemetry_int.main.log", log_mock)
        await self.napp.on_table_enabled(
            KytosEvent(content={"telemetry_int": {"invalid": 1}})
        )
        assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
        assert log_mock.error.call_count == 1
        assert not self.napp.controller.buffers.app.aput.call_count

    async def test_on_flow_mod_error(self, monkeypatch) -> None:
        """Test on_flow_mod_error."""
        api_mock_main, api_mock_int, flow = AsyncMock(), AsyncMock(), MagicMock()
        flow.cookie = 0xA800000000000001
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.main.api",
            api_mock_main,
        )
        monkeypatch.setattr(
            "napps.kytos.telemetry_int.managers.int.api",
            api_mock_int,
        )
        cookie = utils.get_id_from_cookie(flow.cookie)
        api_mock_main.get_evc.return_value = {
            cookie: {"metadata": {"telemetry": {"enabled": True}}}
        }
        api_mock_int.get_stored_flows.return_value = {cookie: [MagicMock()]}
        self.napp.int_manager._remove_int_flows = AsyncMock()

        event = KytosEvent(content={"flow": flow, "error_command": "add"})
        await self.napp.on_flow_mod_error(event)

        assert api_mock_main.get_evc.call_count == 1
        assert api_mock_int.get_stored_flows.call_count == 1
        assert api_mock_int.add_evcs_metadata.call_count == 1
        assert self.napp.int_manager._remove_int_flows.call_count == 1

    async def test_on_mef_eline_evcs_loaded(self):
        """Test on_mef_eline_evcs_loaded."""
        evcs = {"1": {}, "2": {}}
        event = KytosEvent(content=evcs)
        self.napp.int_manager = MagicMock()
        await self.napp.on_mef_eline_evcs_loaded(event)
        self.napp.int_manager.load_uni_src_proxy_ports.assert_called_with(evcs)

    async def test_on_intf_metadata_remove(self):
        """Test on_intf_metadata_removed."""
        intf = MagicMock()
        event = KytosEvent(content={"interface": intf})
        self.napp.int_manager = MagicMock()
        await self.napp.on_intf_metadata_removed(event)
        self.napp.int_manager.handle_pp_metadata_removed.assert_called_with(intf)

    async def test_on_intf_metadata_added(self):
        """Test on_intf_metadata_added."""
        intf = MagicMock()
        event = KytosEvent(content={"interface": intf})
        self.napp.int_manager = MagicMock()
        await self.napp.on_intf_metadata_added(event)
        self.napp.int_manager.handle_pp_metadata_added.assert_called_with(intf)


1			"""Test Main methods."""
2
3	1		import pytest
4
5	1		from unittest.mock import AsyncMock, MagicMock, patch
6	1		from napps.kytos.telemetry_int.main import Main
7	1		from napps.kytos.telemetry_int import utils
8	1		from kytos.lib.helpers import get_controller_mock, get_test_client
9	1		from kytos.core.events import KytosEvent
10
11
12	1		class TestMain:
13			"""Tests for the Main class."""
14
15	1		def setup_method(self):
16			"""Setup."""
17	1		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
18			# pylint: disable=import-outside-toplevel
19	1		controller = get_controller_mock()
20	1		self.napp = Main(controller)
21	1		self.api_client = get_test_client(controller, self.napp)
22	1		self.base_endpoint = "kytos/telemetry_int/v1"
23
24	1	View Code Duplication	async def test_enable_telemetry(self, monkeypatch) -> None:
			0 ignored issues – show Duplication introduced 2023-10-04 15:18 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
25			"""Test enable telemetry."""
26	1		api_mock, flow = AsyncMock(), MagicMock()
27	1		flow.cookie = 0xA800000000000001
28	1		monkeypatch.setattr(
29			"napps.kytos.telemetry_int.main.api",
30			api_mock,
31			)
32
33	1		evc_id = utils.get_id_from_cookie(flow.cookie)
34	1		api_mock.get_evcs.return_value = {
35			evc_id: {"metadata": {"telemetry": {"enabled": False}}}
36			}
37
38	1		self.napp.int_manager = AsyncMock()
39
40	1		endpoint = f"{self.base_endpoint}/evc/enable"
41	1		response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]})
42	1		assert self.napp.int_manager.enable_int.call_count == 1
43	1		assert self.napp.int_manager._remove_int_flows.call_count == 1
44	1		assert response.status_code == 201
45	1		assert response.json() == [evc_id]
46
47	1	View Code Duplication	async def test_redeploy_telemetry(self, monkeypatch) -> None:
			0 ignored issues – show Duplication introduced 2023-10-04 15:18 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
48			"""Test redeploy telemetry."""
49	1		api_mock, flow = AsyncMock(), MagicMock()
50	1		flow.cookie = 0xA800000000000001
51	1		monkeypatch.setattr(
52			"napps.kytos.telemetry_int.main.api",
53			api_mock,
54			)
55
56	1		evc_id = utils.get_id_from_cookie(flow.cookie)
57	1		api_mock.get_evcs.return_value = {
58			evc_id: {"metadata": {"telemetry": {"enabled": False}}}
59			}
60
61	1		self.napp.int_manager = AsyncMock()
62
63	1		endpoint = f"{self.base_endpoint}/evc/redeploy"
64	1		response = await self.api_client.patch(endpoint, json={"evc_ids": [evc_id]})
65	1		assert self.napp.int_manager.redeploy_int.call_count == 1
66	1		assert response.status_code == 201
67	1		assert response.json() == [evc_id]
68
69	1		@pytest.mark.parametrize("route", ["/evc/enable", "/evc/disable"])
70	1		async def test_en_dis_openapi_validation(self, route: str) -> None:
71			"""Test OpenAPI enable/disable basic validation."""
72	1		endpoint = f"{self.base_endpoint}{route}"
73			# wrong evc_ids payload data type
74	1		response = await self.api_client.post(endpoint, json={"evc_ids": 1})
75	1		assert response.status_code == 400
76	1		assert "evc_ids" in response.json()["description"]
77
78	1	View Code Duplication	async def test_disable_telemetry(self, monkeypatch) -> None:
			0 ignored issues – show Duplication introduced 2023-11-29 13:03 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
79			"""Test disable telemetry."""
80	1		api_mock, flow = AsyncMock(), MagicMock()
81	1		flow.cookie = 0xA800000000000001
82	1		monkeypatch.setattr(
83			"napps.kytos.telemetry_int.main.api",
84			api_mock,
85			)
86
87	1		evc_id = utils.get_id_from_cookie(flow.cookie)
88	1		api_mock.get_evcs.return_value = {
89			evc_id: {"metadata": {"telemetry": {"enabled": True}}}
90			}
91
92	1		self.napp.int_manager = AsyncMock()
93
94	1		endpoint = f"{self.base_endpoint}/evc/disable"
95	1		response = await self.api_client.post(endpoint, json={"evc_ids": [evc_id]})
96	1		assert self.napp.int_manager.disable_int.call_count == 1
97	1		assert response.status_code == 200
98	1		assert response.json() == [evc_id]
99
100	1		async def test_get_enabled_evcs(self, monkeypatch) -> None:
101			"""Test get enabled evcs."""
102	1		api_mock, flow = AsyncMock(), MagicMock()
103	1		flow.cookie = 0xA800000000000001
104	1		monkeypatch.setattr(
105			"napps.kytos.telemetry_int.main.api",
106			api_mock,
107			)
108
109	1		evc_id = utils.get_id_from_cookie(flow.cookie)
110	1		api_mock.get_evcs.return_value = {
111			evc_id: {"metadata": {"telemetry": {"enabled": True}}},
112			}
113
114	1		endpoint = f"{self.base_endpoint}/evc"
115	1		response = await self.api_client.get(endpoint)
116	1		assert api_mock.get_evcs.call_args[1] == {"metadata.telemetry.enabled": "true"}
117	1		assert response.status_code == 200
118	1		data = response.json()
119	1		assert len(data) == 1
120	1		assert evc_id in data
121
122	1		async def test_get_evc_compare(self, monkeypatch) -> None:
123			"""Test get evc compre ok case."""
124	1		api_mock, flow = AsyncMock(), MagicMock()
125	1		flow.cookie = 0xA800000000000001
126	1		evc_id = "1"
127	1		monkeypatch.setattr(
128			"napps.kytos.telemetry_int.main.api",
129			api_mock,
130			)
131
132	1		evc_id = utils.get_id_from_cookie(flow.cookie)
133	1		api_mock.get_evcs.return_value = {
134			evc_id: {
135			"id": evc_id,
136			"name": "evc",
137			"metadata": {"telemetry": {"enabled": True}},
138			},
139			}
140	1		api_mock.get_stored_flows.side_effect = [
141			{flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
142			{flow.cookie: [{"id": "some_2", "match": {"in_port": 1}}]},
143			]
144
145	1		endpoint = f"{self.base_endpoint}/evc/compare"
146	1		response = await self.api_client.get(endpoint)
147	1		assert response.status_code == 200
148	1		data = response.json()
149	1		assert len(data) == 0
150
151	1		async def test_get_evc_compare_wrong_metadata(self, monkeypatch) -> None:
152			"""Test get evc compre wrong_metadata_has_int_flows case."""
153	1		api_mock, flow = AsyncMock(), MagicMock()
154	1		flow.cookie = 0xA800000000000001
155	1		evc_id = "1"
156	1		monkeypatch.setattr(
157			"napps.kytos.telemetry_int.main.api",
158			api_mock,
159			)
160
161	1		evc_id = utils.get_id_from_cookie(flow.cookie)
162	1		api_mock.get_evcs.return_value = {
163			evc_id: {"id": evc_id, "name": "evc", "metadata": {}},
164			}
165	1		api_mock.get_stored_flows.side_effect = [
166			{flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
167			{flow.cookie: [{"id": "some_2", "match": {"in_port": 1}}]},
168			]
169
170	1		endpoint = f"{self.base_endpoint}/evc/compare"
171	1		response = await self.api_client.get(endpoint)
172	1		assert response.status_code == 200
173	1		data = response.json()
174	1		assert len(data) == 1
175	1		assert data[0]["id"] == evc_id
176	1		assert data[0]["compare_reason"] == ["wrong_metadata_has_int_flows"]
177	1		assert data[0]["name"] == "evc"
178
179	1		async def test_get_evc_compare_missing_some_int_flows(self, monkeypatch) -> None:
180			"""Test get evc compre missing_some_int_flows case."""
181	1		api_mock, flow = AsyncMock(), MagicMock()
182	1		flow.cookie = 0xA800000000000001
183	1		evc_id = "1"
184	1		monkeypatch.setattr(
185			"napps.kytos.telemetry_int.main.api",
186			api_mock,
187			)
188
189	1		evc_id = utils.get_id_from_cookie(flow.cookie)
190	1		api_mock.get_evcs.return_value = {
191			evc_id: {
192			"id": evc_id,
193			"name": "evc",
194			"metadata": {"telemetry": {"enabled": True}},
195			},
196			}
197	1		api_mock.get_stored_flows.side_effect = [
198			{flow.cookie: [{"id": "some_1", "match": {"in_port": 1}}]},
199			{
200			flow.cookie: [
201			{"id": "some_2", "match": {"in_port": 1}},
202			{"id": "some_3", "match": {"in_port": 1}},
203			]
204			},
205			]
206
207	1		endpoint = f"{self.base_endpoint}/evc/compare"
208	1		response = await self.api_client.get(endpoint)
209	1		assert response.status_code == 200
210	1		data = response.json()
211	1		assert len(data) == 1
212	1		assert data[0]["id"] == evc_id
213	1		assert data[0]["compare_reason"] == ["missing_some_int_flows"]
214	1		assert data[0]["name"] == "evc"
215
216	1		async def test_on_table_enabled(self) -> None:
217			"""Test on_table_enabled."""
218	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
219	1		await self.napp.on_table_enabled(
220			KytosEvent(content={"telemetry_int": {"evpl": 22, "epl": 33}})
221			)
222	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 22, "epl": 33}
223	1		assert self.napp.controller.buffers.app.aput.call_count == 1
224
225	1		async def test_on_table_enabled_no_group(self) -> None:
226			"""Test on_table_enabled no group."""
227	1		await self.napp.on_table_enabled(
228			KytosEvent(content={"mef_eline": {"evpl": 22, "epl": 33}})
229			)
230	1		assert not self.napp.controller.buffers.app.aput.call_count
231
232	1		async def test_on_evc_deleted(self) -> None:
233			"""Test on_evc_deleted."""
234	1		content = {"metadata": {"telemetry": {"enabled": True}}, "evc_id": "some_id"}
235	1		self.napp.int_manager.disable_int = AsyncMock()
236	1		await self.napp.on_evc_deleted(KytosEvent(content=content))
237	1		assert self.napp.int_manager.disable_int.call_count == 1
238
239	1		async def test_on_uni_active_updated(self, monkeypatch) -> None:
240			"""Test on UNI active updated."""
241	1		api_mock = AsyncMock()
242	1		monkeypatch.setattr(
243			"napps.kytos.telemetry_int.main.api",
244			api_mock,
245			)
246	1		content = {
247			"metadata": {"telemetry": {"enabled": True}},
248			"evc_id": "some_id",
249			"active": True,
250			}
251	1		await self.napp.on_uni_active_updated(KytosEvent(content=content))
252	1		assert api_mock.add_evcs_metadata.call_count == 1
253	1		args = api_mock.add_evcs_metadata.call_args[0][1]
254	1		assert args["telemetry"]["status"] == "UP"
255
256	1		content["active"] = False
257	1		await self.napp.on_uni_active_updated(KytosEvent(content=content))
258	1		assert api_mock.add_evcs_metadata.call_count == 2
259	1		args = api_mock.add_evcs_metadata.call_args[0][1]
260	1		assert args["telemetry"]["status"] == "DOWN"
261
262	1		async def test_on_evc_undeployed(self) -> None:
263			"""Test on_evc_undeployed."""
264	1		content = {
265			"enabled": False,
266			"metadata": {"telemetry": {"enabled": False}},
267			"evc_id": "some_id",
268			}
269	1		self.napp.int_manager.remove_int_flows = AsyncMock()
270	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
271	1		assert self.napp.int_manager.remove_int_flows.call_count == 0
272
273	1		content["metadata"]["telemetry"]["enabled"] = True
274	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
275	1		assert self.napp.int_manager.remove_int_flows.call_count == 1
276
277	1		async def test_on_evc_redeployed_link(self) -> None:
278			"""Test on redeployed_link_down\|redeployed_link_up."""
279	1		content = {
280			"enabled": True,
281			"metadata": {"telemetry": {"enabled": False}},
282			"evc_id": "some_id",
283			}
284	1		self.napp.int_manager.redeploy_int = AsyncMock()
285	1		await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
286	1		assert self.napp.int_manager.redeploy_int.call_count == 0
287
288	1		content["metadata"]["telemetry"]["enabled"] = True
289	1		await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
290	1		assert self.napp.int_manager.redeploy_int.call_count == 1
291
292	1		async def test_on_evc_error_redeployed_link_down(self) -> None:
293			"""Test error_redeployed_link_down."""
294	1		content = {
295			"enabled": True,
296			"metadata": {"telemetry": {"enabled": False}},
297			"evc_id": "some_id",
298			}
299	1		self.napp.int_manager.remove_int_flows = AsyncMock()
300	1		await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
301	1		assert self.napp.int_manager.remove_int_flows.call_count == 0
302
303	1		content["metadata"]["telemetry"]["enabled"] = True
304	1		await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
305	1		assert self.napp.int_manager.remove_int_flows.call_count == 1
306
307	1		async def test_on_link_down(self) -> None:
308			"""Test on link_down."""
309	1		self.napp.int_manager.handle_pp_link_down = AsyncMock()
310	1		await self.napp.on_link_down(KytosEvent(content={"link": MagicMock()}))
311	1		assert self.napp.int_manager.handle_pp_link_down.call_count == 1
312
313	1		async def test_on_link_up(self) -> None:
314			"""Test on link_up."""
315	1		self.napp.int_manager.handle_pp_link_up = AsyncMock()
316	1		await self.napp.on_link_up(KytosEvent(content={"link": MagicMock()}))
317	1		assert self.napp.int_manager.handle_pp_link_up.call_count == 1
318
319	1		async def test_on_table_enabled_error(self, monkeypatch) -> None:
320			"""Test on_table_enabled error case."""
321	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
322	1		log_mock = MagicMock()
323	1		monkeypatch.setattr("napps.kytos.telemetry_int.main.log", log_mock)
324	1		await self.napp.on_table_enabled(
325			KytosEvent(content={"telemetry_int": {"invalid": 1}})
326			)
327	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
328	1		assert log_mock.error.call_count == 1
329	1		assert not self.napp.controller.buffers.app.aput.call_count
330
331	1		async def test_on_flow_mod_error(self, monkeypatch) -> None:
332			"""Test on_flow_mod_error."""
333	1		api_mock_main, api_mock_int, flow = AsyncMock(), AsyncMock(), MagicMock()
334	1		flow.cookie = 0xA800000000000001
335	1		monkeypatch.setattr(
336			"napps.kytos.telemetry_int.main.api",
337			api_mock_main,
338			)
339	1		monkeypatch.setattr(
340			"napps.kytos.telemetry_int.managers.int.api",
341			api_mock_int,
342			)
343	1		cookie = utils.get_id_from_cookie(flow.cookie)
344	1		api_mock_main.get_evc.return_value = {
345			cookie: {"metadata": {"telemetry": {"enabled": True}}}
346			}
347	1		api_mock_int.get_stored_flows.return_value = {cookie: [MagicMock()]}
348	1		self.napp.int_manager._remove_int_flows = AsyncMock()
349
350	1		event = KytosEvent(content={"flow": flow, "error_command": "add"})
351	1		await self.napp.on_flow_mod_error(event)
352
353	1		assert api_mock_main.get_evc.call_count == 1
354	1		assert api_mock_int.get_stored_flows.call_count == 1
355	1		assert api_mock_int.add_evcs_metadata.call_count == 1
356	1		assert self.napp.int_manager._remove_int_flows.call_count == 1
357
358	1		async def test_on_mef_eline_evcs_loaded(self):
359			"""Test on_mef_eline_evcs_loaded."""
360	1		evcs = {"1": {}, "2": {}}
361	1		event = KytosEvent(content=evcs)
362	1		self.napp.int_manager = MagicMock()
363	1		await self.napp.on_mef_eline_evcs_loaded(event)
364	1		self.napp.int_manager.load_uni_src_proxy_ports.assert_called_with(evcs)
365
366	1		async def test_on_intf_metadata_remove(self):
367			"""Test on_intf_metadata_removed."""
368	1		intf = MagicMock()
369	1		event = KytosEvent(content={"interface": intf})
370	1		self.napp.int_manager = MagicMock()
371	1		await self.napp.on_intf_metadata_removed(event)
372	1		self.napp.int_manager.handle_pp_metadata_removed.assert_called_with(intf)
373
374	1		async def test_on_intf_metadata_added(self):
375			"""Test on_intf_metadata_added."""
376	1		intf = MagicMock()
377	1		event = KytosEvent(content={"interface": intf})
378	1		self.napp.int_manager = MagicMock()
379	1		await self.napp.on_intf_metadata_added(event)
380			self.napp.int_manager.handle_pp_metadata_added.assert_called_with(intf)
381

kytos-ng / telemetry_int

Push — master ( fb285f...124bde )

test_main.TestMain.test_on_link_down() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like