build.tests.unit.test_utils.TestUtils.test_get_stored_flows() - Code Metrics - Inspection of "feat: Remove automate in main" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#95)

unknown

created 2023-05-01 21:01 UTC

TestUtils.test_get_stored_flows() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	12
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	12
rs	9.9
c	0
b	0
f	0
cc	1
nop	2

"""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, settings


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

    @patch("requests.get")
    def test_get_stored_flows(self, get_mock):
        "Test get_stored_flows"
        response = MagicMock()
        response.status_code = 200
        response.json.return_value = {"result": "ok"}
        get_mock.return_value = response

        api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows/?state=installed'
        result = utils.get_stored_flows()
        get_mock.assert_called_with(api_url)
        self.assertEqual(result['result'], "ok")

    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_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
        expected = {'endpoint': mock_interface.link.endpoint_a}
        result = utils.find_endpoint(mock_switch, port)
        self.assertEqual(result, expected)

    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
        expected = {'endpoint': mock_interface.link.endpoint_b}
        result = utils.find_endpoint(mock_switch, port)
        self.assertEqual(result, expected)

    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)
        assert 'endpoint' in result
        self.assertIsNone(result['endpoint'])

    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)
        result = utils.match_field_dl_vlan(10, 11)
        self.assertFalse(result)
        result = utils.match_field_dl_vlan(3, "5/1")
        self.assertTrue(result)
        result = utils.match_field_dl_vlan(2, "5/1")
        self.assertFalse(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)


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, settings
8
9
10		# pylint: disable=too-many-public-methods, duplicate-code, protected-access
11		class TestUtils(TestCase):
12		"""Test utils.py functions."""
13
14		@patch("requests.get")
15		def test_get_stored_flows(self, get_mock):
16		"Test get_stored_flows"
17		response = MagicMock()
18		response.status_code = 200
19		response.json.return_value = {"result": "ok"}
20		get_mock.return_value = response
21
22		api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows/?state=installed'
23		result = utils.get_stored_flows()
24		get_mock.assert_called_with(api_url)
25		self.assertEqual(result['result'], "ok")
26
27		def test_convert_list_entries(self):
28		"""Verify convert entries with a list of one example"""
29		eth = {"dl_vlan": 100}
30		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
31		switch = {"switch": dpid, "eth": eth}
32		entries = {"trace": switch}
33
34		result = utils.convert_list_entries([entries])
35
36		self.assertEqual(
37		result,
38		[{
39		"dpid": "00:00:00:00:00:00:00:01",
40		"in_port": 1,
41		"dl_vlan": [100],
42		}],
43		)
44
45		def test_convert_entries_vlan(self):
46		"""Verify convert entries with simple example with vlan."""
47
48		eth = {"dl_vlan": 100}
49		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
50		switch = {"switch": dpid, "eth": eth}
51		entries = {"trace": switch}
52
53		result = utils.convert_entries(entries)
54
55		self.assertEqual(
56		result,
57		{
58		"dpid": "00:00:00:00:00:00:00:01",
59		"in_port": 1,
60		"dl_vlan": [100],
61		},
62		)
63
64		def test_prepare_json(self):
65		"""Verify prepare json with simple tracepath result."""
66		trace_result = []
67		trace_step = {
68		"in": {
69		"dpid": "00:00:00:00:00:00:00:01",
70		"port": 1,
71		"time": "2022-06-01 01:01:01.100000",
72		"type": "starting",
73		}
74		}
75		trace_result.append(trace_step)
76
77		trace_step = {
78		"in": {
79		"dpid": "00:00:00:00:00:00:00:03",
80		"port": 3,
81		"time": "2022-06-01 01:01:01.100000",
82		"type": "intermediary",
83		"vlan": 100,
84		},
85		"out": {
86		"port": 1,
87		"vlan": 123,
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": "intermediary",
109		"vlan": 100,
110		"out": {"port": 1, "vlan": 123},
111		},
112		]
113		},
114		)
115
116		def test_prepare_list_json(self):
117		"""Verify prepare list with a simple tracepath result."""
118		trace_result = []
119		trace_step = {
120		"in": {
121		"dpid": "00:00:00:00:00:00:00:01",
122		"port": 1,
123		"time": "2022-06-01 01:01:01.100000",
124		"type": "starting",
125		}
126		}
127		trace_result.append(trace_step)
128
129		trace_step = {
130		"in": {
131		"dpid": "00:00:00:00:00:00:00:03",
132		"port": 3,
133		"time": "2022-06-01 01:01:01.100000",
134		"type": "intermediary",
135		"vlan": 100,
136		},
137		"out": {
138		"port": 1,
139		"vlan": 123,
140		},
141		}
142		trace_result.append(trace_step)
143
144		result = utils._prepare_json(trace_result)
145
146		self.assertEqual(
147		result, [
148		{
149		"dpid": "00:00:00:00:00:00:00:01",
150		"port": 1,
151		"time": "2022-06-01 01:01:01.100000",
152		"type": "starting",
153		},
154		{
155		"dpid": "00:00:00:00:00:00:00:03",
156		"port": 3,
157		"time": "2022-06-01 01:01:01.100000",
158		"type": "intermediary",
159		"vlan": 100,
160		"out": {"port": 1, "vlan": 123},
161		},
162		]
163		)
164
165		def test_prepare_json_empty(self):
166		"""Verify prepare json with empty result."""
167		trace_result = []
168
169		result = utils.prepare_json(trace_result)
170
171		self.assertEqual(result, {"result": []})
172
173		def test_compare_endpoints1(self):
174		"""Test for compare endpoinst for the first internal conditional."""
175		endpoint1 = {
176		"dpid": "00:00:00:00:00:00:00:01",
177		}
178		endpoint2 = {
179		"dpid": "00:00:00:00:00:00:00:02",
180		}
181
182		# Test endpoint1 dpid != endpoint2 dpid
183		result = utils._compare_endpoints(endpoint1, endpoint2)
184		self.assertFalse(result)
185
186		def test_compare_endpoints2(self):
187		"""Test for compare endpoinst for the second internal conditional."""
188		endpoint1 = {
189		"dpid": "00:00:00:00:00:00:00:03",
190		"out_port": 2,
191		"out_vlan": 200,
192		}
193		endpoint2 = {
194		"dpid": "00:00:00:00:00:00:00:03",
195		"in_port": 3,
196		"in_vlan": 100,
197		}
198
199		# Test endpoint1 without in_port
200		result = utils._compare_endpoints(endpoint1, endpoint2)
201		self.assertFalse(result)
202
203		endpoint1 = {
204		"dpid": "00:00:00:00:00:00:00:03",
205		"in_port": 3,
206		"in_vlan": 100,
207		}
208		endpoint2 = {
209		"dpid": "00:00:00:00:00:00:00:03",
210		"in_port": 3,
211		"in_vlan": 100,
212		}
213
214		# Test endpoint2 without out_port
215		result = utils._compare_endpoints(endpoint1, endpoint2)
216		self.assertFalse(result)
217
218		endpoint1 = {
219		"dpid": "00:00:00:00:00:00:00:03",
220		"in_port": 3,
221		"in_vlan": 100,
222		}
223		endpoint2 = {
224		"dpid": "00:00:00:00:00:00:00:03",
225		"out_port": 2,
226		"out_vlan": 200,
227		}
228
229		# Test endpoint1 in_port != endpoint2 out_port
230		result = utils._compare_endpoints(endpoint1, endpoint2)
231		self.assertFalse(result)
232
233		def test_compare_endpoints3(self):
234		"""Test for compare endpoinst for the third internal conditional."""
235		endpoint1 = {
236		"dpid": "00:00:00:00:00:00:00:03",
237		"in_port": 3,
238		"out_port": 2,
239		"in_vlan": 100,
240		}
241		endpoint2 = {
242		"dpid": "00:00:00:00:00:00:00:03",
243		"in_port": 2,
244		"out_port": 3,
245		"out_vlan": 200,
246		}
247
248		# Test endpoint1 in_vlan != endpoint2 out_vlan
249		result = utils._compare_endpoints(endpoint1, endpoint2)
250		self.assertFalse(result)
251
252	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...
253		"""Test for compare endpoinst for the first internal conditional."""
254		endpoint1 = {
255		"dpid": "00:00:00:00:00:00:00:03",
256		"in_port": 3,
257		"out_port": 2,
258		"in_vlan": 100,
259		}
260		endpoint2 = {
261		"dpid": "00:00:00:00:00:00:00:03",
262		"in_port": 2,
263		"out_port": 3,
264		}
265
266		# Test endpoint1 with in_vlan and endpoint2 without out_vlan
267		result = utils._compare_endpoints(endpoint1, endpoint2)
268		self.assertFalse(result)
269
270		endpoint1 = {
271		"dpid": "00:00:00:00:00:00:00:03",
272		"in_port": 3,
273		"out_port": 2,
274		}
275		endpoint2 = {
276		"dpid": "00:00:00:00:00:00:00:03",
277		"in_port": 2,
278		"out_port": 3,
279		"out_vlan": 200,
280		}
281
282		# Test endpoint1 without in_vlan and endpoint2 with out_vlan
283		result = utils._compare_endpoints(endpoint1, endpoint2)
284		self.assertFalse(result)
285
286		def test_compare_endpoints5(self):
287		"""Test for compare endpoinst for the fifth internal conditional."""
288		endpoint1 = {
289		"dpid": "00:00:00:00:00:00:00:01",
290		"in_port": 3,
291		"out_port": 2,
292		"out_vlan": 200,
293		}
294		endpoint2 = {
295		"dpid": "00:00:00:00:00:00:00:01",
296		"in_port": 2,
297		"out_port": 3,
298		"in_vlan": 100,
299		}
300
301		# Test endpoint1 out_vlan != endpoint2 in_vlan
302		result = utils._compare_endpoints(endpoint1, endpoint2)
303		self.assertFalse(result)
304
305	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...
306		"""Test for compare endpoinst for the fifth internal conditional."""
307		endpoint1 = {
308		"dpid": "00:00:00:00:00:00:00:01",
309		"in_port": 3,
310		"out_port": 2,
311		"out_vlan": 200,
312		}
313		endpoint2 = {
314		"dpid": "00:00:00:00:00:00:00:01",
315		"in_port": 2,
316		"out_port": 3,
317		}
318
319		# Test endpoint1 with out_vlan and endpoint2 without in_vlan
320		result = utils._compare_endpoints(endpoint1, endpoint2)
321		self.assertFalse(result)
322
323		endpoint1 = {
324		"dpid": "00:00:00:00:00:00:00:01",
325		"in_port": 3,
326		"out_port": 2,
327		}
328		endpoint2 = {
329		"dpid": "00:00:00:00:00:00:00:01",
330		"in_port": 2,
331		"out_port": 3,
332		"in_vlan": 100,
333		}
334
335		# Test endpoint1 without out_vlan and endpoint2 with in_vlan
336		result = utils._compare_endpoints(endpoint1, endpoint2)
337		self.assertFalse(result)
338
339		def test_compare_endpoints(self):
340		"""Test for compare endpoinst for the fifth internal conditional."""
341		endpoint1 = {
342		"dpid": "00:00:00:00:00:00:00:03",
343		"in_port": 3,
344		"in_vlan": 100,
345		}
346		endpoint2 = {
347		"dpid": "00:00:00:00:00:00:00:03",
348		"out_port": 3,
349		"out_vlan": 100,
350		}
351
352		# Test endpoint1 out_vlan != endpoint2 in_vlan
353		result = utils._compare_endpoints(endpoint1, endpoint2)
354
355		self.assertTrue(result)
356
357		def test_find_endpoint_b(self):
358		"""Test find endpoint with interface equals link endpoint B."""
359		port = 1
360
361		mock_interface = Interface("interface A", port, MagicMock())
362		mock_interface.address = "00:00:00:00:00:00:00:01"
363		mock_interface.link = get_link_mock(
364		"00:00:00:00:00:00:00:02", "00:00:00:00:00:00:00:01"
365		)
366
367		mock_switch = MagicMock()
368		mock_switch.get_interface_by_port_no.return_value = mock_interface
369		expected = {'endpoint': mock_interface.link.endpoint_a}
370		result = utils.find_endpoint(mock_switch, port)
371		self.assertEqual(result, expected)
372
373		def test_find_endpoint_a(self):
374		"""Test find endpoint with interface equals link endpoint A."""
375		port = 1
376
377		mock_interface = Interface("interface A", port, MagicMock())
378		mock_interface.address = "00:00:00:00:00:00:00:01"
379		mock_interface.link = get_link_mock(
380		"00:00:00:00:00:00:00:01", "00:00:00:00:00:00:00:03"
381		)
382
383		mock_switch = MagicMock()
384		mock_switch.get_interface_by_port_no.return_value = mock_interface
385		expected = {'endpoint': mock_interface.link.endpoint_b}
386		result = utils.find_endpoint(mock_switch, port)
387		self.assertEqual(result, expected)
388
389		def test_find_endpoint_link_none(self):
390		"""Test find endpoint without link."""
391		port = 1
392
393		mock_interface = Interface("interface A", port, MagicMock())
394		mock_interface.address = "00:00:00:00:00:00:00:01"
395
396		mock_switch = MagicMock()
397		mock_switch.get_interface_by_port_no.return_value = mock_interface
398
399		result = utils.find_endpoint(mock_switch, port)
400		assert 'endpoint' in result
401		self.assertIsNone(result['endpoint'])
402
403		def test_convert_vlan(self):
404		"""Test convert_vlan function"""
405		value = 100
406		result = utils.convert_vlan(value)
407		assert result[0] == 100
408
409		value = "4096/4096"
410		result = utils.convert_vlan(value)
411		assert result[0] == 4096
412		assert result[1] == 4096
413
414		def test_match_field_dl_vlan(self):
415		"""Test match_field_dl_vlan"""
416
417		result = utils.match_field_dl_vlan(None, 0)
418		self.assertTrue(result)
419		result = utils.match_field_dl_vlan(None, 10)
420		self.assertFalse(result)
421		result = utils.match_field_dl_vlan(None, "4096/4096")
422		self.assertFalse(result)
423		result = utils.match_field_dl_vlan(10, 0)
424		self.assertFalse(result)
425		result = utils.match_field_dl_vlan(10, 10)
426		self.assertTrue(result)
427		result = utils.match_field_dl_vlan(10, "4096/4096")
428		self.assertTrue(result)
429		result = utils.match_field_dl_vlan(10, 11)
430		self.assertFalse(result)
431		result = utils.match_field_dl_vlan(3, "5/1")
432		self.assertTrue(result)
433		result = utils.match_field_dl_vlan(2, "5/1")
434		self.assertFalse(result)
435
436
437		# pylint: disable=too-many-public-methods, too-many-lines
438		class TestUtilsWithController(TestCase):
439		"""Test utils.py."""
440
441		def setUp(self):
442		# The decorator run_on_thread is patched, so methods that listen
443		# for events do not run on threads while tested.
444		# Decorators have to be patched before the methods that are
445		# decorated with them are imported.
446		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
447
448		self.controller = get_controller_mock()
449
450		self.addCleanup(patch.stopall)
451

kytos-ng / sdntrace_cp

Pull Request — master (#95)

TestUtils.test_get_stored_flows() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like