test_utils.test_get_evc_unis() - Code Metrics - Inspection of "Merge pull request #104 from kytos-ng/feat/failove..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 4bf6f4...81fe6d )

by Vinicius

created 2024-06-14 16:49 UTC

test_utils.test_get_evc_unis() A

↳ Parent: test_utils

Complexity

Conditions

Size

Total Lines	20
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	15
nop	0
dl	0
loc	20
ccs	9
cts	9
cp	1
crap	1
rs	9.65
c	0
b	0
f	0

"""Test utils."""

import pytest
from httpx import Response
from unittest.mock import AsyncMock, MagicMock
from napps.kytos.telemetry_int import utils
from napps.kytos.telemetry_int.exceptions import FlowsNotFound, PriorityOverflow


@pytest.mark.parametrize(
    "flow,expected",
    [
        (
            {
                "flow": {
                    "priority": 100,
                    "match": {"in_port": 100},
                    "actions": [{"action_type": "output", "port": 1}],
                }
            },
            {
                "flow": {
                    "priority": 100,
                    "match": {"in_port": 100},
                    "instructions": [
                        {
                            "instruction_type": "apply_actions",
                            "actions": [{"action_type": "output", "port": 1}],
                        }
                    ],
                }
            },
        ),
        (
            {
                "flow": {
                    "priority": 100,
                    "match": {"in_port": 100},
                    "instructions": [
                        {
                            "instruction_type": "apply_actions",
                            "actions": [{"action_type": "output", "port": 1}],
                        }
                    ],
                }
            },
            {
                "flow": {
                    "priority": 100,
                    "match": {"in_port": 100},
                    "instructions": [
                        {
                            "instruction_type": "apply_actions",
                            "actions": [{"action_type": "output", "port": 1}],
                        }
                    ],
                }
            },
        ),
    ],
)
def test_instructions_from_actions(flow, expected) -> None:
    """Test instructions from actions."""
    assert utils.set_instructions_from_actions(flow) == expected


@pytest.mark.parametrize(
    "cookie,expected",
    [
        (0xAA3766C105686749, 0xA83766C105686749),
        (0xAACBEE9338673946, 0xA8CBEE9338673946),
    ],
)
def test_get_new_cookie(cookie, expected) -> None:
    """test get_new_cookie."""
    assert utils.get_new_cookie(cookie) == expected


@pytest.mark.parametrize(
    "cookie,expected_evc_id",
    [
        (0xAA3766C105686749, "3766c105686749"),
        (0xAACBEE9338673946, "cbee9338673946"),
    ],
)
def test_get_id_from_cookie(cookie, expected_evc_id) -> None:
    """test get_id_from_cookie."""
    assert utils.get_id_from_cookie(cookie) == expected_evc_id


def test_set_new_cookie() -> None:
    """Test set_new_cookie."""
    flow = {"flow": {"cookie": 0xAA3766C105686749}}
    utils.set_new_cookie(flow)
    assert flow["flow"]["cookie"] == 0xA83766C105686749


@pytest.mark.parametrize(
    "evc_dict,expected",
    [
        ({"metadata": {"telemetry": {"enabled": True}}}, True),
        ({"metadata": {"telemetry": {"enabled": False}}}, False),
        ({"metadata": {}}, False),
    ],
)
def test_has_int_enabled(evc_dict, expected) -> None:
    """test has_int_enabled."""
    assert utils.has_int_enabled(evc_dict) == expected


def test_get_evc_unis() -> None:
    """test get_evc_unis."""
    evc = {
        "uni_a": {
            "tag": {"tag_type": 1, "value": 200},
            "interface_id": "00:00:00:00:00:00:00:01:1",
        },
        "uni_z": {
            "tag": {"tag_type": 1, "value": 200},
            "interface_id": "00:00:00:00:00:00:00:01:2",
        },
    }
    uni_a, uni_z = utils.get_evc_unis(evc)
    assert uni_a["interface_id"] == evc["uni_a"]["interface_id"]
    assert uni_a["port_number"] == 1
    assert uni_a["switch"] == "00:00:00:00:00:00:00:01"

    assert uni_z["interface_id"] == evc["uni_z"]["interface_id"]
    assert uni_z["port_number"] == 2
    assert uni_z["switch"] == "00:00:00:00:00:00:00:01"


def test_is_intra_switch_evc() -> None:
    """test is_instra_switch_evc."""
    evc = {
        "uni_a": {
            "tag": {"tag_type": 1, "value": 200},
            "interface_id": "00:00:00:00:00:00:00:01:1",
        },
        "uni_z": {
            "tag": {"tag_type": 1, "value": 200},
            "interface_id": "00:00:00:00:00:00:00:01:2",
        },
    }
    assert utils.is_intra_switch_evc(evc)
    evc["uni_a"]["interface_id"] = "00:00:00:00:00:00:00:02:1"
    assert not utils.is_intra_switch_evc(evc)


def test_add_to_apply_actions() -> None:
    """Test add to apply actions."""
    instructions = [
        {
            "instruction_type": "apply_actions",
            "actions": [
                {"action_type": "set_vlan", "vlan_id": 200},
                {"action_type": "output", "port": 5},
            ],
        }
    ]
    assert instructions[0]["actions"][0] == {"action_type": "set_vlan", "vlan_id": 200}
    new_instruction = {"action_type": "add_int_metadata"}
    utils.add_to_apply_actions(instructions, new_instruction, position=0)
    assert instructions[0]["actions"][0] == new_instruction


def test_set_owner() -> None:
    """Test set_owner."""
    flow = {
        "flow": {
            "priority": 100,
            "match": {"in_port": 100},
            "actions": [{"action_type": "output", "port": 1}],
        }
    }
    utils.set_owner(flow)
    assert flow["flow"]["owner"] == "telemetry_int"


@pytest.mark.parametrize(
    "actions,actions_to_change,remove,expected_actions",
    [
        (
            [
                {"action_type": "set_queue", "queue_id": 1},
                {"action_type": "set_vlan", "vlan_id": 200},
                {"action_type": "output", "port": 5},
            ],
            ["set_vlan"],
            True,
            [
                {"action_type": "set_queue", "queue_id": 1},
                {"action_type": "output", "port": 5},
            ],
        ),
        (
            [
                {"action_type": "set_queue", "queue_id": 1},
                {"action_type": "set_vlan", "vlan_id": 200},
                {"action_type": "output", "port": 5},
            ],
            ["set_vlan"],
            False,
            [
                {"action_type": "set_vlan", "vlan_id": 200},
            ],
        ),
    ],
)
def test_modify_actions(actions, actions_to_change, remove, expected_actions) -> None:
    """test modify_actions."""
    assert utils.modify_actions(actions, actions_to_change, remove) == expected_actions


async def test_get_found_stored_flows(monkeypatch, intra_evc_evpl_flows_data) -> None:
    """test get_found_stored_flows."""
    evc_data = intra_evc_evpl_flows_data
    dpid = "00:00:00:00:00:00:00:01"
    cookies = [evc_data[dpid][0]["flow"]["cookie"]]
    # taking the opportunity to also cover the cookie tuple input filter
    cookies = [(c, c) for c in cookies]
    assert cookies

    aclient_mock, awith_mock = AsyncMock(), MagicMock()
    aclient_mock.request.return_value = Response(
        200, json=intra_evc_evpl_flows_data, request=MagicMock()
    )
    awith_mock.return_value.__aenter__.return_value = aclient_mock
    monkeypatch.setattr("httpx.AsyncClient", awith_mock)

    resp = await utils.get_found_stored_flows(cookies)
    assert resp
    for cookie, _cookie in cookies:
        assert cookie in resp


async def test_get_found_stored_flows_exc(monkeypatch) -> None:
    """test get_found_stored_flows exc."""
    mock = AsyncMock()
    mock.return_value = {1: []}
    monkeypatch.setattr("napps.kytos.telemetry_int.utils._get_stored_flows", mock)
    with pytest.raises(FlowsNotFound):
        await utils.get_found_stored_flows()


@pytest.mark.parametrize(
    "flow,expected_prio",
    [
        ({"flow": {"priority": 1}}, 101),
        ({"flow": {"priority": 2**16 - 50}}, 2**16 - 49),
    ],
)
def test_set_priority(flow, expected_prio) -> None:
    """test set priority."""
    resp = utils.set_priority(flow, "some_id")
    assert resp["flow"]["priority"] == expected_prio


def test_set_priority_exc() -> None:
    """test set priority exc."""
    flow = {"flow": {"priority": 2**16}}
    with pytest.raises(PriorityOverflow):
        utils.set_priority(flow, "some_id")


@pytest.mark.parametrize(
    "link,dpid,expected_vlan",
    [
        (
            {
                "endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
                "endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
                "metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
            },
            "00:00:00:00:00:00:00:01",
            2,
        ),
        (
            {
                "endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
                "endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
                "metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
            },
            "00:00:00:00:00:00:00:02",
            2,
        ),
        (
            {
                "endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
                "endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
                "metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
            },
            "00:00:00:00:00:00:00:03",
            None,
        ),
        (
            {
                "endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
                "endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
                "metadata": {},
            },
            "00:00:00:00:00:00:00:02",
            None,
        ),
    ],
)
def test_get_svlan_dpid_link(link, dpid, expected_vlan) -> None:
    """Test get_svlan_dpid_link."""
    assert utils.get_svlan_dpid_link(link, dpid) == expected_vlan


1		"""Test utils."""
2
3	1	import pytest
4	1	from httpx import Response
5	1	from unittest.mock import AsyncMock, MagicMock
6	1	from napps.kytos.telemetry_int import utils
7	1	from napps.kytos.telemetry_int.exceptions import FlowsNotFound, PriorityOverflow
8
9
10	1	@pytest.mark.parametrize(
11		"flow,expected",
12		[
13		(
14		{
15		"flow": {
16		"priority": 100,
17		"match": {"in_port": 100},
18		"actions": [{"action_type": "output", "port": 1}],
19		}
20		},
21		{
22		"flow": {
23		"priority": 100,
24		"match": {"in_port": 100},
25		"instructions": [
26		{
27		"instruction_type": "apply_actions",
28		"actions": [{"action_type": "output", "port": 1}],
29		}
30		],
31		}
32		},
33		),
34		(
35		{
36		"flow": {
37		"priority": 100,
38		"match": {"in_port": 100},
39		"instructions": [
40		{
41		"instruction_type": "apply_actions",
42		"actions": [{"action_type": "output", "port": 1}],
43		}
44		],
45		}
46		},
47		{
48		"flow": {
49		"priority": 100,
50		"match": {"in_port": 100},
51		"instructions": [
52		{
53		"instruction_type": "apply_actions",
54		"actions": [{"action_type": "output", "port": 1}],
55		}
56		],
57		}
58		},
59		),
60		],
61		)
62	1	def test_instructions_from_actions(flow, expected) -> None:
63		"""Test instructions from actions."""
64	1	assert utils.set_instructions_from_actions(flow) == expected
65
66
67	1	@pytest.mark.parametrize(
68		"cookie,expected",
69		[
70		(0xAA3766C105686749, 0xA83766C105686749),
71		(0xAACBEE9338673946, 0xA8CBEE9338673946),
72		],
73		)
74	1	def test_get_new_cookie(cookie, expected) -> None:
75		"""test get_new_cookie."""
76	1	assert utils.get_new_cookie(cookie) == expected
77
78
79	1	@pytest.mark.parametrize(
80		"cookie,expected_evc_id",
81		[
82		(0xAA3766C105686749, "3766c105686749"),
83		(0xAACBEE9338673946, "cbee9338673946"),
84		],
85		)
86	1	def test_get_id_from_cookie(cookie, expected_evc_id) -> None:
87		"""test get_id_from_cookie."""
88	1	assert utils.get_id_from_cookie(cookie) == expected_evc_id
89
90
91	1	def test_set_new_cookie() -> None:
92		"""Test set_new_cookie."""
93	1	flow = {"flow": {"cookie": 0xAA3766C105686749}}
94	1	utils.set_new_cookie(flow)
95	1	assert flow["flow"]["cookie"] == 0xA83766C105686749
96
97
98	1	@pytest.mark.parametrize(
99		"evc_dict,expected",
100		[
101		({"metadata": {"telemetry": {"enabled": True}}}, True),
102		({"metadata": {"telemetry": {"enabled": False}}}, False),
103		({"metadata": {}}, False),
104		],
105		)
106	1	def test_has_int_enabled(evc_dict, expected) -> None:
107		"""test has_int_enabled."""
108	1	assert utils.has_int_enabled(evc_dict) == expected
109
110
111	1	def test_get_evc_unis() -> None:
112		"""test get_evc_unis."""
113	1	evc = {
114		"uni_a": {
115		"tag": {"tag_type": 1, "value": 200},
116		"interface_id": "00:00:00:00:00:00:00:01:1",
117		},
118		"uni_z": {
119		"tag": {"tag_type": 1, "value": 200},
120		"interface_id": "00:00:00:00:00:00:00:01:2",
121		},
122		}
123	1	uni_a, uni_z = utils.get_evc_unis(evc)
124	1	assert uni_a["interface_id"] == evc["uni_a"]["interface_id"]
125	1	assert uni_a["port_number"] == 1
126	1	assert uni_a["switch"] == "00:00:00:00:00:00:00:01"
127
128	1	assert uni_z["interface_id"] == evc["uni_z"]["interface_id"]
129	1	assert uni_z["port_number"] == 2
130	1	assert uni_z["switch"] == "00:00:00:00:00:00:00:01"
131
132
133	1	def test_is_intra_switch_evc() -> None:
134		"""test is_instra_switch_evc."""
135	1	evc = {
136		"uni_a": {
137		"tag": {"tag_type": 1, "value": 200},
138		"interface_id": "00:00:00:00:00:00:00:01:1",
139		},
140		"uni_z": {
141		"tag": {"tag_type": 1, "value": 200},
142		"interface_id": "00:00:00:00:00:00:00:01:2",
143		},
144		}
145	1	assert utils.is_intra_switch_evc(evc)
146	1	evc["uni_a"]["interface_id"] = "00:00:00:00:00:00:00:02:1"
147	1	assert not utils.is_intra_switch_evc(evc)
148
149
150	1	def test_add_to_apply_actions() -> None:
151		"""Test add to apply actions."""
152	1	instructions = [
153		{
154		"instruction_type": "apply_actions",
155		"actions": [
156		{"action_type": "set_vlan", "vlan_id": 200},
157		{"action_type": "output", "port": 5},
158		],
159		}
160		]
161	1	assert instructions[0]["actions"][0] == {"action_type": "set_vlan", "vlan_id": 200}
162	1	new_instruction = {"action_type": "add_int_metadata"}
163	1	utils.add_to_apply_actions(instructions, new_instruction, position=0)
164	1	assert instructions[0]["actions"][0] == new_instruction
165
166
167	1	def test_set_owner() -> None:
168		"""Test set_owner."""
169	1	flow = {
170		"flow": {
171		"priority": 100,
172		"match": {"in_port": 100},
173		"actions": [{"action_type": "output", "port": 1}],
174		}
175		}
176	1	utils.set_owner(flow)
177	1	assert flow["flow"]["owner"] == "telemetry_int"
178
179
180	1	@pytest.mark.parametrize(
181		"actions,actions_to_change,remove,expected_actions",
182		[
183		(
184		[
185		{"action_type": "set_queue", "queue_id": 1},
186		{"action_type": "set_vlan", "vlan_id": 200},
187		{"action_type": "output", "port": 5},
188		],
189		["set_vlan"],
190		True,
191		[
192		{"action_type": "set_queue", "queue_id": 1},
193		{"action_type": "output", "port": 5},
194		],
195		),
196		(
197		[
198		{"action_type": "set_queue", "queue_id": 1},
199		{"action_type": "set_vlan", "vlan_id": 200},
200		{"action_type": "output", "port": 5},
201		],
202		["set_vlan"],
203		False,
204		[
205		{"action_type": "set_vlan", "vlan_id": 200},
206		],
207		),
208		],
209		)
210	1	def test_modify_actions(actions, actions_to_change, remove, expected_actions) -> None:
211		"""test modify_actions."""
212	1	assert utils.modify_actions(actions, actions_to_change, remove) == expected_actions
213
214
215	1	async def test_get_found_stored_flows(monkeypatch, intra_evc_evpl_flows_data) -> None:
216		"""test get_found_stored_flows."""
217	1	evc_data = intra_evc_evpl_flows_data
218	1	dpid = "00:00:00:00:00:00:00:01"
219	1	cookies = [evc_data[dpid][0]["flow"]["cookie"]]
220		# taking the opportunity to also cover the cookie tuple input filter
221	1	cookies = [(c, c) for c in cookies]
222	1	assert cookies
223
224	1	aclient_mock, awith_mock = AsyncMock(), MagicMock()
225	1	aclient_mock.request.return_value = Response(
226		200, json=intra_evc_evpl_flows_data, request=MagicMock()
227		)
228	1	awith_mock.return_value.__aenter__.return_value = aclient_mock
229	1	monkeypatch.setattr("httpx.AsyncClient", awith_mock)
230
231	1	resp = await utils.get_found_stored_flows(cookies)
232	1	assert resp
233	1	for cookie, _cookie in cookies:
234	1	assert cookie in resp
235
236
237	1	async def test_get_found_stored_flows_exc(monkeypatch) -> None:
238		"""test get_found_stored_flows exc."""
239	1	mock = AsyncMock()
240	1	mock.return_value = {1: []}
241	1	monkeypatch.setattr("napps.kytos.telemetry_int.utils._get_stored_flows", mock)
242	1	with pytest.raises(FlowsNotFound):
243	1	await utils.get_found_stored_flows()
244
245
246	1	@pytest.mark.parametrize(
247		"flow,expected_prio",
248		[
249		({"flow": {"priority": 1}}, 101),
250		({"flow": {"priority": 216 - 50}}, 216 - 49),
251		],
252		)
253	1	def test_set_priority(flow, expected_prio) -> None:
254		"""test set priority."""
255	1	resp = utils.set_priority(flow, "some_id")
256	1	assert resp["flow"]["priority"] == expected_prio
257
258
259	1	def test_set_priority_exc() -> None:
260		"""test set priority exc."""
261	1	flow = {"flow": {"priority": 2**16}}
262	1	with pytest.raises(PriorityOverflow):
263	1	utils.set_priority(flow, "some_id")
264
265
266	1	@pytest.mark.parametrize(
267		"link,dpid,expected_vlan",
268		[
269		(
270		{
271		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
272		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
273		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
274		},
275		"00:00:00:00:00:00:00:01",
276		2,
277		),
278		(
279		{
280		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
281		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
282		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
283		},
284		"00:00:00:00:00:00:00:02",
285		2,
286		),
287		(
288		{
289		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
290		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
291		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
292		},
293		"00:00:00:00:00:00:00:03",
294		None,
295		),
296		(
297		{
298		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
299		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
300		"metadata": {},
301		},
302		"00:00:00:00:00:00:00:02",
303		None,
304		),
305		],
306		)
307	1	def test_get_svlan_dpid_link(link, dpid, expected_vlan) -> None:
308		"""Test get_svlan_dpid_link."""
309		assert utils.get_svlan_dpid_link(link, dpid) == expected_vlan
310

kytos-ng / telemetry_int

Push — master ( 4bf6f4...81fe6d )

test_utils.test_get_evc_unis() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like