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

Passed

Push — master ( 124bde...9bde40 )

by Vinicius

created 2024-05-08 13:41 UTC

test_main.TestMain.test_on_intf_metadata_remove() A

↳ Parent: test_main

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	1
dl	0
loc	7
ccs	6
cts	6
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_deployed(self) -> None:
        """Test on_evc_deployed."""
        content = {"metadata": {"telemetry_request": {}}, "evc_id": "some_id"}
        self.napp.int_manager.redeploy_int = AsyncMock()
        self.napp.int_manager.enable_int = AsyncMock()
        await self.napp.on_evc_deployed(KytosEvent(content=content))
        assert self.napp.int_manager.enable_int.call_count == 1
        assert self.napp.int_manager.redeploy_int.call_count == 0

        content = {"metadata": {"telemetry": {"enabled": True}}, "evc_id": "some_id"}
        await self.napp.on_evc_deployed(KytosEvent(content=content))
        assert self.napp.int_manager.enable_int.call_count == 1
        assert self.napp.int_manager.redeploy_int.call_count == 1

    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_deployed(self) -> None:
233			"""Test on_evc_deployed."""
234	1		content = {"metadata": {"telemetry_request": {}}, "evc_id": "some_id"}
235	1		self.napp.int_manager.redeploy_int = AsyncMock()
236	1		self.napp.int_manager.enable_int = AsyncMock()
237	1		await self.napp.on_evc_deployed(KytosEvent(content=content))
238	1		assert self.napp.int_manager.enable_int.call_count == 1
239	1		assert self.napp.int_manager.redeploy_int.call_count == 0
240
241	1		content = {"metadata": {"telemetry": {"enabled": True}}, "evc_id": "some_id"}
242	1		await self.napp.on_evc_deployed(KytosEvent(content=content))
243	1		assert self.napp.int_manager.enable_int.call_count == 1
244	1		assert self.napp.int_manager.redeploy_int.call_count == 1
245
246	1		async def test_on_evc_deleted(self) -> None:
247			"""Test on_evc_deleted."""
248	1		content = {"metadata": {"telemetry": {"enabled": True}}, "evc_id": "some_id"}
249	1		self.napp.int_manager.disable_int = AsyncMock()
250	1		await self.napp.on_evc_deleted(KytosEvent(content=content))
251	1		assert self.napp.int_manager.disable_int.call_count == 1
252
253	1		async def test_on_uni_active_updated(self, monkeypatch) -> None:
254			"""Test on UNI active updated."""
255	1		api_mock = AsyncMock()
256	1		monkeypatch.setattr(
257			"napps.kytos.telemetry_int.main.api",
258			api_mock,
259			)
260	1		content = {
261			"metadata": {"telemetry": {"enabled": True}},
262			"evc_id": "some_id",
263			"active": True,
264			}
265	1		await self.napp.on_uni_active_updated(KytosEvent(content=content))
266	1		assert api_mock.add_evcs_metadata.call_count == 1
267	1		args = api_mock.add_evcs_metadata.call_args[0][1]
268	1		assert args["telemetry"]["status"] == "UP"
269
270	1		content["active"] = False
271	1		await self.napp.on_uni_active_updated(KytosEvent(content=content))
272	1		assert api_mock.add_evcs_metadata.call_count == 2
273	1		args = api_mock.add_evcs_metadata.call_args[0][1]
274	1		assert args["telemetry"]["status"] == "DOWN"
275
276	1		async def test_on_evc_undeployed(self) -> None:
277			"""Test on_evc_undeployed."""
278	1		content = {
279			"enabled": False,
280			"metadata": {"telemetry": {"enabled": False}},
281			"evc_id": "some_id",
282			}
283	1		self.napp.int_manager.remove_int_flows = AsyncMock()
284	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
285	1		assert self.napp.int_manager.remove_int_flows.call_count == 0
286
287	1		content["metadata"]["telemetry"]["enabled"] = True
288	1		await self.napp.on_evc_undeployed(KytosEvent(content=content))
289	1		assert self.napp.int_manager.remove_int_flows.call_count == 1
290
291	1		async def test_on_evc_redeployed_link(self) -> None:
292			"""Test on redeployed_link_down\|redeployed_link_up."""
293	1		content = {
294			"enabled": True,
295			"metadata": {"telemetry": {"enabled": False}},
296			"evc_id": "some_id",
297			}
298	1		self.napp.int_manager.redeploy_int = AsyncMock()
299	1		await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
300	1		assert self.napp.int_manager.redeploy_int.call_count == 0
301
302	1		content["metadata"]["telemetry"]["enabled"] = True
303	1		await self.napp.on_evc_redeployed_link(KytosEvent(content=content))
304	1		assert self.napp.int_manager.redeploy_int.call_count == 1
305
306	1		async def test_on_evc_error_redeployed_link_down(self) -> None:
307			"""Test error_redeployed_link_down."""
308	1		content = {
309			"enabled": True,
310			"metadata": {"telemetry": {"enabled": False}},
311			"evc_id": "some_id",
312			}
313	1		self.napp.int_manager.remove_int_flows = AsyncMock()
314	1		await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
315	1		assert self.napp.int_manager.remove_int_flows.call_count == 0
316
317	1		content["metadata"]["telemetry"]["enabled"] = True
318	1		await self.napp.on_evc_error_redeployed_link_down(KytosEvent(content=content))
319	1		assert self.napp.int_manager.remove_int_flows.call_count == 1
320
321	1		async def test_on_link_down(self) -> None:
322			"""Test on link_down."""
323	1		self.napp.int_manager.handle_pp_link_down = AsyncMock()
324	1		await self.napp.on_link_down(KytosEvent(content={"link": MagicMock()}))
325	1		assert self.napp.int_manager.handle_pp_link_down.call_count == 1
326
327	1		async def test_on_link_up(self) -> None:
328			"""Test on link_up."""
329	1		self.napp.int_manager.handle_pp_link_up = AsyncMock()
330	1		await self.napp.on_link_up(KytosEvent(content={"link": MagicMock()}))
331	1		assert self.napp.int_manager.handle_pp_link_up.call_count == 1
332
333	1		async def test_on_table_enabled_error(self, monkeypatch) -> None:
334			"""Test on_table_enabled error case."""
335	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
336	1		log_mock = MagicMock()
337	1		monkeypatch.setattr("napps.kytos.telemetry_int.main.log", log_mock)
338	1		await self.napp.on_table_enabled(
339			KytosEvent(content={"telemetry_int": {"invalid": 1}})
340			)
341	1		assert self.napp.int_manager.flow_builder.table_group == {"evpl": 2, "epl": 3}
342	1		assert log_mock.error.call_count == 1
343	1		assert not self.napp.controller.buffers.app.aput.call_count
344
345	1		async def test_on_flow_mod_error(self, monkeypatch) -> None:
346			"""Test on_flow_mod_error."""
347	1		api_mock_main, api_mock_int, flow = AsyncMock(), AsyncMock(), MagicMock()
348	1		flow.cookie = 0xA800000000000001
349	1		monkeypatch.setattr(
350			"napps.kytos.telemetry_int.main.api",
351			api_mock_main,
352			)
353	1		monkeypatch.setattr(
354			"napps.kytos.telemetry_int.managers.int.api",
355			api_mock_int,
356			)
357	1		cookie = utils.get_id_from_cookie(flow.cookie)
358	1		api_mock_main.get_evc.return_value = {
359			cookie: {"metadata": {"telemetry": {"enabled": True}}}
360			}
361	1		api_mock_int.get_stored_flows.return_value = {cookie: [MagicMock()]}
362	1		self.napp.int_manager._remove_int_flows = AsyncMock()
363
364	1		event = KytosEvent(content={"flow": flow, "error_command": "add"})
365	1		await self.napp.on_flow_mod_error(event)
366
367	1		assert api_mock_main.get_evc.call_count == 1
368	1		assert api_mock_int.get_stored_flows.call_count == 1
369	1		assert api_mock_int.add_evcs_metadata.call_count == 1
370	1		assert self.napp.int_manager._remove_int_flows.call_count == 1
371
372	1		async def test_on_mef_eline_evcs_loaded(self):
373			"""Test on_mef_eline_evcs_loaded."""
374	1		evcs = {"1": {}, "2": {}}
375	1		event = KytosEvent(content=evcs)
376	1		self.napp.int_manager = MagicMock()
377	1		await self.napp.on_mef_eline_evcs_loaded(event)
378	1		self.napp.int_manager.load_uni_src_proxy_ports.assert_called_with(evcs)
379
380	1		async def test_on_intf_metadata_remove(self):
381			"""Test on_intf_metadata_removed."""
382	1		intf = MagicMock()
383	1		event = KytosEvent(content={"interface": intf})
384	1		self.napp.int_manager = MagicMock()
385	1		await self.napp.on_intf_metadata_removed(event)
386	1		self.napp.int_manager.handle_pp_metadata_removed.assert_called_with(intf)
387
388	1		async def test_on_intf_metadata_added(self):
389			"""Test on_intf_metadata_added."""
390	1		intf = MagicMock()
391	1		event = KytosEvent(content={"interface": intf})
392	1		self.napp.int_manager = MagicMock()
393	1		await self.napp.on_intf_metadata_added(event)
394			self.napp.int_manager.handle_pp_metadata_added.assert_called_with(intf)
395

kytos-ng / telemetry_int

Push — master ( 124bde...9bde40 )

test_main.TestMain.test_on_intf_metadata_remove() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like