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-03 17:08 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, timeout=20)
        assert 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])
        expected = [{
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 1,
                "dl_vlan": [100],
            }]
        assert result == expected

    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)
        expected = {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 1,
                "dl_vlan": [100],
            }
        assert result == expected

    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)
        expected = {
                "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},
                    },
                ]
            }
        assert result == expected

    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)
        expected = [
                    {
                        "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},
                    },
                ]
        assert result == expected

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

        result = utils.prepare_json(trace_result)

        assert 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)
        assert result is False

    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)
        assert result is False

        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)
        assert result is False

        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)
        assert result is False

    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)
        assert result is False

    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)
        assert result is False

        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)
        assert result is False

    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)
        assert result is False

    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)
        assert result is False

        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)
        assert result is False

    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)
        assert result is True

    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)
        assert 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)
        assert 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
        assert result['endpoint'] is None

    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)
        assert result is True
        result = utils.match_field_dl_vlan(None, 10)
        assert result is False
        result = utils.match_field_dl_vlan(None, "4096/4096")
        assert result is False
        result = utils.match_field_dl_vlan(10, 0)
        assert result is False
        result = utils.match_field_dl_vlan(10, 10)
        assert result is True
        result = utils.match_field_dl_vlan(10, "4096/4096")
        assert result is True
        result = utils.match_field_dl_vlan(10, 11)
        assert result is False
        result = utils.match_field_dl_vlan(3, "5/1")
        assert result is True
        result = utils.match_field_dl_vlan(2, "5/1")
        assert result is False


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

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