build.tests.unit.test_utils.TestUtils.test_match_field_dl_vlan() - Code Metrics - Inspection of "feature: Support `untagged` and `any` for `vlan`" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#82)

unknown

created 2023-03-23 17:12 UTC

TestUtils.test_match_field_dl_vlan() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	15
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	13
nop	1
dl	0
loc	15
rs	9.75
c	0
b	0
f	0

"""Module to test the utils.py file."""
from unittest import TestCase
from unittest.mock import patch, MagicMock

from kytos.core.interface import Interface
from kytos.lib.helpers import get_controller_mock, get_link_mock
from napps.amlight.sdntrace_cp import utils


# pylint: disable=too-many-public-methods, duplicate-code, protected-access
class TestUtils(TestCase):
    """Test utils.py functions."""

    def test_convert_list_entries(self):
        """Verify convert entries with a list of one example"""
        eth = {"dl_vlan": 100}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}

        result = utils.convert_list_entries([entries])

        self.assertEqual(
            result,
            [{
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 1,
                "dl_vlan": [100],
            }],
        )

    def test_convert_entries_vlan(self):
        """Verify convert entries with simple example with vlan."""

        eth = {"dl_vlan": 100}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}

        result = utils.convert_entries(entries)

        self.assertEqual(
            result,
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 1,
                "dl_vlan": [100],
            },
        )

    def test_prepare_json(self):
        """Verify prepare json with simple tracepath result."""
        trace_result = []
        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 1,
                "time": "2022-06-01 01:01:01.100000",
                "type": "starting",
            }
        }
        trace_result.append(trace_step)

        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
                "time": "2022-06-01 01:01:01.100000",
                "type": "intermediary",
                "vlan": 100,
            },
            "out": {
                "port": 1,
                "vlan": 123,
            },
        }
        trace_result.append(trace_step)

        result = utils.prepare_json(trace_result)

        self.assertEqual(
            result,
            {
                "result": [
                    {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "port": 1,
                        "time": "2022-06-01 01:01:01.100000",
                        "type": "starting",
                    },
                    {
                        "dpid": "00:00:00:00:00:00:00:03",
                        "port": 3,
                        "time": "2022-06-01 01:01:01.100000",
                        "type": "intermediary",
                        "vlan": 100,
                        "out": {"port": 1, "vlan": 123},
                    },
                ]
            },
        )

    def test_prepare_list_json(self):
        """Verify prepare list with a simple tracepath result."""
        trace_result = []
        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 1,
                "time": "2022-06-01 01:01:01.100000",
                "type": "starting",
            }
        }
        trace_result.append(trace_step)

        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
                "time": "2022-06-01 01:01:01.100000",
                "type": "intermediary",
                "vlan": 100,
            },
            "out": {
                "port": 1,
                "vlan": 123,
            },
        }
        trace_result.append(trace_step)

        result = utils._prepare_json(trace_result)

        self.assertEqual(
            result, [
                    {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "port": 1,
                        "time": "2022-06-01 01:01:01.100000",
                        "type": "starting",
                    },
                    {
                        "dpid": "00:00:00:00:00:00:00:03",
                        "port": 3,
                        "time": "2022-06-01 01:01:01.100000",
                        "type": "intermediary",
                        "vlan": 100,
                        "out": {"port": 1, "vlan": 123},
                    },
                ]
        )

    def test_prepare_json_empty(self):
        """Verify prepare json with empty result."""
        trace_result = []

        result = utils.prepare_json(trace_result)

        self.assertEqual(result, {"result": []})

    def test_format_result(self):
        """Verify format resul with simple tracepath result."""
        trace_result = []
        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 1,
                "time": "2022-06-02 02:02:02.200000",
                "type": "starting",
            }
        }
        trace_result.append(trace_step)

        trace_step = {
            "in": {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
                "time": "2022-06-02 02:02:02.200000",
                "type": "intermediary",
                "vlan": 100,
            },
            "out": {"port": 2, "vlan": 200},
        }
        trace_result.append(trace_step)

        formatted = utils.format_result(trace_result)

        self.assertEqual(
            formatted,
            [
                {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1},
                {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "in_port": 3,
                    "out_port": 2,
                    "out_vlan": 200,
                    "in_vlan": 100,
                },
            ],
        )

    def test_compare_endpoints1(self):
        """Test for compare endpoinst for the first internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:01",
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:02",
        }

        # Test endpoint1 dpid != endpoint2 dpid
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints2(self):
        """Test for compare endpoinst for the second internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "out_port": 2,
            "out_vlan": 200,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "in_vlan": 100,
        }

        # Test endpoint1 without in_port
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "in_vlan": 100,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "in_vlan": 100,
        }

        # Test endpoint2 without out_port
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "in_vlan": 100,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "out_port": 2,
            "out_vlan": 200,
        }

        # Test endpoint1 in_port != endpoint2 out_port
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints3(self):
        """Test for compare endpoinst for the third internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "out_port": 2,
            "in_vlan": 100,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 2,
            "out_port": 3,
            "out_vlan": 200,
        }

        # Test endpoint1 in_vlan != endpoint2 out_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints4(self):

        """Test for compare endpoinst for the first internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "out_port": 2,
            "in_vlan": 100,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 2,
            "out_port": 3,
        }

        # Test endpoint1 with in_vlan and endpoint2 without out_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "out_port": 2,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 2,
            "out_port": 3,
            "out_vlan": 200,
        }

        # Test endpoint1 without in_vlan and endpoint2 with out_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints5(self):
        """Test for compare endpoinst for the fifth internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 3,
            "out_port": 2,
            "out_vlan": 200,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 2,
            "out_port": 3,
            "in_vlan": 100,
        }

        # Test endpoint1 out_vlan != endpoint2 in_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints6(self):

        """Test for compare endpoinst for the fifth internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 3,
            "out_port": 2,
            "out_vlan": 200,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 2,
            "out_port": 3,
        }

        # Test endpoint1 with out_vlan and endpoint2 without in_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 3,
            "out_port": 2,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:01",
            "in_port": 2,
            "out_port": 3,
            "in_vlan": 100,
        }

        # Test endpoint1 without out_vlan and endpoint2 with in_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)
        self.assertFalse(result)

    def test_compare_endpoints(self):
        """Test for compare endpoinst for the fifth internal conditional."""
        endpoint1 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "in_port": 3,
            "in_vlan": 100,
        }
        endpoint2 = {
            "dpid": "00:00:00:00:00:00:00:03",
            "out_port": 3,
            "out_vlan": 100,
        }

        # Test endpoint1 out_vlan != endpoint2 in_vlan
        result = utils._compare_endpoints(endpoint1, endpoint2)

        self.assertTrue(result)

    def test_find_endpoint_b(self):
        """Test find endpoint with interface equals link endpoint B."""
        port = 1

        mock_interface = Interface("interface A", port, MagicMock())
        mock_interface.address = "00:00:00:00:00:00:00:01"
        mock_interface.link = get_link_mock(
            "00:00:00:00:00:00:00:02", "00:00:00:00:00:00:00:01"
        )

        mock_switch = MagicMock()
        mock_switch.get_interface_by_port_no.return_value = mock_interface

        result = utils.find_endpoint(mock_switch, port)
        self.assertEqual(result, mock_interface.link.endpoint_a)

    def test_find_endpoint_a(self):
        """Test find endpoint with interface equals link endpoint A."""
        port = 1

        mock_interface = Interface("interface A", port, MagicMock())
        mock_interface.address = "00:00:00:00:00:00:00:01"
        mock_interface.link = get_link_mock(
            "00:00:00:00:00:00:00:01", "00:00:00:00:00:00:00:03"
        )

        mock_switch = MagicMock()
        mock_switch.get_interface_by_port_no.return_value = mock_interface

        result = utils.find_endpoint(mock_switch, port)
        self.assertEqual(result, mock_interface.link.endpoint_b)

    def test_find_endpoint_link_none(self):
        """Test find endpoint without link."""
        port = 1

        mock_interface = Interface("interface A", port, MagicMock())
        mock_interface.address = "00:00:00:00:00:00:00:01"

        mock_switch = MagicMock()
        mock_switch.get_interface_by_port_no.return_value = mock_interface

        result = utils.find_endpoint(mock_switch, port)
        self.assertIsNone(result)

    def test_convert_vlan(self):
        """Test convert_vlan function"""
        value = 100
        result = utils.convert_vlan(value)
        assert result[0] == 100

        value = "4096/4096"
        result = utils.convert_vlan(value)
        assert result[0] == 4096
        assert result[1] == 4096

    def test_match_field_dl_vlan(self):
        """Test match_field_dl_vlan"""

        result = utils.match_field_dl_vlan(None, 0)
        self.assertTrue(result)
        result = utils.match_field_dl_vlan(None, 10)
        self.assertFalse(result)
        result = utils.match_field_dl_vlan(None, "4096/4096")
        self.assertFalse(result)
        result = utils.match_field_dl_vlan(10, 0)
        self.assertFalse(result)
        result = utils.match_field_dl_vlan(10, 10)
        self.assertTrue(result)
        result = utils.match_field_dl_vlan(10, "4096/4096")
        self.assertTrue(result)


# pylint: disable=too-many-public-methods, too-many-lines
class TestUtilsWithController(TestCase):
    """Test utils.py."""

    def setUp(self):
        # The decorator run_on_thread is patched, so methods that listen
        # for events do not run on threads while tested.
        # Decorators have to be patched before the methods that are
        # decorated with them are imported.
        patch("kytos.core.helpers.run_on_thread", lambda x: x).start()

        self.controller = get_controller_mock()

        self.addCleanup(patch.stopall)

    def test_clean_circuits__empty(self):
        """Test clean circuits for empty circuits."""
        circuits = MagicMock()
        result = utils.clean_circuits(circuits, self.controller)

        self.assertEqual(result, [])

    def test_clean_circuits__no_sub(self):
        """Test clean circuits with just one circuit."""
        formatted = [
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "in_port": 3,
                "out_port": 2,
                "out_vlan": 200,
                "in_vlan": 100,
            },
        ]

        circuits = []
        circuits.append({"circuit": formatted, "entries": []})

        result = utils.clean_circuits(circuits, self.controller)

        self.assertTrue(len(result) == 1)
        self.assertEqual(formatted, result[0]["circuit"])

    def test_clean_circuits_with_sub_circuit(self):

        """Test clean circuits without sub-circuits."""
        formatted_a = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
            },
        ]
        formatted_b = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
            },
        ]

        circuits = []
        circuits.append({"circuit": formatted_a, "entries": []})
        circuits.append({"circuit": formatted_b, "entries": []})

        # Test cleaning one sub-circuit
        result = utils.clean_circuits(circuits, self.controller)

        # Result must be the circuits without the sub-circuit
        self.assertTrue(len(result) == 1)
        self.assertEqual(formatted_b, result[0]["circuit"])

    def test_clean_circuits_without_sub_circuit(self):

        """Test clean circuits with one sub-circuit."""
        formatted_a = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
            },
        ]
        formatted_b = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
            },
            {
                "dpid": "00:00:00:00:00:00:00:04",
            },
        ]

        circuits = []
        circuits.append({"circuit": formatted_a, "entries": []})
        circuits.append({"circuit": formatted_b, "entries": []})

        # Test circuits withou sub-circuits.
        result = utils.clean_circuits(circuits, self.controller)

        # Result must be equal to the circuits parameter
        self.assertTrue(len(result) == 2)
        self.assertEqual(circuits, result)


1		"""Module to test the utils.py file."""
2		from unittest import TestCase
3		from unittest.mock import patch, MagicMock
4
5		from kytos.core.interface import Interface
6		from kytos.lib.helpers import get_controller_mock, get_link_mock
7		from napps.amlight.sdntrace_cp import utils
8
9
10		# pylint: disable=too-many-public-methods, duplicate-code, protected-access
11		class TestUtils(TestCase):
12		"""Test utils.py functions."""
13
14		def test_convert_list_entries(self):
15		"""Verify convert entries with a list of one example"""
16		eth = {"dl_vlan": 100}
17		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
18		switch = {"switch": dpid, "eth": eth}
19		entries = {"trace": switch}
20
21		result = utils.convert_list_entries([entries])
22
23		self.assertEqual(
24		result,
25		[{
26		"dpid": "00:00:00:00:00:00:00:01",
27		"in_port": 1,
28		"dl_vlan": [100],
29		}],
30		)
31
32		def test_convert_entries_vlan(self):
33		"""Verify convert entries with simple example with vlan."""
34
35		eth = {"dl_vlan": 100}
36		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
37		switch = {"switch": dpid, "eth": eth}
38		entries = {"trace": switch}
39
40		result = utils.convert_entries(entries)
41
42		self.assertEqual(
43		result,
44		{
45		"dpid": "00:00:00:00:00:00:00:01",
46		"in_port": 1,
47		"dl_vlan": [100],
48		},
49		)
50
51		def test_prepare_json(self):
52		"""Verify prepare json with simple tracepath result."""
53		trace_result = []
54		trace_step = {
55		"in": {
56		"dpid": "00:00:00:00:00:00:00:01",
57		"port": 1,
58		"time": "2022-06-01 01:01:01.100000",
59		"type": "starting",
60		}
61		}
62		trace_result.append(trace_step)
63
64		trace_step = {
65		"in": {
66		"dpid": "00:00:00:00:00:00:00:03",
67		"port": 3,
68		"time": "2022-06-01 01:01:01.100000",
69		"type": "intermediary",
70		"vlan": 100,
71		},
72		"out": {
73		"port": 1,
74		"vlan": 123,
75		},
76		}
77		trace_result.append(trace_step)
78
79		result = utils.prepare_json(trace_result)
80
81		self.assertEqual(
82		result,
83		{
84		"result": [
85		{
86		"dpid": "00:00:00:00:00:00:00:01",
87		"port": 1,
88		"time": "2022-06-01 01:01:01.100000",
89		"type": "starting",
90		},
91		{
92		"dpid": "00:00:00:00:00:00:00:03",
93		"port": 3,
94		"time": "2022-06-01 01:01:01.100000",
95		"type": "intermediary",
96		"vlan": 100,
97		"out": {"port": 1, "vlan": 123},
98		},
99		]
100		},
101		)
102
103		def test_prepare_list_json(self):
104		"""Verify prepare list with a simple tracepath result."""
105		trace_result = []
106		trace_step = {
107		"in": {
108		"dpid": "00:00:00:00:00:00:00:01",
109		"port": 1,
110		"time": "2022-06-01 01:01:01.100000",
111		"type": "starting",
112		}
113		}
114		trace_result.append(trace_step)
115
116		trace_step = {
117		"in": {
118		"dpid": "00:00:00:00:00:00:00:03",
119		"port": 3,
120		"time": "2022-06-01 01:01:01.100000",
121		"type": "intermediary",
122		"vlan": 100,
123		},
124		"out": {
125		"port": 1,
126		"vlan": 123,
127		},
128		}
129		trace_result.append(trace_step)
130
131		result = utils._prepare_json(trace_result)
132
133		self.assertEqual(
134		result, [
135		{
136		"dpid": "00:00:00:00:00:00:00:01",
137		"port": 1,
138		"time": "2022-06-01 01:01:01.100000",
139		"type": "starting",
140		},
141		{
142		"dpid": "00:00:00:00:00:00:00:03",
143		"port": 3,
144		"time": "2022-06-01 01:01:01.100000",
145		"type": "intermediary",
146		"vlan": 100,
147		"out": {"port": 1, "vlan": 123},
148		},
149		]
150		)
151
152		def test_prepare_json_empty(self):
153		"""Verify prepare json with empty result."""
154		trace_result = []
155
156		result = utils.prepare_json(trace_result)
157
158		self.assertEqual(result, {"result": []})
159
160		def test_format_result(self):
161		"""Verify format resul with simple tracepath result."""
162		trace_result = []
163		trace_step = {
164		"in": {
165		"dpid": "00:00:00:00:00:00:00:01",
166		"port": 1,
167		"time": "2022-06-02 02:02:02.200000",
168		"type": "starting",
169		}
170		}
171		trace_result.append(trace_step)
172
173		trace_step = {
174		"in": {
175		"dpid": "00:00:00:00:00:00:00:03",
176		"port": 3,
177		"time": "2022-06-02 02:02:02.200000",
178		"type": "intermediary",
179		"vlan": 100,
180		},
181		"out": {"port": 2, "vlan": 200},
182		}
183		trace_result.append(trace_step)
184
185		formatted = utils.format_result(trace_result)
186
187		self.assertEqual(
188		formatted,
189		[
190		{"dpid": "00:00:00:00:00:00:00:01", "in_port": 1},
191		{
192		"dpid": "00:00:00:00:00:00:00:03",
193		"in_port": 3,
194		"out_port": 2,
195		"out_vlan": 200,
196		"in_vlan": 100,
197		},
198		],
199		)
200
201		def test_compare_endpoints1(self):
202		"""Test for compare endpoinst for the first internal conditional."""
203		endpoint1 = {
204		"dpid": "00:00:00:00:00:00:00:01",
205		}
206		endpoint2 = {
207		"dpid": "00:00:00:00:00:00:00:02",
208		}
209
210		# Test endpoint1 dpid != endpoint2 dpid
211		result = utils._compare_endpoints(endpoint1, endpoint2)
212		self.assertFalse(result)
213
214		def test_compare_endpoints2(self):
215		"""Test for compare endpoinst for the second internal conditional."""
216		endpoint1 = {
217		"dpid": "00:00:00:00:00:00:00:03",
218		"out_port": 2,
219		"out_vlan": 200,
220		}
221		endpoint2 = {
222		"dpid": "00:00:00:00:00:00:00:03",
223		"in_port": 3,
224		"in_vlan": 100,
225		}
226
227		# Test endpoint1 without in_port
228		result = utils._compare_endpoints(endpoint1, endpoint2)
229		self.assertFalse(result)
230
231		endpoint1 = {
232		"dpid": "00:00:00:00:00:00:00:03",
233		"in_port": 3,
234		"in_vlan": 100,
235		}
236		endpoint2 = {
237		"dpid": "00:00:00:00:00:00:00:03",
238		"in_port": 3,
239		"in_vlan": 100,
240		}
241
242		# Test endpoint2 without out_port
243		result = utils._compare_endpoints(endpoint1, endpoint2)
244		self.assertFalse(result)
245
246		endpoint1 = {
247		"dpid": "00:00:00:00:00:00:00:03",
248		"in_port": 3,
249		"in_vlan": 100,
250		}
251		endpoint2 = {
252		"dpid": "00:00:00:00:00:00:00:03",
253		"out_port": 2,
254		"out_vlan": 200,
255		}
256
257		# Test endpoint1 in_port != endpoint2 out_port
258		result = utils._compare_endpoints(endpoint1, endpoint2)
259		self.assertFalse(result)
260
261		def test_compare_endpoints3(self):
262		"""Test for compare endpoinst for the third internal conditional."""
263		endpoint1 = {
264		"dpid": "00:00:00:00:00:00:00:03",
265		"in_port": 3,
266		"out_port": 2,
267		"in_vlan": 100,
268		}
269		endpoint2 = {
270		"dpid": "00:00:00:00:00:00:00:03",
271		"in_port": 2,
272		"out_port": 3,
273		"out_vlan": 200,
274		}
275
276		# Test endpoint1 in_vlan != endpoint2 out_vlan
277		result = utils._compare_endpoints(endpoint1, endpoint2)
278		self.assertFalse(result)
279
280	View Code Duplication	def test_compare_endpoints4(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
281		"""Test for compare endpoinst for the first internal conditional."""
282		endpoint1 = {
283		"dpid": "00:00:00:00:00:00:00:03",
284		"in_port": 3,
285		"out_port": 2,
286		"in_vlan": 100,
287		}
288		endpoint2 = {
289		"dpid": "00:00:00:00:00:00:00:03",
290		"in_port": 2,
291		"out_port": 3,
292		}
293
294		# Test endpoint1 with in_vlan and endpoint2 without out_vlan
295		result = utils._compare_endpoints(endpoint1, endpoint2)
296		self.assertFalse(result)
297
298		endpoint1 = {
299		"dpid": "00:00:00:00:00:00:00:03",
300		"in_port": 3,
301		"out_port": 2,
302		}
303		endpoint2 = {
304		"dpid": "00:00:00:00:00:00:00:03",
305		"in_port": 2,
306		"out_port": 3,
307		"out_vlan": 200,
308		}
309
310		# Test endpoint1 without in_vlan and endpoint2 with out_vlan
311		result = utils._compare_endpoints(endpoint1, endpoint2)
312		self.assertFalse(result)
313
314		def test_compare_endpoints5(self):
315		"""Test for compare endpoinst for the fifth internal conditional."""
316		endpoint1 = {
317		"dpid": "00:00:00:00:00:00:00:01",
318		"in_port": 3,
319		"out_port": 2,
320		"out_vlan": 200,
321		}
322		endpoint2 = {
323		"dpid": "00:00:00:00:00:00:00:01",
324		"in_port": 2,
325		"out_port": 3,
326		"in_vlan": 100,
327		}
328
329		# Test endpoint1 out_vlan != endpoint2 in_vlan
330		result = utils._compare_endpoints(endpoint1, endpoint2)
331		self.assertFalse(result)
332
333	View Code Duplication	def test_compare_endpoints6(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
334		"""Test for compare endpoinst for the fifth internal conditional."""
335		endpoint1 = {
336		"dpid": "00:00:00:00:00:00:00:01",
337		"in_port": 3,
338		"out_port": 2,
339		"out_vlan": 200,
340		}
341		endpoint2 = {
342		"dpid": "00:00:00:00:00:00:00:01",
343		"in_port": 2,
344		"out_port": 3,
345		}
346
347		# Test endpoint1 with out_vlan and endpoint2 without in_vlan
348		result = utils._compare_endpoints(endpoint1, endpoint2)
349		self.assertFalse(result)
350
351		endpoint1 = {
352		"dpid": "00:00:00:00:00:00:00:01",
353		"in_port": 3,
354		"out_port": 2,
355		}
356		endpoint2 = {
357		"dpid": "00:00:00:00:00:00:00:01",
358		"in_port": 2,
359		"out_port": 3,
360		"in_vlan": 100,
361		}
362
363		# Test endpoint1 without out_vlan and endpoint2 with in_vlan
364		result = utils._compare_endpoints(endpoint1, endpoint2)
365		self.assertFalse(result)
366
367		def test_compare_endpoints(self):
368		"""Test for compare endpoinst for the fifth internal conditional."""
369		endpoint1 = {
370		"dpid": "00:00:00:00:00:00:00:03",
371		"in_port": 3,
372		"in_vlan": 100,
373		}
374		endpoint2 = {
375		"dpid": "00:00:00:00:00:00:00:03",
376		"out_port": 3,
377		"out_vlan": 100,
378		}
379
380		# Test endpoint1 out_vlan != endpoint2 in_vlan
381		result = utils._compare_endpoints(endpoint1, endpoint2)
382
383		self.assertTrue(result)
384
385		def test_find_endpoint_b(self):
386		"""Test find endpoint with interface equals link endpoint B."""
387		port = 1
388
389		mock_interface = Interface("interface A", port, MagicMock())
390		mock_interface.address = "00:00:00:00:00:00:00:01"
391		mock_interface.link = get_link_mock(
392		"00:00:00:00:00:00:00:02", "00:00:00:00:00:00:00:01"
393		)
394
395		mock_switch = MagicMock()
396		mock_switch.get_interface_by_port_no.return_value = mock_interface
397
398		result = utils.find_endpoint(mock_switch, port)
399		self.assertEqual(result, mock_interface.link.endpoint_a)
400
401		def test_find_endpoint_a(self):
402		"""Test find endpoint with interface equals link endpoint A."""
403		port = 1
404
405		mock_interface = Interface("interface A", port, MagicMock())
406		mock_interface.address = "00:00:00:00:00:00:00:01"
407		mock_interface.link = get_link_mock(
408		"00:00:00:00:00:00:00:01", "00:00:00:00:00:00:00:03"
409		)
410
411		mock_switch = MagicMock()
412		mock_switch.get_interface_by_port_no.return_value = mock_interface
413
414		result = utils.find_endpoint(mock_switch, port)
415		self.assertEqual(result, mock_interface.link.endpoint_b)
416
417		def test_find_endpoint_link_none(self):
418		"""Test find endpoint without link."""
419		port = 1
420
421		mock_interface = Interface("interface A", port, MagicMock())
422		mock_interface.address = "00:00:00:00:00:00:00:01"
423
424		mock_switch = MagicMock()
425		mock_switch.get_interface_by_port_no.return_value = mock_interface
426
427		result = utils.find_endpoint(mock_switch, port)
428		self.assertIsNone(result)
429
430		def test_convert_vlan(self):
431		"""Test convert_vlan function"""
432		value = 100
433		result = utils.convert_vlan(value)
434		assert result[0] == 100
435
436		value = "4096/4096"
437		result = utils.convert_vlan(value)
438		assert result[0] == 4096
439		assert result[1] == 4096
440
441		def test_match_field_dl_vlan(self):
442		"""Test match_field_dl_vlan"""
443
444		result = utils.match_field_dl_vlan(None, 0)
445		self.assertTrue(result)
446		result = utils.match_field_dl_vlan(None, 10)
447		self.assertFalse(result)
448		result = utils.match_field_dl_vlan(None, "4096/4096")
449		self.assertFalse(result)
450		result = utils.match_field_dl_vlan(10, 0)
451		self.assertFalse(result)
452		result = utils.match_field_dl_vlan(10, 10)
453		self.assertTrue(result)
454		result = utils.match_field_dl_vlan(10, "4096/4096")
455		self.assertTrue(result)
456
457
458		# pylint: disable=too-many-public-methods, too-many-lines
459		class TestUtilsWithController(TestCase):
460		"""Test utils.py."""
461
462		def setUp(self):
463		# The decorator run_on_thread is patched, so methods that listen
464		# for events do not run on threads while tested.
465		# Decorators have to be patched before the methods that are
466		# decorated with them are imported.
467		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
468
469		self.controller = get_controller_mock()
470
471		self.addCleanup(patch.stopall)
472
473		def test_clean_circuits__empty(self):
474		"""Test clean circuits for empty circuits."""
475		circuits = MagicMock()
476		result = utils.clean_circuits(circuits, self.controller)
477
478		self.assertEqual(result, [])
479
480		def test_clean_circuits__no_sub(self):
481		"""Test clean circuits with just one circuit."""
482		formatted = [
483		{
484		"dpid": "00:00:00:00:00:00:00:03",
485		"in_port": 3,
486		"out_port": 2,
487		"out_vlan": 200,
488		"in_vlan": 100,
489		},
490		]
491
492		circuits = []
493		circuits.append({"circuit": formatted, "entries": []})
494
495		result = utils.clean_circuits(circuits, self.controller)
496
497		self.assertTrue(len(result) == 1)
498		self.assertEqual(formatted, result[0]["circuit"])
499
500	View Code Duplication	def test_clean_circuits_with_sub_circuit(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
501		"""Test clean circuits without sub-circuits."""
502		formatted_a = [
503		{
504		"dpid": "00:00:00:00:00:00:00:01",
505		},
506		{
507		"dpid": "00:00:00:00:00:00:00:03",
508		},
509		]
510		formatted_b = [
511		{
512		"dpid": "00:00:00:00:00:00:00:01",
513		},
514		{
515		"dpid": "00:00:00:00:00:00:00:02",
516		},
517		{
518		"dpid": "00:00:00:00:00:00:00:03",
519		},
520		]
521
522		circuits = []
523		circuits.append({"circuit": formatted_a, "entries": []})
524		circuits.append({"circuit": formatted_b, "entries": []})
525
526		# Test cleaning one sub-circuit
527		result = utils.clean_circuits(circuits, self.controller)
528
529		# Result must be the circuits without the sub-circuit
530		self.assertTrue(len(result) == 1)
531		self.assertEqual(formatted_b, result[0]["circuit"])
532
533	View Code Duplication	def test_clean_circuits_without_sub_circuit(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
534		"""Test clean circuits with one sub-circuit."""
535		formatted_a = [
536		{
537		"dpid": "00:00:00:00:00:00:00:01",
538		},
539		{
540		"dpid": "00:00:00:00:00:00:00:02",
541		},
542		]
543		formatted_b = [
544		{
545		"dpid": "00:00:00:00:00:00:00:01",
546		},
547		{
548		"dpid": "00:00:00:00:00:00:00:03",
549		},
550		{
551		"dpid": "00:00:00:00:00:00:00:04",
552		},
553		]
554
555		circuits = []
556		circuits.append({"circuit": formatted_a, "entries": []})
557		circuits.append({"circuit": formatted_b, "entries": []})
558
559		# Test circuits withou sub-circuits.
560		result = utils.clean_circuits(circuits, self.controller)
561
562		# Result must be equal to the circuits parameter
563		self.assertTrue(len(result) == 2)
564		self.assertEqual(circuits, result)
565

kytos-ng / sdntrace_cp

Pull Request — master (#82)

TestUtils.test_match_field_dl_vlan() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like