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-04-28 19:14 UTC

TestUtils.test_get_stored_flows() A

↳ Parent: build.tests.unit.test_utils

Complexity

Conditions

Size

Total Lines	15
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	12
nop	2
dl	0
loc	15
rs	9.8
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, 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=30
                                )
        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_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": "intermediary",
                "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
        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)

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

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