test_main.TestMain.test_get_evc_compare() - Code Metrics - Inspection of "feat: handled kytos/mef_eline.undeployed event" - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#85)

by Vinicius

created 2024-04-16 15:59 UTC

test_main.TestMain.test_get_evc_compare() A

↳ Parent: test_main

Complexity

Conditions

Size

Total Lines	28
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	21
nop	2
dl	0
loc	28
ccs	13
cts	13
cp	1
crap	1
rs	9.376
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_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_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_evc_undeployed(self) -> None:
240			"""Test on_evc_undeployed."""
241	1		content = {
242			"enabled": False,
243			"metadata": {"telemetry": {"enabled": False}},
244			"evc_id": "some_id",
245			}
246	1		self.napp.int_manager.remove_int_flows = AsyncMock()
247	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
248	1		assert self.napp.int_manager.remove_int_flows.call_count == 0
249
250	1		content["metadata"]["telemetry"]["enabled"] = True
251	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
252	1		assert self.napp.int_manager.remove_int_flows.call_count == 1
253
254	1		async def test_on_link_down(self) -> None:
255			"""Test on link_down."""
256	1		self.napp.int_manager.handle_pp_link_down = AsyncMock()
257	1		await self.napp.on_link_down(KytosEvent(content={"link": MagicMock()}))
258	1		assert self.napp.int_manager.handle_pp_link_down.call_count == 1
259
260	1		async def test_on_link_up(self) -> None:
261			"""Test on link_up."""
262	1		self.napp.int_manager.handle_pp_link_up = AsyncMock()
263	1		await self.napp.on_link_up(KytosEvent(content={"link": MagicMock()}))
264	1		assert self.napp.int_manager.handle_pp_link_up.call_count == 1
265
266	1		async def test_on_table_enabled_error(self, monkeypatch) -> None:
267			"""Test on_table_enabled error case."""
268	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
269	1		log_mock = MagicMock()
270	1		monkeypatch.setattr("napps.kytos.telemetry_int.main.log", log_mock)
271	1		await self.napp.on_table_enabled(
272			KytosEvent(content={"telemetry_int": {"invalid": 1}})
273			)
274	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
275	1		assert log_mock.error.call_count == 1
276	1		assert not self.napp.controller.buffers.app.aput.call_count
277
278	1		async def test_on_flow_mod_error(self, monkeypatch) -> None:
279			"""Test on_flow_mod_error."""
280	1		api_mock_main, api_mock_int, flow = AsyncMock(), AsyncMock(), MagicMock()
281	1		flow.cookie = 0xA800000000000001
282	1		monkeypatch.setattr(
283			"napps.kytos.telemetry_int.main.api",
284			api_mock_main,
285			)
286	1		monkeypatch.setattr(
287			"napps.kytos.telemetry_int.managers.int.api",
288			api_mock_int,
289			)
290	1		cookie = utils.get_id_from_cookie(flow.cookie)
291	1		api_mock_main.get_evc.return_value = {
292			cookie: {"metadata": {"telemetry": {"enabled": True}}}
293			}
294	1		api_mock_int.get_stored_flows.return_value = {cookie: [MagicMock()]}
295	1		self.napp.int_manager._remove_int_flows = AsyncMock()
296
297	1		event = KytosEvent(content={"flow": flow, "error_command": "add"})
298	1		await self.napp.on_flow_mod_error(event)
299
300	1		assert api_mock_main.get_evc.call_count == 1
301	1		assert api_mock_int.get_stored_flows.call_count == 1
302	1		assert api_mock_int.add_evcs_metadata.call_count == 1
303	1		assert self.napp.int_manager._remove_int_flows.call_count == 1
304
305	1		async def test_on_mef_eline_evcs_loaded(self):
306			"""Test on_mef_eline_evcs_loaded."""
307	1		evcs = {"1": {}, "2": {}}
308	1		event = KytosEvent(content=evcs)
309	1		self.napp.int_manager = MagicMock()
310	1		await self.napp.on_mef_eline_evcs_loaded(event)
311	1		self.napp.int_manager.load_uni_src_proxy_ports.assert_called_with(evcs)
312
313	1		async def test_on_intf_metadata_remove(self):
314			"""Test on_intf_metadata_removed."""
315	1		intf = MagicMock()
316	1		event = KytosEvent(content={"interface": intf})
317	1		self.napp.int_manager = MagicMock()
318	1		await self.napp.on_intf_metadata_removed(event)
319	1		self.napp.int_manager.handle_pp_metadata_removed.assert_called_with(intf)
320
321	1		async def test_on_intf_metadata_added(self):
322			"""Test on_intf_metadata_added."""
323	1		intf = MagicMock()
324	1		event = KytosEvent(content={"interface": intf})
325	1		self.napp.int_manager = MagicMock()
326	1		await self.napp.on_intf_metadata_added(event)
327			self.napp.int_manager.handle_pp_metadata_added.assert_called_with(intf)
328

kytos-ng / telemetry_int

Pull Request — master (#85)

test_main.TestMain.test_get_evc_compare() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like