build.tests.unit.test_utils.TestUtils.test_compare_endpoint_trace() - Code Metrics - Inspection of "Checking on EVCs before acquiring lock" - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#626)

by Aldo

created 2025-02-18 23:40 UTC

TestUtils.test_compare_endpoint_trace() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	23
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	17
nop	3
dl	0
loc	23
ccs	9
cts	9
cp	1
crap	1
rs	9.55
c	0
b	0
f	0

"""Module to test the utls.py file."""
from unittest.mock import MagicMock, Mock
from operator import not_
import pytest

from kytos.core.common import EntityStatus
from napps.kytos.mef_eline.exceptions import DisabledSwitch
from napps.kytos.mef_eline.utils import (check_disabled_component,
                                         compare_endpoint_trace,
                                         compare_uni_out_trace,
                                         get_vlan_tags_and_masks, map_dl_vlan,
                                         merge_flow_dicts, prepare_delete_flow,
                                         check_interface_on_evc)


# pylint: disable=too-many-public-methods, too-many-lines
class TestUtils:
    """Test utility functions."""

    @pytest.mark.parametrize(
        "switch,expected",
        [
            (
                MagicMock(dpid="1234"),
                True
            ),
            (
                MagicMock(dpid="2345"),
                False
            )
        ]
    )
    def test_compare_endpoint_trace(self, switch, expected):
        """Test method compare_endpoint_trace"""
        trace = {"dpid": "1234", "port": 2, "vlan": 123}

        endpoint = MagicMock()
        endpoint.port_number = 2
        vlan = 123
        endpoint.switch = switch
        assert compare_endpoint_trace(endpoint, vlan, trace) == expected
        assert compare_endpoint_trace(endpoint, None, trace) == expected

    def test_compare_uni_out_trace(self):
        """Test compare_uni_out_trace method."""
        # case1: trace without 'out' info, should return True
        interface = MagicMock()
        assert compare_uni_out_trace(None, interface, {})

        # case2: trace with valid port and VLAN, should return True
        interface.port_number = 1
        tag_value = 123
        trace = {"out": {"port": 1, "vlan": 123}}
        assert compare_uni_out_trace(tag_value, interface, trace)

        # case3: UNI has VLAN but trace dont have, should return False
        trace = {"out": {"port": 1}}
        assert compare_uni_out_trace(tag_value, interface, trace) is False

        # case4: UNI and trace dont have VLAN should return True
        assert compare_uni_out_trace(None, interface, trace)

        # case5: UNI dont have VLAN but trace has, should return False
        trace = {"out": {"port": 1, "vlan": 123}}
        assert compare_uni_out_trace(None, interface, trace) is False

    def test_map_dl_vlan(self):
        """Test map_dl_vlan"""
        cases = {0: None, "untagged": None, "any": 1, "4096/4096": 1, 10: 10}
        for value, mapped in cases.items():
            result = map_dl_vlan(value)
            assert result == mapped

    @pytest.mark.parametrize(
        "vlan_range,expected",
        [
            (
                [[101, 200]],
                [
                    101,
                    "102/4094",
                    "104/4088",
                    "112/4080",
                    "128/4032",
                    "192/4088",
                    200,
                ]
            ),
            (
                [[101, 90]],
                []
            ),
            (
                [[34, 34]],
                [34]
            ),
            (
                [
                    [34, 34],
                    [128, 128],
                    [130, 135]
                ],
                [
                    34,
                    128,
                    "130/4094",
                    "132/4092"
                ]
            )
        ]
    )
    def test_get_vlan_tags_and_masks(self, vlan_range, expected):
        """Test get_vlan_tags_and_masks"""
        assert get_vlan_tags_and_masks(vlan_range) == expected

    def test_check_disabled_component(self):
        """Test check disabled component"""
        uni_a = MagicMock()
        switch = MagicMock()
        switch.status = EntityStatus.DISABLED
        uni_a.interface.switch = switch

        uni_z = MagicMock()
        uni_z.interface.switch = switch

        # Switch disabled
        with pytest.raises(DisabledSwitch):
            check_disabled_component(uni_a, uni_z)

        # Uni_a interface disabled
        switch.status = EntityStatus.UP
        uni_a.interface.status = EntityStatus.DISABLED
        with pytest.raises(DisabledSwitch):
            check_disabled_component(uni_a, uni_z)

        # Uni_z interface disabled
        uni_a.interface.status = EntityStatus.UP
        uni_z.interface.status = EntityStatus.DISABLED
        with pytest.raises(DisabledSwitch):
            check_disabled_component(uni_a, uni_z)

        # There is no disabled component
        uni_z.interface.status = EntityStatus.UP
        check_disabled_component(uni_a, uni_z)

    @pytest.mark.parametrize(
        "src1,src2,src3,expected",
        [
            (
                {"dpida": [10, 11, 12]},
                {"dpida": [11, 20, 21]},
                {"dpidb": [30, 31, 32]},
                {"dpida": [10, 11, 12, 11, 20, 21], "dpidb": [30, 31, 32]},
            ),
            (
                {"dpida": [10, 11, 12]},
                {"dpida": [11, 20, 21], "dpidb": [40, 41, 42]},
                {"dpidb": [30, 31, 32]},
                {"dpida": [10, 11, 12, 11, 20, 21],
                 "dpidb": [40, 41, 42, 30, 31, 32]},
            ),
        ]
    )
    def test_merge_flow_dicts(self, src1, src2, src3, expected) -> None:
        """test merge flow dicts."""
        assert merge_flow_dicts({}, src1, src2, src3) == expected

    def test_prepare_delete_flow(self):
        """Test prepare_delete_flow"""
        cookie_mask = int(0xffffffffffffffff)
        flow_mod = {'00:01': [
            {
                'match': {'in_port': 1, 'dl_vlan': 22},
                'cookie': 12275899796742638400,
                'actions': [{'action_type': 'pop_vlan'}],
                'owner': 'mef_eline',
                'table_group': 'evpl',
                'table_id': 0,
                'priority': 20000
            },
            {
                'match': {'in_port': 3, 'dl_vlan': 1},
                'cookie': 12275899796742638400,
                'actions': [{'action_type': 'pop_vlan'}],
                'owner': 'mef_eline',
                'table_group': 'evpl',
                'table_id': 0,
                'priority': 20000
            }
        ]}
        actual_flows = prepare_delete_flow(flow_mod)
        assert '00:01' in actual_flows
        for i in range(len(actual_flows['00:01'])):
            assert (actual_flows['00:01'][i]['cookie'] ==
                    flow_mod["00:01"][i]['cookie'])
            assert (actual_flows['00:01'][i]['match'] ==
                    flow_mod["00:01"][i]['match'])
            assert actual_flows['00:01'][i]['cookie_mask'] == cookie_mask

    # pylint: disable=too-many-arguments
    @pytest.mark.parametrize(
        "intf_a_status, intf_z_status, is_active, is_uni, function, expected",
        [
            # link_DOWN
            (
                EntityStatus.DOWN, EntityStatus.DOWN,
                True, True, lambda x: x, True

            ),
            (
                EntityStatus.UP, EntityStatus.UP,
                True, True, lambda x: x, False

            ),
            (
                EntityStatus.DOWN, EntityStatus.UP,
                False, True, lambda x: x, False

            ),
            (
                EntityStatus.UP, EntityStatus.UP,
                False, True, lambda x: x, False

            ),
            (  # Not UNI
                EntityStatus.DOWN, EntityStatus.DOWN,
                True, False, lambda x: x, False

            ),
            # link_up
            (
                EntityStatus.DOWN, EntityStatus.DOWN,
                True, True, not_, False
            ),
            (
                EntityStatus.UP, EntityStatus.UP,
                True, True, not_, False
            ),
            (
                EntityStatus.DOWN, EntityStatus.UP,
                False, True, not_, False
            ),
            (
                EntityStatus.UP, EntityStatus.UP,
                False, True, not_, True
            ),
            (  # Not UNI
                EntityStatus.UP, EntityStatus.UP,
                False, False, not_, False
            ),
        ]
    )
    def test_check_interface_on_evc(
        self,
        intf_a_status,
        intf_z_status,
        is_active,
        is_uni,
        function,
        expected
    ):
        """Test check_interface_on_evc when interface."""
        evc = Mock()
        evc.uni_a.interface.status = intf_a_status
        evc.uni_z.interface.status = intf_z_status
        if is_uni:
            interface = evc.uni_a.interface
        else:
            interface = Mock()
        evc.is_active.return_value = is_active
        assert check_interface_on_evc(evc, interface, function) is expected


1		"""Module to test the utls.py file."""
2	1	from unittest.mock import MagicMock, Mock
3	1	from operator import not_
4	1	import pytest
5
6	1	from kytos.core.common import EntityStatus
7	1	from napps.kytos.mef_eline.exceptions import DisabledSwitch
8	1	from napps.kytos.mef_eline.utils import (check_disabled_component,
9		compare_endpoint_trace,
10		compare_uni_out_trace,
11		get_vlan_tags_and_masks, map_dl_vlan,
12		merge_flow_dicts, prepare_delete_flow,
13		check_interface_on_evc)
14
15
16		# pylint: disable=too-many-public-methods, too-many-lines
17	1	class TestUtils:
18		"""Test utility functions."""
19
20	1	@pytest.mark.parametrize(
21		"switch,expected",
22		[
23		(
24		MagicMock(dpid="1234"),
25		True
26		),
27		(
28		MagicMock(dpid="2345"),
29		False
30		)
31		]
32		)
33	1	def test_compare_endpoint_trace(self, switch, expected):
34		"""Test method compare_endpoint_trace"""
35	1	trace = {"dpid": "1234", "port": 2, "vlan": 123}
36
37	1	endpoint = MagicMock()
38	1	endpoint.port_number = 2
39	1	vlan = 123
40	1	endpoint.switch = switch
41	1	assert compare_endpoint_trace(endpoint, vlan, trace) == expected
42	1	assert compare_endpoint_trace(endpoint, None, trace) == expected
43
44	1	def test_compare_uni_out_trace(self):
45		"""Test compare_uni_out_trace method."""
46		# case1: trace without 'out' info, should return True
47	1	interface = MagicMock()
48	1	assert compare_uni_out_trace(None, interface, {})
49
50		# case2: trace with valid port and VLAN, should return True
51	1	interface.port_number = 1
52	1	tag_value = 123
53	1	trace = {"out": {"port": 1, "vlan": 123}}
54	1	assert compare_uni_out_trace(tag_value, interface, trace)
55
56		# case3: UNI has VLAN but trace dont have, should return False
57	1	trace = {"out": {"port": 1}}
58	1	assert compare_uni_out_trace(tag_value, interface, trace) is False
59
60		# case4: UNI and trace dont have VLAN should return True
61	1	assert compare_uni_out_trace(None, interface, trace)
62
63		# case5: UNI dont have VLAN but trace has, should return False
64	1	trace = {"out": {"port": 1, "vlan": 123}}
65	1	assert compare_uni_out_trace(None, interface, trace) is False
66
67	1	def test_map_dl_vlan(self):
68		"""Test map_dl_vlan"""
69	1	cases = {0: None, "untagged": None, "any": 1, "4096/4096": 1, 10: 10}
70	1	for value, mapped in cases.items():
71	1	result = map_dl_vlan(value)
72	1	assert result == mapped
73
74	1	@pytest.mark.parametrize(
75		"vlan_range,expected",
76		[
77		(
78		[[101, 200]],
79		[
80		101,
81		"102/4094",
82		"104/4088",
83		"112/4080",
84		"128/4032",
85		"192/4088",
86		200,
87		]
88		),
89		(
90		[[101, 90]],
91		[]
92		),
93		(
94		[[34, 34]],
95		[34]
96		),
97		(
98		[
99		[34, 34],
100		[128, 128],
101		[130, 135]
102		],
103		[
104		34,
105		128,
106		"130/4094",
107		"132/4092"
108		]
109		)
110		]
111		)
112	1	def test_get_vlan_tags_and_masks(self, vlan_range, expected):
113		"""Test get_vlan_tags_and_masks"""
114	1	assert get_vlan_tags_and_masks(vlan_range) == expected
115
116	1	def test_check_disabled_component(self):
117		"""Test check disabled component"""
118	1	uni_a = MagicMock()
119	1	switch = MagicMock()
120	1	switch.status = EntityStatus.DISABLED
121	1	uni_a.interface.switch = switch
122
123	1	uni_z = MagicMock()
124	1	uni_z.interface.switch = switch
125
126		# Switch disabled
127	1	with pytest.raises(DisabledSwitch):
128	1	check_disabled_component(uni_a, uni_z)
129
130		# Uni_a interface disabled
131	1	switch.status = EntityStatus.UP
132	1	uni_a.interface.status = EntityStatus.DISABLED
133	1	with pytest.raises(DisabledSwitch):
134	1	check_disabled_component(uni_a, uni_z)
135
136		# Uni_z interface disabled
137	1	uni_a.interface.status = EntityStatus.UP
138	1	uni_z.interface.status = EntityStatus.DISABLED
139	1	with pytest.raises(DisabledSwitch):
140	1	check_disabled_component(uni_a, uni_z)
141
142		# There is no disabled component
143	1	uni_z.interface.status = EntityStatus.UP
144	1	check_disabled_component(uni_a, uni_z)
145
146	1	@pytest.mark.parametrize(
147		"src1,src2,src3,expected",
148		[
149		(
150		{"dpida": [10, 11, 12]},
151		{"dpida": [11, 20, 21]},
152		{"dpidb": [30, 31, 32]},
153		{"dpida": [10, 11, 12, 11, 20, 21], "dpidb": [30, 31, 32]},
154		),
155		(
156		{"dpida": [10, 11, 12]},
157		{"dpida": [11, 20, 21], "dpidb": [40, 41, 42]},
158		{"dpidb": [30, 31, 32]},
159		{"dpida": [10, 11, 12, 11, 20, 21],
160		"dpidb": [40, 41, 42, 30, 31, 32]},
161		),
162		]
163		)
164	1	def test_merge_flow_dicts(self, src1, src2, src3, expected) -> None:
165		"""test merge flow dicts."""
166	1	assert merge_flow_dicts({}, src1, src2, src3) == expected
167
168	1	def test_prepare_delete_flow(self):
169		"""Test prepare_delete_flow"""
170	1	cookie_mask = int(0xffffffffffffffff)
171	1	flow_mod = {'00:01': [
172		{
173		'match': {'in_port': 1, 'dl_vlan': 22},
174		'cookie': 12275899796742638400,
175		'actions': [{'action_type': 'pop_vlan'}],
176		'owner': 'mef_eline',
177		'table_group': 'evpl',
178		'table_id': 0,
179		'priority': 20000
180		},
181		{
182		'match': {'in_port': 3, 'dl_vlan': 1},
183		'cookie': 12275899796742638400,
184		'actions': [{'action_type': 'pop_vlan'}],
185		'owner': 'mef_eline',
186		'table_group': 'evpl',
187		'table_id': 0,
188		'priority': 20000
189		}
190		]}
191	1	actual_flows = prepare_delete_flow(flow_mod)
192	1	assert '00:01' in actual_flows
193	1	for i in range(len(actual_flows['00:01'])):
194	1	assert (actual_flows['00:01'][i]['cookie'] ==
195		flow_mod["00:01"][i]['cookie'])
196	1	assert (actual_flows['00:01'][i]['match'] ==
197		flow_mod["00:01"][i]['match'])
198	1	assert actual_flows['00:01'][i]['cookie_mask'] == cookie_mask
199
200		# pylint: disable=too-many-arguments
201	1	@pytest.mark.parametrize(
202		"intf_a_status, intf_z_status, is_active, is_uni, function, expected",
203		[
204		# link_DOWN
205		(
206		EntityStatus.DOWN, EntityStatus.DOWN,
207		True, True, lambda x: x, True
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-02-18 23:44 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
208		),
209		(
210		EntityStatus.UP, EntityStatus.UP,
211		True, True, lambda x: x, False
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-02-18 23:44 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
212		),
213		(
214		EntityStatus.DOWN, EntityStatus.UP,
215		False, True, lambda x: x, False
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-02-18 23:44 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
216		),
217		(
218		EntityStatus.UP, EntityStatus.UP,
219		False, True, lambda x: x, False
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-02-18 23:44 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
220		),
221		( # Not UNI
222		EntityStatus.DOWN, EntityStatus.DOWN,
223		True, False, lambda x: x, False
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-02-18 23:44 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
224		),
225		# link_up
226		(
227		EntityStatus.DOWN, EntityStatus.DOWN,
228		True, True, not_, False
229		),
230		(
231		EntityStatus.UP, EntityStatus.UP,
232		True, True, not_, False
233		),
234		(
235		EntityStatus.DOWN, EntityStatus.UP,
236		False, True, not_, False
237		),
238		(
239		EntityStatus.UP, EntityStatus.UP,
240		False, True, not_, True
241		),
242		( # Not UNI
243		EntityStatus.UP, EntityStatus.UP,
244		False, False, not_, False
245		),
246		]
247		)
248	1	def test_check_interface_on_evc(
249		self,
250		intf_a_status,
251		intf_z_status,
252		is_active,
253		is_uni,
254		function,
255		expected
256		):
257		"""Test check_interface_on_evc when interface."""
258	1	evc = Mock()
259	1	evc.uni_a.interface.status = intf_a_status
260	1	evc.uni_z.interface.status = intf_z_status
261	1	if is_uni:
262	1	interface = evc.uni_a.interface
263		else:
264	1	interface = Mock()
265	1	evc.is_active.return_value = is_active
266		assert check_interface_on_evc(evc, interface, function) is expected
267

kytos-ng / mef_eline

Pull Request — master (#626)

TestUtils.test_compare_endpoint_trace() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like