build.tests.unit.test_utils.TestUtils.test_prepare_json() - Code Metrics - Inspection of "Improve coverage" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#30)

by Rogerio

created 2022-07-12 02:45 UTC

TestUtils.test_prepare_json() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	42
Code Lines	31

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	31
nop	1
dl	0
loc	42
rs	9.1359
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_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,
                "vlan_vid": [100],
            },
        )

    def test_convert_entries_translation(self):
        """Verify convert entries with all translations."""

        eth = {
            "dl_src": "A",
            "dl_dst": "B",
            "dl_type": "C",
            "dl_vlan": "D",
            "nw_src": "E",
            "nw_dst": "F",
            "nw_tos": "G",
            "nw_proto": "H",
        }
        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,
                "eth_src": "A",
                "eth_dst": "B",
                "eth_type": "C",
                "vlan_vid": ["D"],
                "ip4_src": "E",
                "ip4_dst": "F",
                "ip_tos": "G",
                "ip_proto": "H",
            },
        )

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

kytos-ng / sdntrace_cp

Pull Request — master (#30)

TestUtils.test_prepare_json() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like