build.tests.unit.test_main.TestMain.test_tracepath_loop() - Code Metrics - Inspection of "feaure: Add new case of loop when the outgoing int..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#81)

unknown

created 2023-03-10 17:28 UTC

TestMain.test_tracepath_loop() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	22
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	17
nop	2
dl	0
loc	22
rs	9.55
c	0
b	0
f	0

"""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"] == "intermediary"
        assert result[1]["out"]["port"] == 3

    @patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
    def test_tracepath_loop(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:01",
            "in_port": 1,
            "out_port": 1,
            "entries": entries,
        }

        result = self.napp.tracepath(
                                        entries["trace"]["switch"],
                                        {}
                                    )
        assert len(result) == 2
        # input interface = output interface
        assert result[0]["in"]["type"] == "starting"
        assert result[1]["in"]["type"] == "loop"

    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_trace_missing_parameter(self, mock_stored_flows):
        """Test trace rest call with a missing parameter."""
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/trace/"

        payload = {
            "trace": {
                "switch": {
                    "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)
        assert current_data["result"] == []

    @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"""
        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:0a",
                                "in_port": 1
                            },
                            "eth": {
                                "dl_vlan": 100
                            }
                        }
                    },
                    {
                        "trace": {
                            "switch": {
                                "dpid": "00:00:00:00:00:00:00:02",
                            }
                        }
                    },
                    {
                        "trace": {
                            "switch": {
                                "dpid": "00:00:00:00:00:00:00:02",
                                "in_port": 2
                            },
                            "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) == 4

        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:0a"
        assert result[1][0]["port"] == 1
        assert result[1][0]["type"] == "last"
        assert result[1][0]["vlan"] == 100
        assert result[1][0]["out"] is None

        assert result[2] == []

        assert result[3][0]["dpid"] == "00:00:00:00:00:00:00:02"
        assert result[3][0]["port"] == 2
        assert result[3][0]["type"] == "incomplete"
        assert result[3][0]["vlan"] == 100
        assert result[3][0]["out"] is None

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

    def test_traces_with_loop(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": "output", "port": 1}],
            }
        }

        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["result"]
        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"] == "loop"
        assert result[0][0]["vlan"] == 100
        assert result[0][0]["out"] == {"port": 1, "vlan": 100}

    @patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
    def test_traces_no_action(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_flow1 = {
            "id": 1,
            "flow": {
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
                "match": {"dl_vlan": 100, "in_port": 1}
            }
        }
        stored_flow2 = {
            "id": 1,
            "flow": {
                "table_id": 0,
                "cookie": 84114964,
                "hard_timeout": 0,
                "idle_timeout": 0,
                "priority": 10,
                "match": {"dl_vlan": 100, "in_port": 1},
                "actions": []
            }
        }

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

        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][-1]['type'] == "incomplete"
        assert result[1][-1]['type'] == "incomplete"


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"] == "intermediary"
238		assert result[1]["out"]["port"] == 3
239
240		@patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
241		def test_tracepath_loop(self, mock_trace_step):
242		"""Test tracepath with success result."""
243		eth = {"dl_vlan": 100}
244		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
245		switch = {"switch": dpid, "eth": eth}
246		entries = {"trace": switch}
247		mock_trace_step.return_value = {
248		"dpid": "00:00:00:00:00:00:00:01",
249		"in_port": 1,
250		"out_port": 1,
251		"entries": entries,
252		}
253
254		result = self.napp.tracepath(
255		entries["trace"]["switch"],
256		{}
257		)
258		assert len(result) == 2
259		# input interface = output interface
260		assert result[0]["in"]["type"] == "starting"
261		assert result[1]["in"]["type"] == "loop"
262
263		def test_has_loop(self):
264		"""Test has_loop to detect a tracepath with loop."""
265		trace_result = [
266		{
267		"in": {
268		"dpid": "00:00:00:00:00:00:00:01",
269		"port": 2,
270		},
271		"out": {
272		"port": 1,
273		},
274		},
275		{
276		"in": {
277		"dpid": "00:00:00:00:00:00:00:03",
278		"port": 2,
279		},
280		"out": {
281		"port": 1,
282		},
283		},
284		{
285		"in": {
286		"dpid": "00:00:00:00:00:00:00:03",
287		"port": 3,
288		},
289		"out": {
290		"port": 1,
291		},
292		},
293		{
294		"in": {
295		"dpid": "00:00:00:00:00:00:00:03",
296		"port": 3,
297		},
298		"out": {
299		"port": 1,
300		},
301		},
302		]
303		trace_step = {
304		"dpid": "00:00:00:00:00:00:00:03",
305		"port": 3,
306		}
307
308		result = self.napp.has_loop(trace_step, trace_result)
309
310		self.assertTrue(result)
311
312		def test_has_loop__fail(self):
313		"""Test has_loop to detect a tracepath with loop."""
314		trace_result = [
315		{
316		"in": {
317		"dpid": "00:00:00:00:00:00:00:01",
318		"port": 2,
319		},
320		"out": {
321		"port": 1,
322		},
323		},
324		{
325		"in": {
326		"dpid": "00:00:00:00:00:00:00:02",
327		"port": 2,
328		},
329		"out": {
330		"port": 1,
331		},
332		},
333		]
334		trace_step = {
335		"dpid": "00:00:00:00:00:00:00:03",
336		"port": 2,
337		}
338
339		result = self.napp.has_loop(trace_step, trace_result)
340
341		self.assertFalse(result)
342
343		@patch("napps.amlight.sdntrace_cp.main.settings")
344		def test_update_circuits(self, mock_settings):
345		"""Test update_circuits event listener with success."""
346		mock_settings.FIND_CIRCUITS_IN_FLOWS = True
347
348		self.napp.automate = MagicMock()
349		self.napp.automate.find_circuits = MagicMock()
350
351		self.napp.update_circuits()
352
353		self.napp.automate.find_circuits.assert_called_once()
354
355		@patch("napps.amlight.sdntrace_cp.main.settings")
356		def test_update_circuits__no_settings(self, mock_settings):
357		"""Test update_circuits event listener without
358		settings option enabled."""
359		mock_settings.FIND_CIRCUITS_IN_FLOWS = False
360
361		self.napp.automate = MagicMock()
362		self.napp.automate.find_circuits = MagicMock()
363
364		self.napp.update_circuits()
365
366		self.napp.automate.find_circuits.assert_not_called()
367
368		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
369		def test_trace(self, mock_stored_flows):
370		"""Test trace rest call."""
371		api = get_test_client(get_controller_mock(), self.napp)
372		url = f"{self.server_name_url}/trace/"
373
374		payload = {
375		"trace": {
376		"switch": {
377		"dpid": "00:00:00:00:00:00:00:01",
378		"in_port": 1
379		},
380		"eth": {"dl_vlan": 100},
381		}
382		}
383		stored_flows = {
384		"flow": {
385		"table_id": 0,
386		"cookie": 84114964,
387		"hard_timeout": 0,
388		"idle_timeout": 0,
389		"priority": 10,
390		"match": {"dl_vlan": 100, "in_port": 1},
391		"actions": [
392		{"action_type": "push_vlan"},
393		{"action_type": "set_vlan", "vlan_id": 200},
394		{"action_type": "output", "port": 2}
395		],
396		},
397		"flow_id": 1,
398		"state": "installed",
399		"switch": "00:00:00:00:00:00:00:01",
400		}
401		mock_stored_flows.return_value = {
402		"00:00:00:00:00:00:00:01": [stored_flows]
403		}
404
405		response = api.put(
406		url, data=json.dumps(payload), content_type="application/json"
407		)
408		current_data = json.loads(response.data)
409		result = current_data["result"]
410
411		assert len(result) == 1
412		assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
413		assert result[0]["port"] == 1
414		assert result[0]["type"] == "last"
415		assert result[0]["vlan"] == 100
416		assert result[0]["out"] == {"port": 2, "vlan": 200}
417
418		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
419		def test_trace_missing_parameter(self, mock_stored_flows):
420		"""Test trace rest call with a missing parameter."""
421		api = get_test_client(get_controller_mock(), self.napp)
422		url = f"{self.server_name_url}/trace/"
423
424		payload = {
425		"trace": {
426		"switch": {
427		"in_port": 1
428		},
429		"eth": {"dl_vlan": 100},
430		}
431		}
432		stored_flows = {
433		"flow": {
434		"table_id": 0,
435		"cookie": 84114964,
436		"hard_timeout": 0,
437		"idle_timeout": 0,
438		"priority": 10,
439		"match": {"dl_vlan": 100, "in_port": 1},
440		"actions": [
441		{"action_type": "push_vlan"},
442		{"action_type": "set_vlan", "vlan_id": 200},
443		{"action_type": "output", "port": 2}
444		],
445		},
446		"flow_id": 1,
447		"state": "installed",
448		"switch": "00:00:00:00:00:00:00:01",
449		}
450		mock_stored_flows.return_value = {
451		"00:00:00:00:00:00:00:01": [stored_flows]
452		}
453
454		response = api.put(
455		url, data=json.dumps(payload), content_type="application/json"
456		)
457		current_data = json.loads(response.data)
458		assert current_data["result"] == []
459
460	View Code Duplication	@patch("napps.amlight.sdntrace_cp.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...
461		def test_get_traces(self, mock_stored_flows):
462		"""Test traces rest call."""
463		api = get_test_client(get_controller_mock(), self.napp)
464		url = f"{self.server_name_url}/traces/"
465
466		payload = [{
467		"trace": {
468		"switch": {
469		"dpid": "00:00:00:00:00:00:00:01",
470		"in_port": 1
471		},
472		"eth": {"dl_vlan": 100},
473		}
474		}]
475
476		stored_flow = {
477		"id": 1,
478		"flow": {
479		"table_id": 0,
480		"cookie": 84114964,
481		"hard_timeout": 0,
482		"idle_timeout": 0,
483		"priority": 10,
484		"match": {"dl_vlan": 100, "in_port": 1},
485		"actions": [
486		{"action_type": "pop_vlan"},
487		{"action_type": "output", "port": 2},
488		],
489		}
490		}
491
492		mock_stored_flows.return_value = {
493		"00:00:00:00:00:00:00:01": [stored_flow]
494		}
495
496		response = api.put(
497		url, data=json.dumps(payload), content_type="application/json"
498		)
499		current_data = json.loads(response.data)
500		result1 = current_data["result"]
501
502		assert len(result1) == 1
503		assert len(result1[0]) == 1
504		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
505		assert result1[0][0]["port"] == 1
506		assert result1[0][0]["type"] == "last"
507		assert result1[0][0]["vlan"] == 100
508		assert result1[0][0]["out"] == {"port": 2}
509
510		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
511		def test_traces(self, mock_stored_flows):
512		"""Test traces rest call"""
513		api = get_test_client(get_controller_mock(), self.napp)
514		url = f"{self.server_name_url}/traces/"
515
516		payload = [
517		{
518		"trace": {
519		"switch": {
520		"dpid": "00:00:00:00:00:00:00:01",
521		"in_port": 1
522		},
523		"eth": {
524		"dl_vlan": 100
525		}
526		}
527		},
528		{
529		"trace": {
530		"switch": {
531		"dpid": "00:00:00:00:00:00:00:0a",
532		"in_port": 1
533		},
534		"eth": {
535		"dl_vlan": 100
536		}
537		}
538		},
539		{
540		"trace": {
541		"switch": {
542		"dpid": "00:00:00:00:00:00:00:02",
543		}
544		}
545		},
546		{
547		"trace": {
548		"switch": {
549		"dpid": "00:00:00:00:00:00:00:02",
550		"in_port": 2
551		},
552		"eth": {
553		"dl_vlan": 100
554		}
555		}
556		}
557		]
558
559		stored_flow = {
560		"id": 1,
561		"flow": {
562		"table_id": 0,
563		"cookie": 84114964,
564		"hard_timeout": 0,
565		"idle_timeout": 0,
566		"priority": 10,
567		"match": {"dl_vlan": 100, "in_port": 1},
568		"actions": [{"action_type": "output", "port": 2}],
569		}
570		}
571
572		mock_stored_flows.return_value = {
573		"00:00:00:00:00:00:00:01": [stored_flow],
574		"00:00:00:00:00:00:00:02": [stored_flow],
575		}
576
577		response = api.put(
578		url, data=json.dumps(payload), content_type="application/json"
579		)
580		current_data = json.loads(response.data)
581		result = current_data["result"]
582		assert len(result) == 4
583
584		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
585		assert result[0][0]["port"] == 1
586		assert result[0][0]["type"] == "last"
587		assert result[0][0]["vlan"] == 100
588		assert result[0][0]["out"] == {"port": 2, "vlan": 100}
589
590		assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:0a"
591		assert result[1][0]["port"] == 1
592		assert result[1][0]["type"] == "last"
593		assert result[1][0]["vlan"] == 100
594		assert result[1][0]["out"] is None
595
596		assert result[2] == []
597
598		assert result[3][0]["dpid"] == "00:00:00:00:00:00:00:02"
599		assert result[3][0]["port"] == 2
600		assert result[3][0]["type"] == "incomplete"
601		assert result[3][0]["vlan"] == 100
602		assert result[3][0]["out"] is None
603
604	View Code Duplication	@patch("napps.amlight.sdntrace_cp.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...
605		def test_traces_with_loop(self, mock_stored_flows):
606		"""Test traces rest call"""
607		api = get_test_client(get_controller_mock(), self.napp)
608		url = f"{self.server_name_url}/traces/"
609
610		payload = [
611		{
612		"trace": {
613		"switch": {
614		"dpid": "00:00:00:00:00:00:00:01",
615		"in_port": 1
616		},
617		"eth": {
618		"dl_vlan": 100
619		}
620		}
621		}
622		]
623
624		stored_flow = {
625		"id": 1,
626		"flow": {
627		"table_id": 0,
628		"cookie": 84114964,
629		"hard_timeout": 0,
630		"idle_timeout": 0,
631		"priority": 10,
632		"match": {"dl_vlan": 100, "in_port": 1},
633		"actions": [{"action_type": "output", "port": 1}],
634		}
635		}
636
637		mock_stored_flows.return_value = {
638		"00:00:00:00:00:00:00:01": [stored_flow],
639		}
640
641		response = api.put(
642		url, data=json.dumps(payload), content_type="application/json"
643		)
644		current_data = json.loads(response.data)
645		result = current_data["result"]
646		assert len(result) == 1
647		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
648		assert result[0][0]["port"] == 1
649		assert result[0][0]["type"] == "loop"
650		assert result[0][0]["vlan"] == 100
651		assert result[0][0]["out"] == {"port": 1, "vlan": 100}
652
653		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
654		def test_traces_no_action(self, mock_stored_flows):
655		"""Test traces rest call for two traces with different switches."""
656		api = get_test_client(get_controller_mock(), self.napp)
657		url = f"{self.server_name_url}/traces/"
658
659		payload = [
660		{
661		"trace": {
662		"switch": {
663		"dpid": "00:00:00:00:00:00:00:01",
664		"in_port": 1
665		},
666		"eth": {"dl_vlan": 100},
667		}
668		},
669		{
670		"trace": {
671		"switch": {
672		"dpid": "00:00:00:00:00:00:00:02",
673		"in_port": 1},
674		"eth": {"dl_vlan": 100},
675		}
676		}
677		]
678
679		stored_flow1 = {
680		"id": 1,
681		"flow": {
682		"table_id": 0,
683		"cookie": 84114964,
684		"hard_timeout": 0,
685		"idle_timeout": 0,
686		"priority": 10,
687		"match": {"dl_vlan": 100, "in_port": 1}
688		}
689		}
690		stored_flow2 = {
691		"id": 1,
692		"flow": {
693		"table_id": 0,
694		"cookie": 84114964,
695		"hard_timeout": 0,
696		"idle_timeout": 0,
697		"priority": 10,
698		"match": {"dl_vlan": 100, "in_port": 1},
699		"actions": []
700		}
701		}
702
703		mock_stored_flows.return_value = {
704		"00:00:00:00:00:00:00:01": [stored_flow1],
705		"00:00:00:00:00:00:00:02": [stored_flow2]
706		}
707
708		response = api.put(
709		url, data=json.dumps(payload), content_type="application/json"
710		)
711		current_data = json.loads(response.data)
712		result = current_data["result"]
713		assert len(result) == 2
714		assert result[0][-1]['type'] == "incomplete"
715		assert result[1][-1]['type'] == "incomplete"
716

kytos-ng / sdntrace_cp

Pull Request — master (#81)

TestMain.test_tracepath_loop() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like