build.tests.unit.test_main.TestMain.test_trace() - Code Metrics - Inspection of "Adding output port information to help traces for..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#59)

by Italo Valcy

created 2023-01-20 18:29 UTC

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

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	49
Code Lines	37

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	37
nop	3
dl	0
loc	49
rs	8.9919
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())

    @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 = {
            "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]
        }

        switch_01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        switch_01.is_enabled.return_value = True

        self.napp.controller.switches = {
            "00:00:00:00:00:00:00:01": switch_01
        }

        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[1]["in"]["dpid"] == "00:00:00:00:00:00:00:02"
        assert result[1]["in"]["port"] == 2
        assert result[1]["in"]["type"] == "trace"

    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.utils.get_stored_flows")
    @patch("napps.amlight.sdntrace_cp.utils.requests")
    def test_trace(self, mock_request_get, 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,
                },
                "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]
        }
        mock_json = MagicMock()
        mock_json.json.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flows]
        }
        mock_request_get.get.return_value = mock_json

        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"] == "starting"
        assert result[0]["vlan"] == 100
        assert result[0]["out"] is None

    @patch("napps.amlight.sdntrace_cp.utils.get_stored_flows")
    @patch("napps.amlight.sdntrace_cp.utils.requests")
    def test_get_traces(self, mock_request_get, 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,
                "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]
        }
        mock_json = MagicMock()
        mock_json.json.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }
        mock_request_get.get.return_value = mock_json
        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"]

        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"] == "starting"
        assert result1[0][0]["vlan"] == 100
        assert result1[0][0]["out"] is None

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

    @patch("napps.amlight.sdntrace_cp.utils.requests")
    def test_traces(self, mock_request_get, 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]
        }
        mock_json = MagicMock()
        mock_json.json.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }
        mock_request_get.get.return_value = mock_json

        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.utils.get_stored_flows")

    @patch("napps.amlight.sdntrace_cp.utils.requests")
    def test_traces_same_switch(self, mock_request_get, 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]
        }
        mock_json = MagicMock()
        mock_json.json.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }
        mock_request_get.get.return_value = mock_json

        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.utils.get_stored_flows")
    @patch("napps.amlight.sdntrace_cp.utils.requests")
    def test_traces_twice(self, mock_request_get, 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]
        }
        mock_json = MagicMock()
        mock_json.json.return_value = {
            "00:00:00:00:00:00:00:01": [stored_flow]
        }
        mock_request_get.get.return_value = mock_json

        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		stored_flows = {
105		"flow": {
106		"table_id": 0,
107		"cookie": 84114964,
108		"hard_timeout": 0,
109		"idle_timeout": 0,
110		"priority": 10,
111		},
112		"flow_id": 1,
113		"state": "installed",
114		"switch": "00:00:00:00:00:00:00:01",
115		}
116
117		stored_flows_arg = {
118		"00:00:00:00:00:00:00:01": [stored_flows]
119		}
120
121		result = self.napp.trace_step(switch, entries, stored_flows_arg)
122
123		mock_flow_match.assert_called_once()
124		self.assertEqual(
125		result,
126		{
127		"dpid": "00:00:00:00:00:00:00:01",
128		"in_port": 2,
129		"out_port": 1,
130		"entries": ["entries"],
131		},
132		)
133
134		@patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
135		def test_trace_step__no_endpoint(self, mock_flow_match):
136		"""Test trace_step without endpoints available for switch/port."""
137		mock_flow_match.return_value = ["1"], ["entries"], 1
138		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
139
140		mock_interface = Interface("interface A", 1, MagicMock())
141		mock_interface.address = "00:00:00:00:00:00:00:01"
142		mock_interface.link = None
143
144		# Patch for utils.find_endpoint
145		switch.get_interface_by_port_no.return_value = mock_interface
146
147		entries = MagicMock()
148
149		stored_flows = {
150		"flow": {
151		"table_id": 0,
152		"cookie": 84114964,
153		"hard_timeout": 0,
154		"idle_timeout": 0,
155		"priority": 10,
156		},
157		"flow_id": 1,
158		"state": "installed",
159		"switch": "00:00:00:00:00:00:00:01",
160		}
161
162		stored_flows_arg = {
163		"00:00:00:00:00:00:00:01": [stored_flows]
164		}
165
166		result = self.napp.trace_step(switch, entries, stored_flows_arg)
167
168		mock_flow_match.assert_called_once()
169		self.assertEqual(result, {"entries": ["entries"], "out_port": 1})
170
171		def test_trace_step__no_flow(self):
172		"""Test trace_step without flows for the switch."""
173		switch = get_switch_mock("00:00:00:00:00:00:00:01")
174		entries = MagicMock()
175
176		stored_flows = {
177		"flow": {
178		"table_id": 0,
179		"cookie": 84114964,
180		"hard_timeout": 0,
181		"idle_timeout": 0,
182		"priority": 10,
183		},
184		"flow_id": 1,
185		"state": "installed",
186		"switch": "00:00:00:00:00:00:00:01",
187		}
188
189		stored_flows_arg = {
190		"00:00:00:00:00:00:00:01": [stored_flows]
191		}
192
193		result = self.napp.trace_step(switch, entries, stored_flows_arg)
194		assert result is None
195
196		@patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
197		def test_tracepath(self, mock_trace_step):
198		"""Test tracepath with success result."""
199		eth = {"dl_vlan": 100}
200		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
201		switch = {"switch": dpid, "eth": eth}
202		entries = {"trace": switch}
203		mock_trace_step.return_value = {
204		"dpid": "00:00:00:00:00:00:00:02",
205		"in_port": 2,
206		"out_port": 3,
207		"entries": entries,
208		}
209		stored_flows = {
210		"flow": {
211		"table_id": 0,
212		"cookie": 84114964,
213		"hard_timeout": 0,
214		"idle_timeout": 0,
215		"priority": 10,
216		},
217		"flow_id": 1,
218		"state": "installed",
219		"switch": "00:00:00:00:00:00:00:01",
220		}
221
222		stored_flows_arg = {
223		"00:00:00:00:00:00:00:01": [stored_flows]
224		}
225
226		switch_01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
227		switch_01.is_enabled.return_value = True
228
229		self.napp.controller.switches = {
230		"00:00:00:00:00:00:00:01": switch_01
231		}
232
233		result = self.napp.tracepath(
234		entries["trace"]["switch"],
235		stored_flows_arg
236		)
237
238		assert result[0]["in"]["dpid"] == "00:00:00:00:00:00:00:01"
239		assert result[0]["in"]["port"] == 1
240		assert result[0]["in"]["type"] == "starting"
241
242		assert result[1]["in"]["dpid"] == "00:00:00:00:00:00:00:02"
243		assert result[1]["in"]["port"] == 2
244		assert result[1]["in"]["type"] == "trace"
245
246		def test_has_loop(self):
247		"""Test has_loop to detect a tracepath with loop."""
248		trace_result = [
249		{
250		"in": {
251		"dpid": "00:00:00:00:00:00:00:01",
252		"port": 2,
253		},
254		"out": {
255		"port": 1,
256		},
257		},
258		{
259		"in": {
260		"dpid": "00:00:00:00:00:00:00:03",
261		"port": 2,
262		},
263		"out": {
264		"port": 1,
265		},
266		},
267		{
268		"in": {
269		"dpid": "00:00:00:00:00:00:00:03",
270		"port": 3,
271		},
272		"out": {
273		"port": 1,
274		},
275		},
276		{
277		"in": {
278		"dpid": "00:00:00:00:00:00:00:03",
279		"port": 3,
280		},
281		"out": {
282		"port": 1,
283		},
284		},
285		]
286		trace_step = {
287		"dpid": "00:00:00:00:00:00:00:03",
288		"port": 3,
289		}
290
291		result = self.napp.has_loop(trace_step, trace_result)
292
293		self.assertTrue(result)
294
295		def test_has_loop__fail(self):
296		"""Test has_loop to detect a tracepath with loop."""
297		trace_result = [
298		{
299		"in": {
300		"dpid": "00:00:00:00:00:00:00:01",
301		"port": 2,
302		},
303		"out": {
304		"port": 1,
305		},
306		},
307		{
308		"in": {
309		"dpid": "00:00:00:00:00:00:00:02",
310		"port": 2,
311		},
312		"out": {
313		"port": 1,
314		},
315		},
316		]
317		trace_step = {
318		"dpid": "00:00:00:00:00:00:00:03",
319		"port": 2,
320		}
321
322		result = self.napp.has_loop(trace_step, trace_result)
323
324		self.assertFalse(result)
325
326		@patch("napps.amlight.sdntrace_cp.main.settings")
327		def test_update_circuits(self, mock_settings):
328		"""Test update_circuits event listener with success."""
329		mock_settings.FIND_CIRCUITS_IN_FLOWS = True
330
331		self.napp.automate = MagicMock()
332		self.napp.automate.find_circuits = MagicMock()
333
334		self.napp.update_circuits()
335
336		self.napp.automate.find_circuits.assert_called_once()
337
338		@patch("napps.amlight.sdntrace_cp.main.settings")
339		def test_update_circuits__no_settings(self, mock_settings):
340		"""Test update_circuits event listener without
341		settings option enabled."""
342		mock_settings.FIND_CIRCUITS_IN_FLOWS = False
343
344		self.napp.automate = MagicMock()
345		self.napp.automate.find_circuits = MagicMock()
346
347		self.napp.update_circuits()
348
349		self.napp.automate.find_circuits.assert_not_called()
350
351		@patch("napps.amlight.sdntrace_cp.utils.get_stored_flows")
352		@patch("napps.amlight.sdntrace_cp.utils.requests")
353		def test_trace(self, mock_request_get, mock_stored_flows):
354		"""Test trace rest call."""
355		api = get_test_client(get_controller_mock(), self.napp)
356		url = f"{self.server_name_url}/trace/"
357
358		payload = {
359		"trace": {
360		"switch": {
361		"dpid": "00:00:00:00:00:00:00:01",
362		"in_port": 1
363		},
364		"eth": {"dl_vlan": 100},
365		}
366		}
367		stored_flows = {
368		"flow": {
369		"table_id": 0,
370		"cookie": 84114964,
371		"hard_timeout": 0,
372		"idle_timeout": 0,
373		"priority": 10,
374		},
375		"flow_id": 1,
376		"state": "installed",
377		"switch": "00:00:00:00:00:00:00:01",
378		}
379		mock_stored_flows.return_value = {
380		"00:00:00:00:00:00:00:01": [stored_flows]
381		}
382		mock_json = MagicMock()
383		mock_json.json.return_value = {
384		"00:00:00:00:00:00:00:01": [stored_flows]
385		}
386		mock_request_get.get.return_value = mock_json
387
388		response = api.put(
389		url, data=json.dumps(payload), content_type="application/json"
390		)
391		current_data = json.loads(response.data)
392		result = current_data["result"]
393
394		assert len(result) == 1
395		assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
396		assert result[0]["port"] == 1
397		assert result[0]["type"] == "starting"
398		assert result[0]["vlan"] == 100
399		assert result[0]["out"] is None
400
401		@patch("napps.amlight.sdntrace_cp.utils.get_stored_flows")
402		@patch("napps.amlight.sdntrace_cp.utils.requests")
403		def test_get_traces(self, mock_request_get, mock_stored_flows):
404		"""Test traces rest call."""
405		api = get_test_client(get_controller_mock(), self.napp)
406		url = f"{self.server_name_url}/traces/"
407
408		payload = [{
409		"trace": {
410		"switch": {
411		"dpid": "00:00:00:00:00:00:00:01",
412		"in_port": 1
413		},
414		"eth": {"dl_vlan": 100},
415		}
416		}]
417
418		stored_flow = {
419		"id": 1,
420		"table_id": 0,
421		"cookie": 84114964,
422		"hard_timeout": 0,
423		"idle_timeout": 0,
424		"priority": 10,
425		}
426
427		mock_stored_flows.return_value = {
428		"00:00:00:00:00:00:00:01": [stored_flow]
429		}
430		mock_json = MagicMock()
431		mock_json.json.return_value = {
432		"00:00:00:00:00:00:00:01": [stored_flow]
433		}
434		mock_request_get.get.return_value = mock_json
435		response = api.put(
436		url, data=json.dumps(payload), content_type="application/json"
437		)
438		current_data = json.loads(response.data)
439		result1 = current_data["00:00:00:00:00:00:00:01"]
440
441		assert len(result1) == 1
442		assert len(result1[0]) == 1
443		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
444		assert result1[0][0]["port"] == 1
445		assert result1[0][0]["type"] == "starting"
446		assert result1[0][0]["vlan"] == 100
447		assert result1[0][0]["out"] is None
448
449	View Code Duplication	@patch("napps.amlight.sdntrace_cp.utils.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...
450		@patch("napps.amlight.sdntrace_cp.utils.requests")
451		def test_traces(self, mock_request_get, mock_stored_flows):
452		"""Test traces rest call for two traces with different switches."""
453		api = get_test_client(get_controller_mock(), self.napp)
454		url = f"{self.server_name_url}/traces/"
455
456		payload = [
457		{
458		"trace": {
459		"switch": {
460		"dpid": "00:00:00:00:00:00:00:01",
461		"in_port": 1
462		},
463		"eth": {"dl_vlan": 100},
464		}
465		},
466		{
467		"trace": {
468		"switch": {
469		"dpid": "00:00:00:00:00:00:00:02",
470		"in_port": 1},
471		"eth": {"dl_vlan": 100},
472		}
473		}
474		]
475
476		stored_flow = {
477		"id": 1,
478		"table_id": 0,
479		"cookie": 84114964,
480		"hard_timeout": 0,
481		"idle_timeout": 0,
482		"priority": 10,
483		}
484
485		mock_stored_flows.return_value = {
486		"00:00:00:00:00:00:00:01": [stored_flow]
487		}
488		mock_json = MagicMock()
489		mock_json.json.return_value = {
490		"00:00:00:00:00:00:00:01": [stored_flow]
491		}
492		mock_request_get.get.return_value = mock_json
493
494		response = api.put(
495		url, data=json.dumps(payload), content_type="application/json"
496		)
497		current_data = json.loads(response.data)
498		result1 = current_data["00:00:00:00:00:00:00:01"]
499		result2 = current_data["00:00:00:00:00:00:00:02"]
500
501		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
502		assert result1[0][0]["port"] == 1
503		assert result1[0][0]["type"] == "starting"
504		assert result1[0][0]["vlan"] == 100
505
506		assert result2[0][0]["dpid"] == "00:00:00:00:00:00:00:02"
507		assert result2[0][0]["port"] == 1
508		assert result2[0][0]["type"] == "starting"
509		assert result2[0][0]["vlan"] == 100
510
511	View Code Duplication	@patch("napps.amlight.sdntrace_cp.utils.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...
512		@patch("napps.amlight.sdntrace_cp.utils.requests")
513		def test_traces_same_switch(self, mock_request_get, mock_stored_flows):
514		"""Test traces rest call for two traces with samw switches."""
515		api = get_test_client(get_controller_mock(), self.napp)
516		url = f"{self.server_name_url}/traces/"
517
518		payload = [
519		{
520		"trace": {
521		"switch": {
522		"dpid": "00:00:00:00:00:00:00:01",
523		"in_port": 1
524		},
525		"eth": {"dl_vlan": 100},
526		}
527		},
528		{
529		"trace": {
530		"switch": {
531		"dpid": "00:00:00:00:00:00:00:01",
532		"in_port": 2
533		},
534		"eth": {"dl_vlan": 100},
535		}
536		}
537		]
538
539		stored_flow = {
540		"id": 1,
541		"table_id": 0,
542		"cookie": 84114964,
543		"hard_timeout": 0,
544		"idle_timeout": 0,
545		"priority": 10,
546		}
547
548		mock_stored_flows.return_value = {
549		"00:00:00:00:00:00:00:01": [stored_flow]
550		}
551		mock_json = MagicMock()
552		mock_json.json.return_value = {
553		"00:00:00:00:00:00:00:01": [stored_flow]
554		}
555		mock_request_get.get.return_value = mock_json
556
557		response = api.put(
558		url, data=json.dumps(payload), content_type="application/json"
559		)
560		current_data = json.loads(response.data)
561		result = current_data["00:00:00:00:00:00:00:01"]
562
563		assert len(current_data) == 1
564
565		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
566		assert result[0][0]["port"] == 1
567		assert result[0][0]["type"] == "starting"
568		assert result[0][0]["vlan"] == 100
569
570		assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:01"
571		assert result[1][0]["port"] == 2
572		assert result[1][0]["type"] == "starting"
573		assert result[1][0]["vlan"] == 100
574
575		@patch("napps.amlight.sdntrace_cp.utils.get_stored_flows")
576		@patch("napps.amlight.sdntrace_cp.utils.requests")
577		def test_traces_twice(self, mock_request_get, mock_stored_flows):
578		"""Test traces rest call for two equal traces."""
579		api = get_test_client(get_controller_mock(), self.napp)
580		url = f"{self.server_name_url}/traces/"
581
582		payload = [
583		{
584		"trace": {
585		"switch": {
586		"dpid": "00:00:00:00:00:00:00:01",
587		"in_port": 1
588		},
589		"eth": {"dl_vlan": 100},
590		}
591		},
592		{
593		"trace": {
594		"switch": {
595		"dpid": "00:00:00:00:00:00:00:01",
596		"in_port": 1
597		},
598		"eth": {"dl_vlan": 100},
599		}
600		}
601		]
602		stored_flow = {
603		"id": 1,
604		"table_id": 0,
605		"cookie": 84114964,
606		"hard_timeout": 0,
607		"idle_timeout": 0,
608		"priority": 10,
609		}
610
611		mock_stored_flows.return_value = {
612		"00:00:00:00:00:00:00:01": [stored_flow]
613		}
614		mock_json = MagicMock()
615		mock_json.json.return_value = {
616		"00:00:00:00:00:00:00:01": [stored_flow]
617		}
618		mock_request_get.get.return_value = mock_json
619
620		response = api.put(
621		url, data=json.dumps(payload), content_type="application/json"
622		)
623		current_data = json.loads(response.data)
624		result = current_data["00:00:00:00:00:00:00:01"]
625
626		assert len(current_data) == 1
627		assert len(result) == 1
628
629		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
630		assert result[0][0]["port"] == 1
631		assert result[0][0]["type"] == "starting"
632		assert result[0][0]["vlan"] == 100
633

kytos-ng / sdntrace_cp

Pull Request — master (#59)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like