build.tests.unit.test_utils.TestUtils.test_convert_entries_translation() - Code Metrics - Inspection of "Add PUT traces endpoint and functionality match_an..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#47)

unknown

created 2022-10-18 20:01 UTC

TestUtils.test_convert_entries_translation() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	32
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	27
nop	1
dl	0
loc	32
rs	9.232
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": "trace",
                "vlan": 100,
            }
        }
        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": "trace",
                        "vlan": 100,
                    },
                ]
            },
        )

    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": "trace",
                "vlan": 100,
            }
        }
        trace_result.append(trace_step)

        result = utils.prepare_list_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": "trace",
                        "vlan": 100,
                    },
                ]
        )

    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": "trace",
                "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)


# 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": "trace",
70		"vlan": 100,
71		}
72		}
73		trace_result.append(trace_step)
74
75		result = utils.prepare_json(trace_result)
76
77		self.assertEqual(
78		result,
79		{
80		"result": [
81		{
82		"dpid": "00:00:00:00:00:00:00:01",
83		"port": 1,
84		"time": "2022-06-01 01:01:01.100000",
85		"type": "starting",
86		},
87		{
88		"dpid": "00:00:00:00:00:00:00:03",
89		"port": 3,
90		"time": "2022-06-01 01:01:01.100000",
91		"type": "trace",
92		"vlan": 100,
93		},
94		]
95		},
96		)
97
98		def test_prepare_list_json(self):
99		"""Verify prepare list with a simple tracepath result."""
100		trace_result = []
101		trace_step = {
102		"in": {
103		"dpid": "00:00:00:00:00:00:00:01",
104		"port": 1,
105		"time": "2022-06-01 01:01:01.100000",
106		"type": "starting",
107		}
108		}
109		trace_result.append(trace_step)
110
111		trace_step = {
112		"in": {
113		"dpid": "00:00:00:00:00:00:00:03",
114		"port": 3,
115		"time": "2022-06-01 01:01:01.100000",
116		"type": "trace",
117		"vlan": 100,
118		}
119		}
120		trace_result.append(trace_step)
121
122		result = utils.prepare_list_json(trace_result)
123
124		self.assertEqual(
125		result, [
126		{
127		"dpid": "00:00:00:00:00:00:00:01",
128		"port": 1,
129		"time": "2022-06-01 01:01:01.100000",
130		"type": "starting",
131		},
132		{
133		"dpid": "00:00:00:00:00:00:00:03",
134		"port": 3,
135		"time": "2022-06-01 01:01:01.100000",
136		"type": "trace",
137		"vlan": 100,
138		},
139		]
140		)
141
142		def test_prepare_json_empty(self):
143		"""Verify prepare json with empty result."""
144		trace_result = []
145
146		result = utils.prepare_json(trace_result)
147
148		self.assertEqual(result, {"result": []})
149
150		def test_format_result(self):
151		"""Verify format resul with simple tracepath result."""
152		trace_result = []
153		trace_step = {
154		"in": {
155		"dpid": "00:00:00:00:00:00:00:01",
156		"port": 1,
157		"time": "2022-06-02 02:02:02.200000",
158		"type": "starting",
159		}
160		}
161		trace_result.append(trace_step)
162
163		trace_step = {
164		"in": {
165		"dpid": "00:00:00:00:00:00:00:03",
166		"port": 3,
167		"time": "2022-06-02 02:02:02.200000",
168		"type": "trace",
169		"vlan": 100,
170		},
171		"out": {"port": 2, "vlan": 200},
172		}
173		trace_result.append(trace_step)
174
175		formatted = utils.format_result(trace_result)
176
177		self.assertEqual(
178		formatted,
179		[
180		{"dpid": "00:00:00:00:00:00:00:01", "in_port": 1},
181		{
182		"dpid": "00:00:00:00:00:00:00:03",
183		"in_port": 3,
184		"out_port": 2,
185		"out_vlan": 200,
186		"in_vlan": 100,
187		},
188		],
189		)
190
191		def test_compare_endpoints1(self):
192		"""Test for compare endpoinst for the first internal conditional."""
193		endpoint1 = {
194		"dpid": "00:00:00:00:00:00:00:01",
195		}
196		endpoint2 = {
197		"dpid": "00:00:00:00:00:00:00:02",
198		}
199
200		# Test endpoint1 dpid != endpoint2 dpid
201		result = utils._compare_endpoints(endpoint1, endpoint2)
202		self.assertFalse(result)
203
204		def test_compare_endpoints2(self):
205		"""Test for compare endpoinst for the second internal conditional."""
206		endpoint1 = {
207		"dpid": "00:00:00:00:00:00:00:03",
208		"out_port": 2,
209		"out_vlan": 200,
210		}
211		endpoint2 = {
212		"dpid": "00:00:00:00:00:00:00:03",
213		"in_port": 3,
214		"in_vlan": 100,
215		}
216
217		# Test endpoint1 without in_port
218		result = utils._compare_endpoints(endpoint1, endpoint2)
219		self.assertFalse(result)
220
221		endpoint1 = {
222		"dpid": "00:00:00:00:00:00:00:03",
223		"in_port": 3,
224		"in_vlan": 100,
225		}
226		endpoint2 = {
227		"dpid": "00:00:00:00:00:00:00:03",
228		"in_port": 3,
229		"in_vlan": 100,
230		}
231
232		# Test endpoint2 without out_port
233		result = utils._compare_endpoints(endpoint1, endpoint2)
234		self.assertFalse(result)
235
236		endpoint1 = {
237		"dpid": "00:00:00:00:00:00:00:03",
238		"in_port": 3,
239		"in_vlan": 100,
240		}
241		endpoint2 = {
242		"dpid": "00:00:00:00:00:00:00:03",
243		"out_port": 2,
244		"out_vlan": 200,
245		}
246
247		# Test endpoint1 in_port != endpoint2 out_port
248		result = utils._compare_endpoints(endpoint1, endpoint2)
249		self.assertFalse(result)
250
251		def test_compare_endpoints3(self):
252		"""Test for compare endpoinst for the third internal conditional."""
253		endpoint1 = {
254		"dpid": "00:00:00:00:00:00:00:03",
255		"in_port": 3,
256		"out_port": 2,
257		"in_vlan": 100,
258		}
259		endpoint2 = {
260		"dpid": "00:00:00:00:00:00:00:03",
261		"in_port": 2,
262		"out_port": 3,
263		"out_vlan": 200,
264		}
265
266		# Test endpoint1 in_vlan != endpoint2 out_vlan
267		result = utils._compare_endpoints(endpoint1, endpoint2)
268		self.assertFalse(result)
269
270	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...
271		"""Test for compare endpoinst for the first internal conditional."""
272		endpoint1 = {
273		"dpid": "00:00:00:00:00:00:00:03",
274		"in_port": 3,
275		"out_port": 2,
276		"in_vlan": 100,
277		}
278		endpoint2 = {
279		"dpid": "00:00:00:00:00:00:00:03",
280		"in_port": 2,
281		"out_port": 3,
282		}
283
284		# Test endpoint1 with in_vlan and endpoint2 without out_vlan
285		result = utils._compare_endpoints(endpoint1, endpoint2)
286		self.assertFalse(result)
287
288		endpoint1 = {
289		"dpid": "00:00:00:00:00:00:00:03",
290		"in_port": 3,
291		"out_port": 2,
292		}
293		endpoint2 = {
294		"dpid": "00:00:00:00:00:00:00:03",
295		"in_port": 2,
296		"out_port": 3,
297		"out_vlan": 200,
298		}
299
300		# Test endpoint1 without in_vlan and endpoint2 with out_vlan
301		result = utils._compare_endpoints(endpoint1, endpoint2)
302		self.assertFalse(result)
303
304		def test_compare_endpoints5(self):
305		"""Test for compare endpoinst for the fifth internal conditional."""
306		endpoint1 = {
307		"dpid": "00:00:00:00:00:00:00:01",
308		"in_port": 3,
309		"out_port": 2,
310		"out_vlan": 200,
311		}
312		endpoint2 = {
313		"dpid": "00:00:00:00:00:00:00:01",
314		"in_port": 2,
315		"out_port": 3,
316		"in_vlan": 100,
317		}
318
319		# Test endpoint1 out_vlan != endpoint2 in_vlan
320		result = utils._compare_endpoints(endpoint1, endpoint2)
321		self.assertFalse(result)
322
323	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...
324		"""Test for compare endpoinst for the fifth internal conditional."""
325		endpoint1 = {
326		"dpid": "00:00:00:00:00:00:00:01",
327		"in_port": 3,
328		"out_port": 2,
329		"out_vlan": 200,
330		}
331		endpoint2 = {
332		"dpid": "00:00:00:00:00:00:00:01",
333		"in_port": 2,
334		"out_port": 3,
335		}
336
337		# Test endpoint1 with out_vlan and endpoint2 without in_vlan
338		result = utils._compare_endpoints(endpoint1, endpoint2)
339		self.assertFalse(result)
340
341		endpoint1 = {
342		"dpid": "00:00:00:00:00:00:00:01",
343		"in_port": 3,
344		"out_port": 2,
345		}
346		endpoint2 = {
347		"dpid": "00:00:00:00:00:00:00:01",
348		"in_port": 2,
349		"out_port": 3,
350		"in_vlan": 100,
351		}
352
353		# Test endpoint1 without out_vlan and endpoint2 with in_vlan
354		result = utils._compare_endpoints(endpoint1, endpoint2)
355		self.assertFalse(result)
356
357		def test_compare_endpoints(self):
358		"""Test for compare endpoinst for the fifth internal conditional."""
359		endpoint1 = {
360		"dpid": "00:00:00:00:00:00:00:03",
361		"in_port": 3,
362		"in_vlan": 100,
363		}
364		endpoint2 = {
365		"dpid": "00:00:00:00:00:00:00:03",
366		"out_port": 3,
367		"out_vlan": 100,
368		}
369
370		# Test endpoint1 out_vlan != endpoint2 in_vlan
371		result = utils._compare_endpoints(endpoint1, endpoint2)
372
373		self.assertTrue(result)
374
375		def test_find_endpoint_b(self):
376		"""Test find endpoint with interface equals link endpoint B."""
377		port = 1
378
379		mock_interface = Interface("interface A", port, MagicMock())
380		mock_interface.address = "00:00:00:00:00:00:00:01"
381		mock_interface.link = get_link_mock(
382		"00:00:00:00:00:00:00:02", "00:00:00:00:00:00:00:01"
383		)
384
385		mock_switch = MagicMock()
386		mock_switch.get_interface_by_port_no.return_value = mock_interface
387
388		result = utils.find_endpoint(mock_switch, port)
389		self.assertEqual(result, mock_interface.link.endpoint_a)
390
391		def test_find_endpoint_a(self):
392		"""Test find endpoint with interface equals link endpoint A."""
393		port = 1
394
395		mock_interface = Interface("interface A", port, MagicMock())
396		mock_interface.address = "00:00:00:00:00:00:00:01"
397		mock_interface.link = get_link_mock(
398		"00:00:00:00:00:00:00:01", "00:00:00:00:00:00:00:03"
399		)
400
401		mock_switch = MagicMock()
402		mock_switch.get_interface_by_port_no.return_value = mock_interface
403
404		result = utils.find_endpoint(mock_switch, port)
405		self.assertEqual(result, mock_interface.link.endpoint_b)
406
407		def test_find_endpoint_link_none(self):
408		"""Test find endpoint without link."""
409		port = 1
410
411		mock_interface = Interface("interface A", port, MagicMock())
412		mock_interface.address = "00:00:00:00:00:00:00:01"
413
414		mock_switch = MagicMock()
415		mock_switch.get_interface_by_port_no.return_value = mock_interface
416
417		result = utils.find_endpoint(mock_switch, port)
418		self.assertIsNone(result)
419
420
421		# pylint: disable=too-many-public-methods, too-many-lines
422		class TestUtilsWithController(TestCase):
423		"""Test utils.py."""
424
425		def setUp(self):
426		# The decorator run_on_thread is patched, so methods that listen
427		# for events do not run on threads while tested.
428		# Decorators have to be patched before the methods that are
429		# decorated with them are imported.
430		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
431
432		self.controller = get_controller_mock()
433
434		self.addCleanup(patch.stopall)
435
436		def test_clean_circuits__empty(self):
437		"""Test clean circuits for empty circuits."""
438		circuits = MagicMock()
439		result = utils.clean_circuits(circuits, self.controller)
440
441		self.assertEqual(result, [])
442
443		def test_clean_circuits__no_sub(self):
444		"""Test clean circuits with just one circuit."""
445		formatted = [
446		{
447		"dpid": "00:00:00:00:00:00:00:03",
448		"in_port": 3,
449		"out_port": 2,
450		"out_vlan": 200,
451		"in_vlan": 100,
452		},
453		]
454
455		circuits = []
456		circuits.append({"circuit": formatted, "entries": []})
457
458		result = utils.clean_circuits(circuits, self.controller)
459
460		self.assertTrue(len(result) == 1)
461		self.assertEqual(formatted, result[0]["circuit"])
462
463	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...
464		"""Test clean circuits without sub-circuits."""
465		formatted_a = [
466		{
467		"dpid": "00:00:00:00:00:00:00:01",
468		},
469		{
470		"dpid": "00:00:00:00:00:00:00:03",
471		},
472		]
473		formatted_b = [
474		{
475		"dpid": "00:00:00:00:00:00:00:01",
476		},
477		{
478		"dpid": "00:00:00:00:00:00:00:02",
479		},
480		{
481		"dpid": "00:00:00:00:00:00:00:03",
482		},
483		]
484
485		circuits = []
486		circuits.append({"circuit": formatted_a, "entries": []})
487		circuits.append({"circuit": formatted_b, "entries": []})
488
489		# Test cleaning one sub-circuit
490		result = utils.clean_circuits(circuits, self.controller)
491
492		# Result must be the circuits without the sub-circuit
493		self.assertTrue(len(result) == 1)
494		self.assertEqual(formatted_b, result[0]["circuit"])
495
496	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...
497		"""Test clean circuits with one sub-circuit."""
498		formatted_a = [
499		{
500		"dpid": "00:00:00:00:00:00:00:01",
501		},
502		{
503		"dpid": "00:00:00:00:00:00:00:02",
504		},
505		]
506		formatted_b = [
507		{
508		"dpid": "00:00:00:00:00:00:00:01",
509		},
510		{
511		"dpid": "00:00:00:00:00:00:00:03",
512		},
513		{
514		"dpid": "00:00:00:00:00:00:00:04",
515		},
516		]
517
518		circuits = []
519		circuits.append({"circuit": formatted_a, "entries": []})
520		circuits.append({"circuit": formatted_b, "entries": []})
521
522		# Test circuits withou sub-circuits.
523		result = utils.clean_circuits(circuits, self.controller)
524
525		# Result must be equal to the circuits parameter
526		self.assertTrue(len(result) == 2)
527		self.assertEqual(circuits, result)
528

kytos-ng / sdntrace_cp

Pull Request — master (#47)

TestUtils.test_convert_entries_translation() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like