build.tests.unit.test_utils.TestUtils.test_match_field_ip() - Code Metrics - Inspection of "Merge pull request #96 from kytos-ng/fix/do_match" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ad70c0...a80f24 )

unknown

created 2023-05-10 20:31 UTC

TestUtils.test_match_field_ip() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	45
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	23
CRAP Score	1

Importance

Changes

Metric	Value
eloc	35
dl	0
loc	45
ccs	23
cts	23
cp	1
rs	9.0399
c	0
b	0
f	0
cc	1
nop	1
crap	1

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

    def test_match_field_ip(self):
        """Test match_field_ip"""
        # IPv4 cases
        result = utils.match_field_ip('192.168.20.21', '192.168.20.21')
        assert result
        result = utils.match_field_ip('192.168.20.21', '192.168.20.21/10')
        assert result
        result = utils.match_field_ip('192.168.20.21', '192.168.20.21/32')
        assert result
        result = utils.match_field_ip(
                                        '192.168.20.21',
                                        '192.168.20.21/255.255.255.255'
                                    )
        assert result
        result = utils.match_field_ip('192.168.20.30', '192.168.20.21')
        assert not result
        result = utils.match_field_ip('192.200.20.30', '192.168.20.21/10')
        assert not result

        # IPv6 cases
        result = utils.match_field_ip(
                                        '2002:db8::8a3f:362:7897',
                                        '2002:db8::8a3f:362:7897'
                                    )
        assert result
        result = utils.match_field_ip(
                                        '2002:db8::8a3f:362:7897',
                                        '2002:db8::8a3f:362:7897/10'
                                    )
        assert result
        result = utils.match_field_ip(
                                        '2002:db8::8a3f:362:7897',
                                        '2002:db8::8a3f:362:7897/128'
                                    )
        assert result
        result = utils.match_field_ip(
                                        '2002:db8::8a3f:362:7',
                                        '2002:db8::8a3f:362:7897'
                                    )
        assert not result
        result = utils.match_field_ip(
                                        '3002:db8::9a3f:362:7897',
                                        '2002:db8::8a3f:362:7897/10'
                                    )
        assert not 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	1		from unittest import TestCase
3	1		from unittest.mock import patch, MagicMock
4
5	1		from kytos.core.interface import Interface
6	1		from kytos.lib.helpers import get_controller_mock, get_link_mock
7	1		from napps.amlight.sdntrace_cp import utils, settings
8
9
10			# pylint: disable=too-many-public-methods, duplicate-code, protected-access
11	1		class TestUtils(TestCase):
12			"""Test utils.py functions."""
13
14	1		@patch("requests.get")
15	1		def test_get_stored_flows(self, get_mock):
16			"Test get_stored_flows"
17	1		response = MagicMock()
18	1		response.status_code = 200
19	1		response.json.return_value = {"result": "ok"}
20	1		get_mock.return_value = response
21
22	1		api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows/?state=installed'
23	1		result = utils.get_stored_flows()
24	1		get_mock.assert_called_with(api_url, timeout=20)
25	1		assert result['result'] == "ok"
26
27	1		def test_convert_list_entries(self):
28			"""Verify convert entries with a list of one example"""
29	1		eth = {"dl_vlan": 100}
30	1		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
31	1		switch = {"switch": dpid, "eth": eth}
32	1		entries = {"trace": switch}
33
34	1		result = utils.convert_list_entries([entries])
35	1		expected = [{
36			"dpid": "00:00:00:00:00:00:00:01",
37			"in_port": 1,
38			"dl_vlan": [100],
39			}]
40	1		assert result == expected
41
42	1		def test_convert_entries_vlan(self):
43			"""Verify convert entries with simple example with vlan."""
44
45	1		eth = {"dl_vlan": 100}
46	1		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
47	1		switch = {"switch": dpid, "eth": eth}
48	1		entries = {"trace": switch}
49
50	1		result = utils.convert_entries(entries)
51	1		expected = {
52			"dpid": "00:00:00:00:00:00:00:01",
53			"in_port": 1,
54			"dl_vlan": [100],
55			}
56	1		assert result == expected
57
58	1		def test_prepare_json(self):
59			"""Verify prepare json with simple tracepath result."""
60	1		trace_result = []
61	1		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	1		trace_result.append(trace_step)
70
71	1		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	1		trace_result.append(trace_step)
85
86	1		result = utils.prepare_json(trace_result)
87	1		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	1		assert result == expected
106
107	1		def test_prepare_list_json(self):
108			"""Verify prepare list with a simple tracepath result."""
109	1		trace_result = []
110	1		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	1		trace_result.append(trace_step)
119
120	1		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	1		trace_result.append(trace_step)
134
135	1		result = utils._prepare_json(trace_result)
136	1		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	1		assert result == expected
153
154	1		def test_prepare_json_empty(self):
155			"""Verify prepare json with empty result."""
156	1		trace_result = []
157
158	1		result = utils.prepare_json(trace_result)
159
160	1		assert result == {"result": []}
161
162	1		def test_compare_endpoints1(self):
163			"""Test for compare endpoinst for the first internal conditional."""
164	1		endpoint1 = {
165			"dpid": "00:00:00:00:00:00:00:01",
166			}
167	1		endpoint2 = {
168			"dpid": "00:00:00:00:00:00:00:02",
169			}
170
171			# Test endpoint1 dpid != endpoint2 dpid
172	1		result = utils._compare_endpoints(endpoint1, endpoint2)
173	1		assert not result
174
175	1		def test_compare_endpoints2(self):
176			"""Test for compare endpoinst for the second internal conditional."""
177	1		endpoint1 = {
178			"dpid": "00:00:00:00:00:00:00:03",
179			"out_port": 2,
180			"out_vlan": 200,
181			}
182	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
190	1		assert not result
191
192	1		endpoint1 = {
193			"dpid": "00:00:00:00:00:00:00:03",
194			"in_port": 3,
195			"in_vlan": 100,
196			}
197	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
205	1		assert not result
206
207	1		endpoint1 = {
208			"dpid": "00:00:00:00:00:00:00:03",
209			"in_port": 3,
210			"in_vlan": 100,
211			}
212	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
220	1		assert not result
221
222	1		def test_compare_endpoints3(self):
223			"""Test for compare endpoinst for the third internal conditional."""
224	1		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	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
239	1		assert not result
240
241	1	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	1		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	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
257	1		assert not result
258
259	1		endpoint1 = {
260			"dpid": "00:00:00:00:00:00:00:03",
261			"in_port": 3,
262			"out_port": 2,
263			}
264	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
273	1		assert not result
274
275	1		def test_compare_endpoints5(self):
276			"""Test for compare endpoinst for the fifth internal conditional."""
277	1		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	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
292	1		assert not result
293
294	1	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	1		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	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
310	1		assert not result
311
312	1		endpoint1 = {
313			"dpid": "00:00:00:00:00:00:00:01",
314			"in_port": 3,
315			"out_port": 2,
316			}
317	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
326	1		assert not result
327
328	1		def test_compare_endpoints(self):
329			"""Test for compare endpoinst for the fifth internal conditional."""
330	1		endpoint1 = {
331			"dpid": "00:00:00:00:00:00:00:03",
332			"in_port": 3,
333			"in_vlan": 100,
334			}
335	1		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	1		result = utils._compare_endpoints(endpoint1, endpoint2)
343	1		assert result
344
345	1		def test_find_endpoint_b(self):
346			"""Test find endpoint with interface equals link endpoint B."""
347	1		port = 1
348
349	1		mock_interface = Interface("interface A", port, MagicMock())
350	1		mock_interface.address = "00:00:00:00:00:00:00:01"
351	1		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	1		mock_switch = MagicMock()
356	1		mock_switch.get_interface_by_port_no.return_value = mock_interface
357	1		expected = {'endpoint': mock_interface.link.endpoint_a}
358	1		result = utils.find_endpoint(mock_switch, port)
359	1		assert result == expected
360
361	1		def test_find_endpoint_a(self):
362			"""Test find endpoint with interface equals link endpoint A."""
363	1		port = 1
364
365	1		mock_interface = Interface("interface A", port, MagicMock())
366	1		mock_interface.address = "00:00:00:00:00:00:00:01"
367	1		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	1		mock_switch = MagicMock()
372	1		mock_switch.get_interface_by_port_no.return_value = mock_interface
373	1		expected = {'endpoint': mock_interface.link.endpoint_b}
374	1		result = utils.find_endpoint(mock_switch, port)
375	1		assert result == expected
376
377	1		def test_find_endpoint_link_none(self):
378			"""Test find endpoint without link."""
379	1		port = 1
380
381	1		mock_interface = Interface("interface A", port, MagicMock())
382	1		mock_interface.address = "00:00:00:00:00:00:00:01"
383
384	1		mock_switch = MagicMock()
385	1		mock_switch.get_interface_by_port_no.return_value = mock_interface
386
387	1		result = utils.find_endpoint(mock_switch, port)
388	1		assert 'endpoint' in result
389	1		assert result['endpoint'] is None
390
391	1		def test_convert_vlan(self):
392			"""Test convert_vlan function"""
393	1		value = 100
394	1		result = utils.convert_vlan(value)
395	1		assert result[0] == 100
396
397	1		value = "4096/4096"
398	1		result = utils.convert_vlan(value)
399	1		assert result[0] == 4096
400	1		assert result[1] == 4096
401
402	1		def test_match_field_dl_vlan(self):
403			"""Test match_field_dl_vlan"""
404
405	1		result = utils.match_field_dl_vlan(None, 0)
406	1		assert result
407	1		result = utils.match_field_dl_vlan(None, 10)
408	1		assert not result
409	1		result = utils.match_field_dl_vlan(None, "4096/4096")
410	1		assert not result
411	1		result = utils.match_field_dl_vlan([10], 0)
412	1		assert not result
413	1		result = utils.match_field_dl_vlan([10], 10)
414	1		assert result
415	1		result = utils.match_field_dl_vlan([10], "4096/4096")
416	1		assert result
417	1		result = utils.match_field_dl_vlan([10], 11)
418	1		assert not result
419	1		result = utils.match_field_dl_vlan([3], "5/1")
420	1		assert result
421	1		result = utils.match_field_dl_vlan([2], "5/1")
422	1		assert not result
423
424	1		def test_match_field_ip(self):
425			"""Test match_field_ip"""
426			# IPv4 cases
427	1		result = utils.match_field_ip('192.168.20.21', '192.168.20.21')
428	1		assert result
429	1		result = utils.match_field_ip('192.168.20.21', '192.168.20.21/10')
430	1		assert result
431	1		result = utils.match_field_ip('192.168.20.21', '192.168.20.21/32')
432	1		assert result
433	1		result = utils.match_field_ip(
434			'192.168.20.21',
435			'192.168.20.21/255.255.255.255'
436			)
437	1		assert result
438	1		result = utils.match_field_ip('192.168.20.30', '192.168.20.21')
439	1		assert not result
440	1		result = utils.match_field_ip('192.200.20.30', '192.168.20.21/10')
441	1		assert not result
442
443			# IPv6 cases
444	1		result = utils.match_field_ip(
445			'2002:db8::8a3f:362:7897',
446			'2002:db8::8a3f:362:7897'
447			)
448	1		assert result
449	1		result = utils.match_field_ip(
450			'2002:db8::8a3f:362:7897',
451			'2002:db8::8a3f:362:7897/10'
452			)
453	1		assert result
454	1		result = utils.match_field_ip(
455			'2002:db8::8a3f:362:7897',
456			'2002:db8::8a3f:362:7897/128'
457			)
458	1		assert result
459	1		result = utils.match_field_ip(
460			'2002:db8::8a3f:362:7',
461			'2002:db8::8a3f:362:7897'
462			)
463	1		assert not result
464	1		result = utils.match_field_ip(
465			'3002:db8::9a3f:362:7897',
466			'2002:db8::8a3f:362:7897/10'
467			)
468	1		assert not result
469
470
471			# pylint: disable=too-many-public-methods, too-many-lines
472	1		class TestUtilsWithController(TestCase):
473			"""Test utils.py."""
474
475	1		def setUp(self):
476			# The decorator run_on_thread is patched, so methods that listen
477			# for events do not run on threads while tested.
478			# Decorators have to be patched before the methods that are
479			# decorated with them are imported.
480			patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
481
482			self.controller = get_controller_mock()
483
484			self.addCleanup(patch.stopall)
485

kytos-ng / sdntrace_cp

Push — master ( ad70c0...a80f24 )

TestUtils.test_match_field_ip() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like