test_utils.test_get_id_from_cookie() - Code Metrics - Inspection of "Merge pull request #79 from kytos-ng/chore/code_co..." - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 92444a...eafb59 )

by Vinicius

created 2024-01-12 15:49 UTC

test_utils.test_get_id_from_cookie() A

↳ Parent: test_utils

Complexity

Conditions

Size

Total Lines	10
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	2
dl	0
loc	10
ccs	3
cts	3
cp	1
crap	1
rs	10
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")


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

kytos-ng / telemetry_int

Push — master ( 92444a...eafb59 )

test_utils.test_get_id_from_cookie() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like