test_utils.test_instructions_from_actions() - Code Metrics - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

test_utils.test_instructions_from_actions() A
last analyzed 2025-06-27 17:39 UTC

↳ Parent: test_utils

Complexity

Conditions

Size

Total Lines	55
Code Lines	29

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	29
nop	2
dl	0
loc	55
ccs	3
cts	3
cp	1
crap	1
rs	9.184
c	0
b	0
f	0

How to fix Long Method

"""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"),
        (0xAA0F756E60D34E4B, "0f756e60d34e4b"),
        (0xAA00756E60D34E4B, "00756e60d34e4b"),
    ],
)
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


@pytest.mark.parametrize(
    "instructions,instruction_type,action_type,expected",
    [
        (
            [
                {
                    "instruction_type": "apply_actions",
                    "actions": [{"action_type": "push_int"}],
                },
                {"instruction_type": "goto_table", "table_id": 2},
            ],
            "apply_actions",
            "push_int",
            True,
        ),
        (
            [
                {"instruction_type": "goto_table", "table_id": 2},
            ],
            "apply_actions",
            "push_int",
            False,
        ),
        (
            [
                {
                    "instruction_type": "apply_actions",
                    "actions": [{"action_type": "push_int"}],
                },
                {"instruction_type": "goto_table", "table_id": 2},
            ],
            "apply_actions",
            "pop_int",
            False,
        ),
    ],
)
def test_has_instruction_and_action_type(
    instructions, instruction_type, action_type, expected
) -> None:
    """Test add to apply actions."""
    assert (
        utils.has_instruction_and_action_type(
            instructions, instruction_type, action_type
        )
        == expected
    )


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		(0xAA0F756E60D34E4B, "0f756e60d34e4b"),
85		(0xAA00756E60D34E4B, "00756e60d34e4b"),
86		],
87		)
88	1	def test_get_id_from_cookie(cookie, expected_evc_id) -> None:
89		"""test get_id_from_cookie."""
90	1	assert utils.get_id_from_cookie(cookie) == expected_evc_id
91
92
93	1	def test_set_new_cookie() -> None:
94		"""Test set_new_cookie."""
95	1	flow = {"flow": {"cookie": 0xAA3766C105686749}}
96	1	utils.set_new_cookie(flow)
97	1	assert flow["flow"]["cookie"] == 0xA83766C105686749
98
99
100	1	@pytest.mark.parametrize(
101		"evc_dict,expected",
102		[
103		({"metadata": {"telemetry": {"enabled": True}}}, True),
104		({"metadata": {"telemetry": {"enabled": False}}}, False),
105		({"metadata": {}}, False),
106		],
107		)
108	1	def test_has_int_enabled(evc_dict, expected) -> None:
109		"""test has_int_enabled."""
110	1	assert utils.has_int_enabled(evc_dict) == expected
111
112
113	1	def test_get_evc_unis() -> None:
114		"""test get_evc_unis."""
115	1	evc = {
116		"uni_a": {
117		"tag": {"tag_type": 1, "value": 200},
118		"interface_id": "00:00:00:00:00:00:00:01:1",
119		},
120		"uni_z": {
121		"tag": {"tag_type": 1, "value": 200},
122		"interface_id": "00:00:00:00:00:00:00:01:2",
123		},
124		}
125	1	uni_a, uni_z = utils.get_evc_unis(evc)
126	1	assert uni_a["interface_id"] == evc["uni_a"]["interface_id"]
127	1	assert uni_a["port_number"] == 1
128	1	assert uni_a["switch"] == "00:00:00:00:00:00:00:01"
129
130	1	assert uni_z["interface_id"] == evc["uni_z"]["interface_id"]
131	1	assert uni_z["port_number"] == 2
132	1	assert uni_z["switch"] == "00:00:00:00:00:00:00:01"
133
134
135	1	def test_is_intra_switch_evc() -> None:
136		"""test is_instra_switch_evc."""
137	1	evc = {
138		"uni_a": {
139		"tag": {"tag_type": 1, "value": 200},
140		"interface_id": "00:00:00:00:00:00:00:01:1",
141		},
142		"uni_z": {
143		"tag": {"tag_type": 1, "value": 200},
144		"interface_id": "00:00:00:00:00:00:00:01:2",
145		},
146		}
147	1	assert utils.is_intra_switch_evc(evc)
148	1	evc["uni_a"]["interface_id"] = "00:00:00:00:00:00:00:02:1"
149	1	assert not utils.is_intra_switch_evc(evc)
150
151
152	1	def test_add_to_apply_actions() -> None:
153		"""Test add to apply actions."""
154	1	instructions = [
155		{
156		"instruction_type": "apply_actions",
157		"actions": [
158		{"action_type": "set_vlan", "vlan_id": 200},
159		{"action_type": "output", "port": 5},
160		],
161		}
162		]
163	1	assert instructions[0]["actions"][0] == {"action_type": "set_vlan", "vlan_id": 200}
164	1	new_instruction = {"action_type": "add_int_metadata"}
165	1	utils.add_to_apply_actions(instructions, new_instruction, position=0)
166	1	assert instructions[0]["actions"][0] == new_instruction
167
168
169	1	@pytest.mark.parametrize(
170		"instructions,instruction_type,action_type,expected",
171		[
172		(
173		[
174		{
175		"instruction_type": "apply_actions",
176		"actions": [{"action_type": "push_int"}],
177		},
178		{"instruction_type": "goto_table", "table_id": 2},
179		],
180		"apply_actions",
181		"push_int",
182		True,
183		),
184		(
185		[
186		{"instruction_type": "goto_table", "table_id": 2},
187		],
188		"apply_actions",
189		"push_int",
190		False,
191		),
192		(
193		[
194		{
195		"instruction_type": "apply_actions",
196		"actions": [{"action_type": "push_int"}],
197		},
198		{"instruction_type": "goto_table", "table_id": 2},
199		],
200		"apply_actions",
201		"pop_int",
202		False,
203		),
204		],
205		)
206	1	def test_has_instruction_and_action_type(
207		instructions, instruction_type, action_type, expected
208		) -> None:
209		"""Test add to apply actions."""
210	1	assert (
211		utils.has_instruction_and_action_type(
212		instructions, instruction_type, action_type
213		)
214		== expected
215		)
216
217
218	1	def test_set_owner() -> None:
219		"""Test set_owner."""
220	1	flow = {
221		"flow": {
222		"priority": 100,
223		"match": {"in_port": 100},
224		"actions": [{"action_type": "output", "port": 1}],
225		}
226		}
227	1	utils.set_owner(flow)
228	1	assert flow["flow"]["owner"] == "telemetry_int"
229
230
231	1	@pytest.mark.parametrize(
232		"actions,actions_to_change,remove,expected_actions",
233		[
234		(
235		[
236		{"action_type": "set_queue", "queue_id": 1},
237		{"action_type": "set_vlan", "vlan_id": 200},
238		{"action_type": "output", "port": 5},
239		],
240		["set_vlan"],
241		True,
242		[
243		{"action_type": "set_queue", "queue_id": 1},
244		{"action_type": "output", "port": 5},
245		],
246		),
247		(
248		[
249		{"action_type": "set_queue", "queue_id": 1},
250		{"action_type": "set_vlan", "vlan_id": 200},
251		{"action_type": "output", "port": 5},
252		],
253		["set_vlan"],
254		False,
255		[
256		{"action_type": "set_vlan", "vlan_id": 200},
257		],
258		),
259		],
260		)
261	1	def test_modify_actions(actions, actions_to_change, remove, expected_actions) -> None:
262		"""test modify_actions."""
263	1	assert utils.modify_actions(actions, actions_to_change, remove) == expected_actions
264
265
266	1	async def test_get_found_stored_flows(monkeypatch, intra_evc_evpl_flows_data) -> None:
267		"""test get_found_stored_flows."""
268	1	evc_data = intra_evc_evpl_flows_data
269	1	dpid = "00:00:00:00:00:00:00:01"
270	1	cookies = [evc_data[dpid][0]["flow"]["cookie"]]
271		# taking the opportunity to also cover the cookie tuple input filter
272	1	cookies = [(c, c) for c in cookies]
273	1	assert cookies
274
275	1	aclient_mock, awith_mock = AsyncMock(), MagicMock()
276	1	aclient_mock.request.return_value = Response(
277		200, json=intra_evc_evpl_flows_data, request=MagicMock()
278		)
279	1	awith_mock.return_value.__aenter__.return_value = aclient_mock
280	1	monkeypatch.setattr("httpx.AsyncClient", awith_mock)
281
282	1	resp = await utils.get_found_stored_flows(cookies)
283	1	assert resp
284	1	for cookie, _cookie in cookies:
285	1	assert cookie in resp
286
287
288	1	async def test_get_found_stored_flows_exc(monkeypatch) -> None:
289		"""test get_found_stored_flows exc."""
290	1	mock = AsyncMock()
291	1	mock.return_value = {1: []}
292	1	monkeypatch.setattr("napps.kytos.telemetry_int.utils._get_stored_flows", mock)
293	1	with pytest.raises(FlowsNotFound):
294	1	await utils.get_found_stored_flows()
295
296
297	1	@pytest.mark.parametrize(
298		"flow,expected_prio",
299		[
300		({"flow": {"priority": 1}}, 101),
301		({"flow": {"priority": 216 - 50}}, 216 - 49),
302		],
303		)
304	1	def test_set_priority(flow, expected_prio) -> None:
305		"""test set priority."""
306	1	resp = utils.set_priority(flow, "some_id")
307	1	assert resp["flow"]["priority"] == expected_prio
308
309
310	1	def test_set_priority_exc() -> None:
311		"""test set priority exc."""
312	1	flow = {"flow": {"priority": 2**16}}
313	1	with pytest.raises(PriorityOverflow):
314	1	utils.set_priority(flow, "some_id")
315
316
317	1	@pytest.mark.parametrize(
318		"link,dpid,expected_vlan",
319		[
320		(
321		{
322		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
323		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
324		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
325		},
326		"00:00:00:00:00:00:00:01",
327		2,
328		),
329		(
330		{
331		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
332		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
333		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
334		},
335		"00:00:00:00:00:00:00:02",
336		2,
337		),
338		(
339		{
340		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
341		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
342		"metadata": {"s_vlan": {"tag_type": "vlan", "value": 2}},
343		},
344		"00:00:00:00:00:00:00:03",
345		None,
346		),
347		(
348		{
349		"endpoint_a": {"switch": "00:00:00:00:00:00:00:01"},
350		"endpoint_b": {"switch": "00:00:00:00:00:00:00:02"},
351		"metadata": {},
352		},
353		"00:00:00:00:00:00:00:02",
354		None,
355		),
356		],
357		)
358	1	def test_get_svlan_dpid_link(link, dpid, expected_vlan) -> None:
359		"""Test get_svlan_dpid_link."""
360		assert utils.get_svlan_dpid_link(link, dpid) == expected_vlan
361

kytos-ng / telemetry_int

test_utils.test_instructions_from_actions() A last analyzed 2025-06-27 17:39 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

test_utils.test_instructions_from_actions() A
last analyzed 2025-06-27 17:39 UTC