build.tests.unit.test_main.TestMain.test_traces() - Code Metrics - Inspection of "Add PUT traces endpoint and functionality match_an..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#47)

unknown

created 2022-10-17 21:40 UTC

build.tests.unit.test_main.TestMain.test_traces() B

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	55
Code Lines	37

Duplication

Lines	55
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	37
nop	2
dl	55
loc	55
rs	8.9919
c	0
b	0
f	0

How to fix Long Method

"""Module to test the main napp file."""
import json
from unittest import TestCase
from unittest.mock import patch, MagicMock

from kytos.core.interface import Interface
from kytos.lib.helpers import (
    get_interface_mock,
    get_switch_mock,
    get_controller_mock,
    get_link_mock,
    get_test_client,
)


# pylint: disable=too-many-public-methods, too-many-lines
class TestMain(TestCase):
    """Test the Main class."""

    def setUp(self):
        """Execute steps before each tests.

        Set the server_name_url_url from amlight/sdntrace_cp
        """
        self.server_name_url = "http://localhost:8181/api/amlight/sdntrace_cp"

        # 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()
        # pylint: disable=import-outside-toplevel
        from napps.amlight.sdntrace_cp.main import Main

        self.napp = Main(get_controller_mock())

    @staticmethod
    def get_napp_urls(napp):
        """Return the amlight/sdntrace_cp urls.

        The urls will be like:

        urls = [
            (options, methods, url)
        ]

        """
        controller = napp.controller
        controller.api_server.register_napp_endpoints(napp)

        urls = []
        for rule in controller.api_server.app.url_map.iter_rules():
            options = {}
            for arg in rule.arguments:
                options[arg] = f"[{0}]".format(arg)

            if f"{napp.username}/{napp.name}" in str(rule):
                urls.append((options, rule.methods, f"{str(rule)}"))

        return urls

    def test_verify_api_urls(self):
        """Verify all APIs registered."""

        expected_urls = [
            (
                {},
                {"OPTIONS", "HEAD", "PUT"},
                "/api/amlight/sdntrace_cp/trace/ ",
            ),
            (
                {},
                {"OPTIONS", "HEAD", "PUT"},
                "/api/amlight/sdntrace_cp/traces/ ",
            ),
        ]
        urls = self.get_napp_urls(self.napp)
        self.assertEqual(len(expected_urls), len(urls))

    @patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
    def test_trace_step(self, mock_flow_match):
        """Test trace_step success result."""
        mock_flow_match.return_value = ["1"], ["entries"], 1
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)

        mock_interface = Interface("interface A", 1, MagicMock())
        mock_interface.address = "00:00:00:00:00:00:00:01"

        iface1 = get_interface_mock(
            "", 1, get_switch_mock("00:00:00:00:00:00:00:01")
        )
        iface2 = get_interface_mock(
            "", 2, get_switch_mock("00:00:00:00:00:00:00:02")
        )
        mock_interface.link = get_link_mock(iface1, iface2)
        mock_interface.link.endpoint_a.port_number = 1
        mock_interface.link.endpoint_a.port_number = 2

        # Patch for utils.find_endpoint
        switch.get_interface_by_port_no.return_value = mock_interface

        entries = MagicMock()

        result = self.napp.trace_step(switch, entries)

        mock_flow_match.assert_called_once()
        self.assertEqual(
            result,
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 2,
                "out_port": 1,
                "entries": ["entries"],
            },
        )

    @patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
    def test_trace_step__no_endpoint(self, mock_flow_match):
        """Test trace_step without endpoints available for switch/port."""
        mock_flow_match.return_value = ["1"], ["entries"], 1
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)

        mock_interface = Interface("interface A", 1, MagicMock())
        mock_interface.address = "00:00:00:00:00:00:00:01"
        mock_interface.link = None

        # Patch for utils.find_endpoint
        switch.get_interface_by_port_no.return_value = mock_interface

        entries = MagicMock()

        result = self.napp.trace_step(switch, entries)

        mock_flow_match.assert_called_once()
        self.assertEqual(result, {"entries": ["entries"], "out_port": 1})

    @patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
    def test_trace_step__no_flow(self, mock_stored_flows):
        """Test trace_step without flows for the switch."""
        switch = get_switch_mock("00:00:00:00:00:00:00:01")
        entries = MagicMock()

        stored_flows = {
            "flow": {
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
            },
            "flow_id": 1,
            "state": "installed",
            "switch": "00:00:00:00:00:00:00:01",
        }

        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }

        result = self.napp.trace_step(switch, entries)
        assert result is None

    @patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
    def test_tracepath(self, mock_trace_step):
        """Test tracepath with success result."""
        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}
        mock_trace_step.return_value = {
            "dpid": "00:00:00:00:00:00:00:02",
            "in_port": 2,
            "out_port": 3,
            "entries": entries,
        }

        result = self.napp.tracepath(entries["trace"]["switch"])

        assert result[0]["in"]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0]["in"]["port"] == 1
        assert result[0]["in"]["type"] == "starting"
        assert result[0]["out"]["port"] == 3

        assert result[1]["in"]["dpid"] == "00:00:00:00:00:00:00:02"
        assert result[1]["in"]["port"] == 2
        assert result[1]["in"]["type"] == "trace"
        assert result[1]["out"]["port"] == 3

    def test_has_loop(self):
        """Test has_loop to detect a tracepath with loop."""
        trace_result = [
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "port": 2,
                },
                "out": {
                    "port": 1,
                },
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "port": 2,
                },
                "out": {
                    "port": 1,
                },
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "port": 3,
                },
                "out": {
                    "port": 1,
                },
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "port": 3,
                },
                "out": {
                    "port": 1,
                },
            },
        ]
        trace_step = {
            "dpid": "00:00:00:00:00:00:00:03",
            "port": 3,
        }

        result = self.napp.has_loop(trace_step, trace_result)

        self.assertTrue(result)

    def test_has_loop__fail(self):
        """Test has_loop to detect a tracepath with loop."""
        trace_result = [
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "port": 2,
                },
                "out": {
                    "port": 1,
                },
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:02",
                    "port": 2,
                },
                "out": {
                    "port": 1,
                },
            },
        ]
        trace_step = {
            "dpid": "00:00:00:00:00:00:00:03",
            "port": 2,
        }

        result = self.napp.has_loop(trace_step, trace_result)

        self.assertFalse(result)

    @patch("napps.amlight.sdntrace_cp.main.settings")
    def test_update_circuits(self, mock_settings):
        """Test update_circuits event listener with success."""
        mock_settings.FIND_CIRCUITS_IN_FLOWS = True

        self.napp.automate = MagicMock()
        self.napp.automate.find_circuits = MagicMock()

        event = MagicMock()
        self.napp.update_circuits(event)

        self.napp.automate.find_circuits.assert_called_once()

    @patch("napps.amlight.sdntrace_cp.main.settings")
    def test_update_circuits__no_settings(self, mock_settings):
        """Test update_circuits event listener without
        settings option enabled."""
        mock_settings.FIND_CIRCUITS_IN_FLOWS = False

        self.napp.automate = MagicMock()
        self.napp.automate.find_circuits = MagicMock()

        event = MagicMock()
        self.napp.update_circuits(event)

        self.napp.automate.find_circuits.assert_not_called()

    @patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
    def test_trace(self, mock_stored_flows):
        """Test trace rest call."""
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/trace/"

        payload = {
            "trace": {
                "switch": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1
                    },
                "eth": {"dl_vlan": 100},
            }
        }
        stored_flows = {
            "00:00:00:00:00:00:00:01": {
                "flow": {
                    "table_id": 0,
                    "cookie": 84114964,
                    "hard_timeout": 0,
                    "idle_timeout": 0,
                    "priority": 10,
                },
                "flow_id": 1,
                "state": "installed",
                "switch": "00:00:00:00:00:00:00:01",
            }
        }
        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["result"]

        assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0]["port"] == 1
        assert result[0]["type"] == "starting"
        assert result[0]["vlan"] == 100

    @patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")

    def test_traces(self, mock_stored_flows):
        """Test traces rest call for two traces with different switches."""
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/traces/"

        payload = [
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "in_port": 1
                        },
                    "eth": {"dl_vlan": 100},
                }
            },
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:02",
                        "in_port": 1},
                    "eth": {"dl_vlan": 100},
                }
            }
        ]

        stored_flow = {
                "id": 1,
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
        }

        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result1 = current_data["00:00:00:00:00:00:00:01"]
        result2 = current_data["00:00:00:00:00:00:00:02"]

        assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result1[0][0]["port"] == 1
        assert result1[0][0]["type"] == "starting"
        assert result1[0][0]["vlan"] == 100

        assert result2[0][0]["dpid"] == "00:00:00:00:00:00:00:02"
        assert result2[0][0]["port"] == 1
        assert result2[0][0]["type"] == "starting"
        assert result2[0][0]["vlan"] == 100

    @patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")

    def test_traces_same_switch(self, mock_stored_flows):
        """Test traces rest call for two traces with samw switches."""
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/traces/"

        payload = [
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "in_port": 1
                    },
                    "eth": {"dl_vlan": 100},
                }
            },
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "in_port": 2
                    },
                    "eth": {"dl_vlan": 100},
                }
            }
        ]

        stored_flow = {
                "id": 1,
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
        }

        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["00:00:00:00:00:00:00:01"]

        assert len(current_data) == 1

        assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0][0]["port"] == 1
        assert result[0][0]["type"] == "starting"
        assert result[0][0]["vlan"] == 100

        assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[1][0]["port"] == 2
        assert result[1][0]["type"] == "starting"
        assert result[1][0]["vlan"] == 100

    @patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
    def test_traces_twice(self, mock_stored_flows):
        """Test traces rest call for two equal traces."""
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/traces/"

        payload = [
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "in_port": 1
                        },
                    "eth": {"dl_vlan": 100},
                }
            },
            {
                "trace": {
                    "switch": {
                        "dpid": "00:00:00:00:00:00:00:01",
                        "in_port": 1
                        },
                    "eth": {"dl_vlan": 100},
                }
            }
        ]
        stored_flow = {
                "id": 1,
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
        }

        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["00:00:00:00:00:00:00:01"]

        assert len(current_data) == 1
        assert len(result) == 1

        assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0][0]["port"] == 1
        assert result[0][0]["type"] == "starting"
        assert result[0][0]["vlan"] == 100


1		"""Module to test the main napp file."""
2		import json
3		from unittest import TestCase
4		from unittest.mock import patch, MagicMock
5
6		from kytos.core.interface import Interface
7		from kytos.lib.helpers import (
8		get_interface_mock,
9		get_switch_mock,
10		get_controller_mock,
11		get_link_mock,
12		get_test_client,
13		)
14
15
16		# pylint: disable=too-many-public-methods, too-many-lines
17		class TestMain(TestCase):
18		"""Test the Main class."""
19
20		def setUp(self):
21		"""Execute steps before each tests.
22
23		Set the server_name_url_url from amlight/sdntrace_cp
24		"""
25		self.server_name_url = "http://localhost:8181/api/amlight/sdntrace_cp"
26
27		# The decorator run_on_thread is patched, so methods that listen
28		# for events do not run on threads while tested.
29		# Decorators have to be patched before the methods that are
30		# decorated with them are imported.
31		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
32		# pylint: disable=import-outside-toplevel
33		from napps.amlight.sdntrace_cp.main import Main
34
35		self.napp = Main(get_controller_mock())
36
37		@staticmethod
38		def get_napp_urls(napp):
39		"""Return the amlight/sdntrace_cp urls.
40
41		The urls will be like:
42
43		urls = [
44		(options, methods, url)
45		]
46
47		"""
48		controller = napp.controller
49		controller.api_server.register_napp_endpoints(napp)
50
51		urls = []
52		for rule in controller.api_server.app.url_map.iter_rules():
53		options = {}
54		for arg in rule.arguments:
55		options[arg] = f"[{0}]".format(arg)
56
57		if f"{napp.username}/{napp.name}" in str(rule):
58		urls.append((options, rule.methods, f"{str(rule)}"))
59
60		return urls
61
62		def test_verify_api_urls(self):
63		"""Verify all APIs registered."""
64
65		expected_urls = [
66		(
67		{},
68		{"OPTIONS", "HEAD", "PUT"},
69		"/api/amlight/sdntrace_cp/trace/ ",
70		),
71		(
72		{},
73		{"OPTIONS", "HEAD", "PUT"},
74		"/api/amlight/sdntrace_cp/traces/ ",
75		),
76		]
77		urls = self.get_napp_urls(self.napp)
78		self.assertEqual(len(expected_urls), len(urls))
79
80		@patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
81		def test_trace_step(self, mock_flow_match):
82		"""Test trace_step success result."""
83		mock_flow_match.return_value = ["1"], ["entries"], 1
84		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
85
86		mock_interface = Interface("interface A", 1, MagicMock())
87		mock_interface.address = "00:00:00:00:00:00:00:01"
88
89		iface1 = get_interface_mock(
90		"", 1, get_switch_mock("00:00:00:00:00:00:00:01")
91		)
92		iface2 = get_interface_mock(
93		"", 2, get_switch_mock("00:00:00:00:00:00:00:02")
94		)
95		mock_interface.link = get_link_mock(iface1, iface2)
96		mock_interface.link.endpoint_a.port_number = 1
97		mock_interface.link.endpoint_a.port_number = 2
98
99		# Patch for utils.find_endpoint
100		switch.get_interface_by_port_no.return_value = mock_interface
101
102		entries = MagicMock()
103
104		result = self.napp.trace_step(switch, entries)
105
106		mock_flow_match.assert_called_once()
107		self.assertEqual(
108		result,
109		{
110		"dpid": "00:00:00:00:00:00:00:01",
111		"in_port": 2,
112		"out_port": 1,
113		"entries": ["entries"],
114		},
115		)
116
117		@patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
118		def test_trace_step__no_endpoint(self, mock_flow_match):
119		"""Test trace_step without endpoints available for switch/port."""
120		mock_flow_match.return_value = ["1"], ["entries"], 1
121		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
122
123		mock_interface = Interface("interface A", 1, MagicMock())
124		mock_interface.address = "00:00:00:00:00:00:00:01"
125		mock_interface.link = None
126
127		# Patch for utils.find_endpoint
128		switch.get_interface_by_port_no.return_value = mock_interface
129
130		entries = MagicMock()
131
132		result = self.napp.trace_step(switch, entries)
133
134		mock_flow_match.assert_called_once()
135		self.assertEqual(result, {"entries": ["entries"], "out_port": 1})
136
137		@patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
138		def test_trace_step__no_flow(self, mock_stored_flows):
139		"""Test trace_step without flows for the switch."""
140		switch = get_switch_mock("00:00:00:00:00:00:00:01")
141		entries = MagicMock()
142
143		stored_flows = {
144		"flow": {
145		"table_id": 0,
146		"cookie": 84114964,
147		"hard_timeout": 0,
148		"idle_timeout": 0,
149		"priority": 10,
150		},
151		"flow_id": 1,
152		"state": "installed",
153		"switch": "00:00:00:00:00:00:00:01",
154		}
155
156		mock_stored_flows.return_value = {
157		"00:00:00:00:00:00:00:01": [stored_flows]
158		}
159
160		result = self.napp.trace_step(switch, entries)
161		assert result is None
162
163		@patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
164		def test_tracepath(self, mock_trace_step):
165		"""Test tracepath with success result."""
166		eth = {"dl_vlan": 100}
167		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
168		switch = {"switch": dpid, "eth": eth}
169		entries = {"trace": switch}
170		mock_trace_step.return_value = {
171		"dpid": "00:00:00:00:00:00:00:02",
172		"in_port": 2,
173		"out_port": 3,
174		"entries": entries,
175		}
176
177		result = self.napp.tracepath(entries["trace"]["switch"])
178
179		assert result[0]["in"]["dpid"] == "00:00:00:00:00:00:00:01"
180		assert result[0]["in"]["port"] == 1
181		assert result[0]["in"]["type"] == "starting"
182		assert result[0]["out"]["port"] == 3
183
184		assert result[1]["in"]["dpid"] == "00:00:00:00:00:00:00:02"
185		assert result[1]["in"]["port"] == 2
186		assert result[1]["in"]["type"] == "trace"
187		assert result[1]["out"]["port"] == 3
188
189		def test_has_loop(self):
190		"""Test has_loop to detect a tracepath with loop."""
191		trace_result = [
192		{
193		"in": {
194		"dpid": "00:00:00:00:00:00:00:01",
195		"port": 2,
196		},
197		"out": {
198		"port": 1,
199		},
200		},
201		{
202		"in": {
203		"dpid": "00:00:00:00:00:00:00:03",
204		"port": 2,
205		},
206		"out": {
207		"port": 1,
208		},
209		},
210		{
211		"in": {
212		"dpid": "00:00:00:00:00:00:00:03",
213		"port": 3,
214		},
215		"out": {
216		"port": 1,
217		},
218		},
219		{
220		"in": {
221		"dpid": "00:00:00:00:00:00:00:03",
222		"port": 3,
223		},
224		"out": {
225		"port": 1,
226		},
227		},
228		]
229		trace_step = {
230		"dpid": "00:00:00:00:00:00:00:03",
231		"port": 3,
232		}
233
234		result = self.napp.has_loop(trace_step, trace_result)
235
236		self.assertTrue(result)
237
238		def test_has_loop__fail(self):
239		"""Test has_loop to detect a tracepath with loop."""
240		trace_result = [
241		{
242		"in": {
243		"dpid": "00:00:00:00:00:00:00:01",
244		"port": 2,
245		},
246		"out": {
247		"port": 1,
248		},
249		},
250		{
251		"in": {
252		"dpid": "00:00:00:00:00:00:00:02",
253		"port": 2,
254		},
255		"out": {
256		"port": 1,
257		},
258		},
259		]
260		trace_step = {
261		"dpid": "00:00:00:00:00:00:00:03",
262		"port": 2,
263		}
264
265		result = self.napp.has_loop(trace_step, trace_result)
266
267		self.assertFalse(result)
268
269		@patch("napps.amlight.sdntrace_cp.main.settings")
270		def test_update_circuits(self, mock_settings):
271		"""Test update_circuits event listener with success."""
272		mock_settings.FIND_CIRCUITS_IN_FLOWS = True
273
274		self.napp.automate = MagicMock()
275		self.napp.automate.find_circuits = MagicMock()
276
277		event = MagicMock()
278		self.napp.update_circuits(event)
279
280		self.napp.automate.find_circuits.assert_called_once()
281
282		@patch("napps.amlight.sdntrace_cp.main.settings")
283		def test_update_circuits__no_settings(self, mock_settings):
284		"""Test update_circuits event listener without
285		settings option enabled."""
286		mock_settings.FIND_CIRCUITS_IN_FLOWS = False
287
288		self.napp.automate = MagicMock()
289		self.napp.automate.find_circuits = MagicMock()
290
291		event = MagicMock()
292		self.napp.update_circuits(event)
293
294		self.napp.automate.find_circuits.assert_not_called()
295
296		@patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
297		def test_trace(self, mock_stored_flows):
298		"""Test trace rest call."""
299		api = get_test_client(get_controller_mock(), self.napp)
300		url = f"{self.server_name_url}/trace/"
301
302		payload = {
303		"trace": {
304		"switch": {
305		"dpid": "00:00:00:00:00:00:00:01",
306		"in_port": 1
307		},
308		"eth": {"dl_vlan": 100},
309		}
310		}
311		stored_flows = {
312		"00:00:00:00:00:00:00:01": {
313		"flow": {
314		"table_id": 0,
315		"cookie": 84114964,
316		"hard_timeout": 0,
317		"idle_timeout": 0,
318		"priority": 10,
319		},
320		"flow_id": 1,
321		"state": "installed",
322		"switch": "00:00:00:00:00:00:00:01",
323		}
324		}
325		mock_stored_flows.return_value = {
326		"00:00:00:00:00:00:00:01": [stored_flows]
327		}
328
329		response = api.put(
330		url, data=json.dumps(payload), content_type="application/json"
331		)
332		current_data = json.loads(response.data)
333		result = current_data["result"]
334
335		assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
336		assert result[0]["port"] == 1
337		assert result[0]["type"] == "starting"
338		assert result[0]["vlan"] == 100
339
340	View Code Duplication	@patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
		0 ignored issues – show Duplication introduced 2022-10-13 20:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
341		def test_traces(self, mock_stored_flows):
342		"""Test traces rest call for two traces with different switches."""
343		api = get_test_client(get_controller_mock(), self.napp)
344		url = f"{self.server_name_url}/traces/"
345
346		payload = [
347		{
348		"trace": {
349		"switch": {
350		"dpid": "00:00:00:00:00:00:00:01",
351		"in_port": 1
352		},
353		"eth": {"dl_vlan": 100},
354		}
355		},
356		{
357		"trace": {
358		"switch": {
359		"dpid": "00:00:00:00:00:00:00:02",
360		"in_port": 1},
361		"eth": {"dl_vlan": 100},
362		}
363		}
364		]
365
366		stored_flow = {
367		"id": 1,
368		"table_id": 0,
369		"cookie": 84114964,
370		"hard_timeout": 0,
371		"idle_timeout": 0,
372		"priority": 10,
373		}
374
375		mock_stored_flows.return_value = {
376		"00:00:00:00:00:00:00:01": [stored_flow]
377		}
378
379		response = api.put(
380		url, data=json.dumps(payload), content_type="application/json"
381		)
382		current_data = json.loads(response.data)
383		result1 = current_data["00:00:00:00:00:00:00:01"]
384		result2 = current_data["00:00:00:00:00:00:00:02"]
385
386		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
387		assert result1[0][0]["port"] == 1
388		assert result1[0][0]["type"] == "starting"
389		assert result1[0][0]["vlan"] == 100
390
391		assert result2[0][0]["dpid"] == "00:00:00:00:00:00:00:02"
392		assert result2[0][0]["port"] == 1
393		assert result2[0][0]["type"] == "starting"
394		assert result2[0][0]["vlan"] == 100
395
396	View Code Duplication	@patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
		0 ignored issues – show Duplication introduced 2022-10-13 20:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
397		def test_traces_same_switch(self, mock_stored_flows):
398		"""Test traces rest call for two traces with samw switches."""
399		api = get_test_client(get_controller_mock(), self.napp)
400		url = f"{self.server_name_url}/traces/"
401
402		payload = [
403		{
404		"trace": {
405		"switch": {
406		"dpid": "00:00:00:00:00:00:00:01",
407		"in_port": 1
408		},
409		"eth": {"dl_vlan": 100},
410		}
411		},
412		{
413		"trace": {
414		"switch": {
415		"dpid": "00:00:00:00:00:00:00:01",
416		"in_port": 2
417		},
418		"eth": {"dl_vlan": 100},
419		}
420		}
421		]
422
423		stored_flow = {
424		"id": 1,
425		"table_id": 0,
426		"cookie": 84114964,
427		"hard_timeout": 0,
428		"idle_timeout": 0,
429		"priority": 10,
430		}
431
432		mock_stored_flows.return_value = {
433		"00:00:00:00:00:00:00:01": [stored_flow]
434		}
435
436		response = api.put(
437		url, data=json.dumps(payload), content_type="application/json"
438		)
439		current_data = json.loads(response.data)
440		result = current_data["00:00:00:00:00:00:00:01"]
441
442		assert len(current_data) == 1
443
444		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
445		assert result[0][0]["port"] == 1
446		assert result[0][0]["type"] == "starting"
447		assert result[0][0]["vlan"] == 100
448
449		assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:01"
450		assert result[1][0]["port"] == 2
451		assert result[1][0]["type"] == "starting"
452		assert result[1][0]["vlan"] == 100
453
454		@patch("napps.amlight.sdntrace_cp.main.Main.get_stored_flows")
455		def test_traces_twice(self, mock_stored_flows):
456		"""Test traces rest call for two equal traces."""
457		api = get_test_client(get_controller_mock(), self.napp)
458		url = f"{self.server_name_url}/traces/"
459
460		payload = [
461		{
462		"trace": {
463		"switch": {
464		"dpid": "00:00:00:00:00:00:00:01",
465		"in_port": 1
466		},
467		"eth": {"dl_vlan": 100},
468		}
469		},
470		{
471		"trace": {
472		"switch": {
473		"dpid": "00:00:00:00:00:00:00:01",
474		"in_port": 1
475		},
476		"eth": {"dl_vlan": 100},
477		}
478		}
479		]
480		stored_flow = {
481		"id": 1,
482		"table_id": 0,
483		"cookie": 84114964,
484		"hard_timeout": 0,
485		"idle_timeout": 0,
486		"priority": 10,
487		}
488
489		mock_stored_flows.return_value = {
490		"00:00:00:00:00:00:00:01": [stored_flow]
491		}
492
493		response = api.put(
494		url, data=json.dumps(payload), content_type="application/json"
495		)
496		current_data = json.loads(response.data)
497		result = current_data["00:00:00:00:00:00:00:01"]
498
499		assert len(current_data) == 1
500		assert len(result) == 1
501
502		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
503		assert result[0][0]["port"] == 1
504		assert result[0][0]["type"] == "starting"
505		assert result[0][0]["vlan"] == 100
506

kytos-ng / sdntrace_cp

Pull Request — master (#47)

build.tests.unit.test_main.TestMain.test_traces() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like