build.tests.unit.test_main.TestMain.test_traces_twice() - Code Metrics - Inspection of "fix: Return a list of paths with the `PUT /traces`..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#75)

by Vinicius

created 2023-02-27 14:27 UTC

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

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	57
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	38
nop	2
dl	0
loc	57
rs	8.968
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())

        sw1 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        intf_sw1 = get_interface_mock("eth1", 1, sw1)
        intf_sw1.link = None
        sw1.get_interface_by_port_no.return_value = intf_sw1
        sw2 = get_switch_mock("00:00:00:00:00:00:00:02", 0x04)
        intf_sw2 = get_interface_mock("eth1", 1, sw2)
        intf_sw2.link = None
        sw2.get_interface_by_port_no.return_value = intf_sw2
        self.napp.controller.switches = {sw1.dpid: sw1, sw2.dpid: sw2}

    @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()

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

        stored_flows_arg = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }

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

        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()

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

        stored_flows_arg = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }

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

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

    def test_trace_step__no_flow(self):
        """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",
        }

        stored_flows_arg = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }

        result = self.napp.trace_step(switch, entries, stored_flows_arg)
        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,
        }

        stored_flows_arg = {
            "00:00:00:00:00:00:00:01": [],
            "00:00:00:00:00:00:00:02": [],
        }

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

        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()

        self.napp.update_circuits()

        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()

        self.napp.update_circuits()

        self.napp.automate.find_circuits.assert_not_called()

    @patch("napps.amlight.sdntrace_cp.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 = {
                "flow": {
                    "table_id": 0,
                    "cookie": 84114964,
                    "hard_timeout": 0,
                    "idle_timeout": 0,
                    "priority": 10,
                    "match": {"dl_vlan": 100, "in_port": 1},
                    "actions": [
                        {"action_type": "push_vlan"},
                        {"action_type": "set_vlan", "vlan_id": 200},
                        {"action_type": "output", "port": 2}
                    ],
                },
                "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 len(result) == 1
        assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0]["port"] == 1
        assert result[0]["type"] == "last"
        assert result[0]["vlan"] == 100
        assert result[0]["out"] == {"port": 2, "vlan": 200}

    @patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
    def test_get_traces(self, mock_stored_flows):
        """Test traces rest call."""
        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},
            }
        }]

        stored_flow = {
            "id": 1,
            "flow": {
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
                "match": {"dl_vlan": 100, "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 2},
                ],
            }
        }

        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["result"]

        assert len(result1) == 1
        assert len(result1[0]) == 1
        assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result1[0][0]["port"] == 1
        assert result1[0][0]["type"] == "last"
        assert result1[0][0]["vlan"] == 100
        assert result1[0][0]["out"] == {"port": 2}

    @patch("napps.amlight.sdntrace_cp.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,
            "flow": {
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
                "match": {"dl_vlan": 100, "in_port": 1},
                "actions": [{"action_type": "output", "port": 2}],
            }
        }

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

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

        assert len(result) == 2

        assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0][0]["port"] == 1
        assert result[0][0]["type"] == "last"
        assert result[0][0]["vlan"] == 100
        assert result[0][0]["out"] == {"port": 2, "vlan": 100}

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

kytos-ng / sdntrace_cp

Pull Request — master (#75)

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

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like