build.tests.unit.test_main.TestMain.test_update_circuits__no_settings() - Code Metrics - Inspection of "feat: Remove automate in main" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#95)

unknown

created 2023-04-28 19:14 UTC

circuits__no_settings() A

↳ Parent: build.tests.unit.test_main.TestMain.test_trace()

Complexity

Conditions

Size

Total Lines	12
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	2
dl	0
loc	12
rs	10
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/v1"

        # 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/v1/trace/ ",
            ),
            (
                {},
                {"OPTIONS", "HEAD", "PUT"},
                "/api/amlight/sdntrace_cp/v1/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.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"""
        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",
                                "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) == 3

        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][0]["dpid"] == "00:00:00:00:00:00:00:02"
        assert result[2][0]["port"] == 2
        assert result[2][0]["type"] == "incomplete"
        assert result[2][0]["vlan"] == 100
        assert result[2][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"

    @patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
    def test_get_traces_untagged(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": 10},
            }
        }, {
            "trace": {
                "switch": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1
                    }
            }
        }
        ]

        stored_flow1 = {
            "flow": {
                "match": {"dl_vlan": 0, "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 2},
                ],
            }
        }
        stored_flow2 = {
            "flow": {
                "match": {"in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 3},
                ],
            }
        }
        stored_flow3 = {
            "flow": {
                "match": {"dl_vlan": 10, "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 1},
                ],
            }
        }
        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [
                stored_flow1,
                stored_flow2,
                stored_flow3
            ]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["result"]
        assert result[0][0]["type"] == "last"
        assert result[0][0]["out"] == {"port": 3}
        assert result[1][0]["type"] == "last"
        assert result[1][0]["out"] == {"port": 2}

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

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["result"]
        assert result[0][0]["type"] == "loop"
        assert result[0][0]["out"] == {"port": 1}
        assert result[1][0]["type"] == "last"
        assert result[1][0]["out"] == {"port": 3}

    @patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
    def test_get_traces_any(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": 10}
            }
        }, {
            "trace": {
                "switch": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1
                    }
            }
        }
        ]

        stored_flow1 = {
            "flow": {
                "match": {"dl_vlan": "4096/4096", "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 2},
                ],
            }
        }
        stored_flow2 = {
            "flow": {
                "match": {"dl_vlan": 10, "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 1},
                ],
            }
        }
        stored_flow3 = {
            "flow": {
                "match": {"dl_vlan": "10/10", "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 3},
                ],
            }
        }
        stored_flow4 = {
            "flow": {
                "match": {"dl_vlan": "20/10", "in_port": 1},
                "actions": [
                    {"action_type": "pop_vlan"},
                    {"action_type": "output", "port": 1},
                ],
            }
        }
        mock_stored_flows.return_value = {
            "00:00:00:00:00:00:00:01": [
                stored_flow1,
                stored_flow2,
                stored_flow3,
                stored_flow4
            ]
        }

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["result"]
        assert result[0][0]["type"] == "last"
        assert result[0][0]["out"] == {"port": 2}
        assert result[1][0]["type"] == "incomplete"
        assert result[1][0]["out"] is None

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

        response = api.put(
            url, data=json.dumps(payload), content_type="application/json"
        )
        current_data = json.loads(response.data)
        result = current_data["result"]
        assert result[0][0]["type"] == "last"
        assert result[0][0]["out"] == {"port": 3}
        assert result[1][0]["type"] == "incomplete"
        assert result[1][0]["out"] is None


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 \
26		= "http://localhost:8181/api/amlight/sdntrace_cp/v1"
27
28		# The decorator run_on_thread is patched, so methods that listen
29		# for events do not run on threads while tested.
30		# Decorators have to be patched before the methods that are
31		# decorated with them are imported.
32		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
33		# pylint: disable=import-outside-toplevel
34		from napps.amlight.sdntrace_cp.main import Main
35
36		self.napp = Main(get_controller_mock())
37
38		sw1 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
39		intf_sw1 = get_interface_mock("eth1", 1, sw1)
40		intf_sw1.link = None
41		sw1.get_interface_by_port_no.return_value = intf_sw1
42		sw2 = get_switch_mock("00:00:00:00:00:00:00:02", 0x04)
43		intf_sw2 = get_interface_mock("eth1", 1, sw2)
44		intf_sw2.link = None
45		sw2.get_interface_by_port_no.return_value = intf_sw2
46		self.napp.controller.switches = {sw1.dpid: sw1, sw2.dpid: sw2}
47
48		@staticmethod
49		def get_napp_urls(napp):
50		"""Return the amlight/sdntrace_cp urls.
51
52		The urls will be like:
53
54		urls = [
55		(options, methods, url)
56		]
57
58		"""
59		controller = napp.controller
60		controller.api_server.register_napp_endpoints(napp)
61
62		urls = []
63		for rule in controller.api_server.app.url_map.iter_rules():
64		options = {}
65		for arg in rule.arguments:
66		options[arg] = f"[{0}]".format(arg)
67
68		if f"{napp.username}/{napp.name}" in str(rule):
69		urls.append((options, rule.methods, f"{str(rule)}"))
70
71		return urls
72
73		def test_verify_api_urls(self):
74		"""Verify all APIs registered."""
75
76		expected_urls = [
77		(
78		{},
79		{"OPTIONS", "HEAD", "PUT"},
80		"/api/amlight/sdntrace_cp/v1/trace/ ",
81		),
82		(
83		{},
84		{"OPTIONS", "HEAD", "PUT"},
85		"/api/amlight/sdntrace_cp/v1/traces/ ",
86		),
87		]
88		urls = self.get_napp_urls(self.napp)
89		self.assertEqual(len(expected_urls), len(urls))
90
91		@patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
92		def test_trace_step(self, mock_flow_match):
93		"""Test trace_step success result."""
94		mock_flow_match.return_value = ["1"], ["entries"], 1
95		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
96
97		mock_interface = Interface("interface A", 1, MagicMock())
98		mock_interface.address = "00:00:00:00:00:00:00:01"
99
100		iface1 = get_interface_mock(
101		"", 1, get_switch_mock("00:00:00:00:00:00:00:01")
102		)
103		iface2 = get_interface_mock(
104		"", 2, get_switch_mock("00:00:00:00:00:00:00:02")
105		)
106		mock_interface.link = get_link_mock(iface1, iface2)
107		mock_interface.link.endpoint_a.port_number = 1
108		mock_interface.link.endpoint_a.port_number = 2
109
110		# Patch for utils.find_endpoint
111		switch.get_interface_by_port_no.return_value = mock_interface
112
113		entries = MagicMock()
114
115		stored_flows = {
116		"flow": {
117		"table_id": 0,
118		"cookie": 84114964,
119		"hard_timeout": 0,
120		"idle_timeout": 0,
121		"priority": 10,
122		},
123		"flow_id": 1,
124		"state": "installed",
125		"switch": "00:00:00:00:00:00:00:01",
126		}
127
128		stored_flows_arg = {
129		"00:00:00:00:00:00:00:01": [stored_flows]
130		}
131
132		result = self.napp.trace_step(switch, entries, stored_flows_arg)
133
134		mock_flow_match.assert_called_once()
135		self.assertEqual(
136		result,
137		{
138		"dpid": "00:00:00:00:00:00:00:01",
139		"in_port": 2,
140		"out_port": 1,
141		"entries": ["entries"],
142		},
143		)
144
145		@patch("napps.amlight.sdntrace_cp.main.Main.match_and_apply")
146		def test_trace_step__no_endpoint(self, mock_flow_match):
147		"""Test trace_step without endpoints available for switch/port."""
148		mock_flow_match.return_value = ["1"], ["entries"], 1
149		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
150
151		mock_interface = Interface("interface A", 1, MagicMock())
152		mock_interface.address = "00:00:00:00:00:00:00:01"
153		mock_interface.link = None
154
155		# Patch for utils.find_endpoint
156		switch.get_interface_by_port_no.return_value = mock_interface
157
158		entries = MagicMock()
159
160		stored_flows = {
161		"flow": {
162		"table_id": 0,
163		"cookie": 84114964,
164		"hard_timeout": 0,
165		"idle_timeout": 0,
166		"priority": 10,
167		},
168		"flow_id": 1,
169		"state": "installed",
170		"switch": "00:00:00:00:00:00:00:01",
171		}
172
173		stored_flows_arg = {
174		"00:00:00:00:00:00:00:01": [stored_flows]
175		}
176
177		result = self.napp.trace_step(switch, entries, stored_flows_arg)
178
179		mock_flow_match.assert_called_once()
180		self.assertEqual(result, {"entries": ["entries"], "out_port": 1})
181
182		def test_trace_step__no_flow(self):
183		"""Test trace_step without flows for the switch."""
184		switch = get_switch_mock("00:00:00:00:00:00:00:01")
185		entries = MagicMock()
186
187		stored_flows = {
188		"flow": {
189		"table_id": 0,
190		"cookie": 84114964,
191		"hard_timeout": 0,
192		"idle_timeout": 0,
193		"priority": 10,
194		},
195		"flow_id": 1,
196		"state": "installed",
197		"switch": "00:00:00:00:00:00:00:01",
198		}
199
200		stored_flows_arg = {
201		"00:00:00:00:00:00:00:01": [stored_flows]
202		}
203
204		result = self.napp.trace_step(switch, entries, stored_flows_arg)
205		assert result is None
206
207		@patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
208		def test_tracepath(self, mock_trace_step):
209		"""Test tracepath with success result."""
210		eth = {"dl_vlan": 100}
211		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
212		switch = {"switch": dpid, "eth": eth}
213		entries = {"trace": switch}
214		mock_trace_step.return_value = {
215		"dpid": "00:00:00:00:00:00:00:02",
216		"in_port": 2,
217		"out_port": 3,
218		"entries": entries,
219		}
220
221		stored_flows_arg = {
222		"00:00:00:00:00:00:00:01": [],
223		"00:00:00:00:00:00:00:02": [],
224		}
225
226		result = self.napp.tracepath(
227		entries["trace"]["switch"],
228		stored_flows_arg
229		)
230
231		assert result[0]["in"]["dpid"] == "00:00:00:00:00:00:00:01"
232		assert result[0]["in"]["port"] == 1
233		assert result[0]["in"]["type"] == "starting"
234		assert result[0]["out"]["port"] == 3
235
236		assert result[1]["in"]["dpid"] == "00:00:00:00:00:00:00:02"
237		assert result[1]["in"]["port"] == 2
238		assert result[1]["in"]["type"] == "intermediary"
239		assert result[1]["out"]["port"] == 3
240
241		@patch("napps.amlight.sdntrace_cp.main.Main.trace_step")
242		def test_tracepath_loop(self, mock_trace_step):
243		"""Test tracepath with success result."""
244		eth = {"dl_vlan": 100}
245		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
246		switch = {"switch": dpid, "eth": eth}
247		entries = {"trace": switch}
248		mock_trace_step.return_value = {
249		"dpid": "00:00:00:00:00:00:00:01",
250		"in_port": 1,
251		"out_port": 1,
252		"entries": entries,
253		}
254
255		result = self.napp.tracepath(
256		entries["trace"]["switch"],
257		{}
258		)
259		assert len(result) == 2
260		# input interface = output interface
261		assert result[0]["in"]["type"] == "starting"
262		assert result[1]["in"]["type"] == "loop"
263
264		def test_has_loop(self):
265		"""Test has_loop to detect a tracepath with loop."""
266		trace_result = [
267		{
268		"in": {
269		"dpid": "00:00:00:00:00:00:00:01",
270		"port": 2,
271		},
272		"out": {
273		"port": 1,
274		},
275		},
276		{
277		"in": {
278		"dpid": "00:00:00:00:00:00:00:03",
279		"port": 2,
280		},
281		"out": {
282		"port": 1,
283		},
284		},
285		{
286		"in": {
287		"dpid": "00:00:00:00:00:00:00:03",
288		"port": 3,
289		},
290		"out": {
291		"port": 1,
292		},
293		},
294		{
295		"in": {
296		"dpid": "00:00:00:00:00:00:00:03",
297		"port": 3,
298		},
299		"out": {
300		"port": 1,
301		},
302		},
303		]
304		trace_step = {
305		"dpid": "00:00:00:00:00:00:00:03",
306		"port": 3,
307		}
308
309		result = self.napp.has_loop(trace_step, trace_result)
310
311		self.assertTrue(result)
312
313		def test_has_loop__fail(self):
314		"""Test has_loop to detect a tracepath with loop."""
315		trace_result = [
316		{
317		"in": {
318		"dpid": "00:00:00:00:00:00:00:01",
319		"port": 2,
320		},
321		"out": {
322		"port": 1,
323		},
324		},
325		{
326		"in": {
327		"dpid": "00:00:00:00:00:00:00:02",
328		"port": 2,
329		},
330		"out": {
331		"port": 1,
332		},
333		},
334		]
335		trace_step = {
336		"dpid": "00:00:00:00:00:00:00:03",
337		"port": 2,
338		}
339
340		result = self.napp.has_loop(trace_step, trace_result)
341
342		self.assertFalse(result)
343
344		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
345		def test_trace(self, mock_stored_flows):
346		"""Test trace rest call."""
347		api = get_test_client(get_controller_mock(), self.napp)
348		url = f"{self.server_name_url}/trace/"
349
350		payload = {
351		"trace": {
352		"switch": {
353		"dpid": "00:00:00:00:00:00:00:01",
354		"in_port": 1
355		},
356		"eth": {"dl_vlan": 100},
357		}
358		}
359		stored_flows = {
360		"flow": {
361		"table_id": 0,
362		"cookie": 84114964,
363		"hard_timeout": 0,
364		"idle_timeout": 0,
365		"priority": 10,
366		"match": {"dl_vlan": 100, "in_port": 1},
367		"actions": [
368		{"action_type": "push_vlan"},
369		{"action_type": "set_vlan", "vlan_id": 200},
370		{"action_type": "output", "port": 2}
371		],
372		},
373		"flow_id": 1,
374		"state": "installed",
375		"switch": "00:00:00:00:00:00:00:01",
376		}
377		mock_stored_flows.return_value = {
378		"00:00:00:00:00:00:00:01": [stored_flows]
379		}
380
381		response = api.put(
382		url, data=json.dumps(payload), content_type="application/json"
383		)
384		current_data = json.loads(response.data)
385		result = current_data["result"]
386
387		assert len(result) == 1
388		assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
389		assert result[0]["port"] == 1
390		assert result[0]["type"] == "last"
391		assert result[0]["vlan"] == 100
392		assert result[0]["out"] == {"port": 2, "vlan": 200}
393
394	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...
395		def test_get_traces(self, mock_stored_flows):
396		"""Test traces rest call."""
397		api = get_test_client(get_controller_mock(), self.napp)
398		url = f"{self.server_name_url}/traces/"
399
400		payload = [{
401		"trace": {
402		"switch": {
403		"dpid": "00:00:00:00:00:00:00:01",
404		"in_port": 1
405		},
406		"eth": {"dl_vlan": 100},
407		}
408		}]
409
410		stored_flow = {
411		"id": 1,
412		"flow": {
413		"table_id": 0,
414		"cookie": 84114964,
415		"hard_timeout": 0,
416		"idle_timeout": 0,
417		"priority": 10,
418		"match": {"dl_vlan": 100, "in_port": 1},
419		"actions": [
420		{"action_type": "pop_vlan"},
421		{"action_type": "output", "port": 2},
422		],
423		}
424		}
425
426		mock_stored_flows.return_value = {
427		"00:00:00:00:00:00:00:01": [stored_flow]
428		}
429
430		response = api.put(
431		url, data=json.dumps(payload), content_type="application/json"
432		)
433		current_data = json.loads(response.data)
434		result1 = current_data["result"]
435
436		assert len(result1) == 1
437		assert len(result1[0]) == 1
438		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
439		assert result1[0][0]["port"] == 1
440		assert result1[0][0]["type"] == "last"
441		assert result1[0][0]["vlan"] == 100
442		assert result1[0][0]["out"] == {"port": 2}
443
444		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
445		def test_traces(self, mock_stored_flows):
446		"""Test traces rest call"""
447		api = get_test_client(get_controller_mock(), self.napp)
448		url = f"{self.server_name_url}/traces/"
449
450		payload = [
451		{
452		"trace": {
453		"switch": {
454		"dpid": "00:00:00:00:00:00:00:01",
455		"in_port": 1
456		},
457		"eth": {
458		"dl_vlan": 100
459		}
460		}
461		},
462		{
463		"trace": {
464		"switch": {
465		"dpid": "00:00:00:00:00:00:00:0a",
466		"in_port": 1
467		},
468		"eth": {
469		"dl_vlan": 100
470		}
471		}
472		},
473		{
474		"trace": {
475		"switch": {
476		"dpid": "00:00:00:00:00:00:00:02",
477		"in_port": 2
478		},
479		"eth": {
480		"dl_vlan": 100
481		}
482		}
483		}
484		]
485
486		stored_flow = {
487		"id": 1,
488		"flow": {
489		"table_id": 0,
490		"cookie": 84114964,
491		"hard_timeout": 0,
492		"idle_timeout": 0,
493		"priority": 10,
494		"match": {"dl_vlan": 100, "in_port": 1},
495		"actions": [{"action_type": "output", "port": 2}],
496		}
497		}
498
499		mock_stored_flows.return_value = {
500		"00:00:00:00:00:00:00:01": [stored_flow],
501		"00:00:00:00:00:00:00:02": [stored_flow],
502		}
503
504		response = api.put(
505		url, data=json.dumps(payload), content_type="application/json"
506		)
507		current_data = json.loads(response.data)
508		result = current_data["result"]
509		assert len(result) == 3
510
511		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
512		assert result[0][0]["port"] == 1
513		assert result[0][0]["type"] == "last"
514		assert result[0][0]["vlan"] == 100
515		assert result[0][0]["out"] == {"port": 2, "vlan": 100}
516
517		assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:0a"
518		assert result[1][0]["port"] == 1
519		assert result[1][0]["type"] == "last"
520		assert result[1][0]["vlan"] == 100
521		assert result[1][0]["out"] is None
522
523		assert result[2][0]["dpid"] == "00:00:00:00:00:00:00:02"
524		assert result[2][0]["port"] == 2
525		assert result[2][0]["type"] == "incomplete"
526		assert result[2][0]["vlan"] == 100
527		assert result[2][0]["out"] is None
528
529	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...
530		def test_traces_with_loop(self, mock_stored_flows):
531		"""Test traces rest call"""
532		api = get_test_client(get_controller_mock(), self.napp)
533		url = f"{self.server_name_url}/traces/"
534
535		payload = [
536		{
537		"trace": {
538		"switch": {
539		"dpid": "00:00:00:00:00:00:00:01",
540		"in_port": 1
541		},
542		"eth": {
543		"dl_vlan": 100
544		}
545		}
546		}
547		]
548
549		stored_flow = {
550		"id": 1,
551		"flow": {
552		"table_id": 0,
553		"cookie": 84114964,
554		"hard_timeout": 0,
555		"idle_timeout": 0,
556		"priority": 10,
557		"match": {"dl_vlan": 100, "in_port": 1},
558		"actions": [{"action_type": "output", "port": 1}],
559		}
560		}
561
562		mock_stored_flows.return_value = {
563		"00:00:00:00:00:00:00:01": [stored_flow],
564		}
565
566		response = api.put(
567		url, data=json.dumps(payload), content_type="application/json"
568		)
569		current_data = json.loads(response.data)
570		result = current_data["result"]
571		assert len(result) == 1
572		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
573		assert result[0][0]["port"] == 1
574		assert result[0][0]["type"] == "loop"
575		assert result[0][0]["vlan"] == 100
576		assert result[0][0]["out"] == {"port": 1, "vlan": 100}
577
578		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
579		def test_traces_no_action(self, mock_stored_flows):
580		"""Test traces rest call for two traces with different switches."""
581		api = get_test_client(get_controller_mock(), self.napp)
582		url = f"{self.server_name_url}/traces/"
583
584		payload = [
585		{
586		"trace": {
587		"switch": {
588		"dpid": "00:00:00:00:00:00:00:01",
589		"in_port": 1
590		},
591		"eth": {"dl_vlan": 100},
592		}
593		},
594		{
595		"trace": {
596		"switch": {
597		"dpid": "00:00:00:00:00:00:00:02",
598		"in_port": 1},
599		"eth": {"dl_vlan": 100},
600		}
601		}
602		]
603
604		stored_flow1 = {
605		"id": 1,
606		"flow": {
607		"table_id": 0,
608		"cookie": 84114964,
609		"hard_timeout": 0,
610		"idle_timeout": 0,
611		"priority": 10,
612		"match": {"dl_vlan": 100, "in_port": 1}
613		}
614		}
615		stored_flow2 = {
616		"id": 1,
617		"flow": {
618		"table_id": 0,
619		"cookie": 84114964,
620		"hard_timeout": 0,
621		"idle_timeout": 0,
622		"priority": 10,
623		"match": {"dl_vlan": 100, "in_port": 1},
624		"actions": []
625		}
626		}
627
628		mock_stored_flows.return_value = {
629		"00:00:00:00:00:00:00:01": [stored_flow1],
630		"00:00:00:00:00:00:00:02": [stored_flow2]
631		}
632
633		response = api.put(
634		url, data=json.dumps(payload), content_type="application/json"
635		)
636		current_data = json.loads(response.data)
637		result = current_data["result"]
638		assert len(result) == 2
639		assert result[0][-1]['type'] == "incomplete"
640		assert result[1][-1]['type'] == "incomplete"
641
642		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
643		def test_get_traces_untagged(self, mock_stored_flows):
644		"""Test traces rest call."""
645		api = get_test_client(get_controller_mock(), self.napp)
646		url = f"{self.server_name_url}/traces/"
647
648		payload = [{
649		"trace": {
650		"switch": {
651		"dpid": "00:00:00:00:00:00:00:01",
652		"in_port": 1
653		},
654		"eth": {"dl_vlan": 10},
655		}
656		}, {
657		"trace": {
658		"switch": {
659		"dpid": "00:00:00:00:00:00:00:01",
660		"in_port": 1
661		}
662		}
663		}
664		]
665
666		stored_flow1 = {
667		"flow": {
668		"match": {"dl_vlan": 0, "in_port": 1},
669		"actions": [
670		{"action_type": "pop_vlan"},
671		{"action_type": "output", "port": 2},
672		],
673		}
674		}
675		stored_flow2 = {
676		"flow": {
677		"match": {"in_port": 1},
678		"actions": [
679		{"action_type": "pop_vlan"},
680		{"action_type": "output", "port": 3},
681		],
682		}
683		}
684		stored_flow3 = {
685		"flow": {
686		"match": {"dl_vlan": 10, "in_port": 1},
687		"actions": [
688		{"action_type": "pop_vlan"},
689		{"action_type": "output", "port": 1},
690		],
691		}
692		}
693		mock_stored_flows.return_value = {
694		"00:00:00:00:00:00:00:01": [
695		stored_flow1,
696		stored_flow2,
697		stored_flow3
698		]
699		}
700
701		response = api.put(
702		url, data=json.dumps(payload), content_type="application/json"
703		)
704		current_data = json.loads(response.data)
705		result = current_data["result"]
706		assert result[0][0]["type"] == "last"
707		assert result[0][0]["out"] == {"port": 3}
708		assert result[1][0]["type"] == "last"
709		assert result[1][0]["out"] == {"port": 2}
710
711		mock_stored_flows.return_value = {
712		"00:00:00:00:00:00:00:01": [
713		stored_flow3,
714		stored_flow2,
715		stored_flow1
716		]
717		}
718
719		response = api.put(
720		url, data=json.dumps(payload), content_type="application/json"
721		)
722		current_data = json.loads(response.data)
723		result = current_data["result"]
724		assert result[0][0]["type"] == "loop"
725		assert result[0][0]["out"] == {"port": 1}
726		assert result[1][0]["type"] == "last"
727		assert result[1][0]["out"] == {"port": 3}
728
729		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
730		def test_get_traces_any(self, mock_stored_flows):
731		"""Test traces rest call."""
732		api = get_test_client(get_controller_mock(), self.napp)
733		url = f"{self.server_name_url}/traces/"
734
735		payload = [{
736		"trace": {
737		"switch": {
738		"dpid": "00:00:00:00:00:00:00:01",
739		"in_port": 1
740		},
741		"eth": {"dl_vlan": 10}
742		}
743		}, {
744		"trace": {
745		"switch": {
746		"dpid": "00:00:00:00:00:00:00:01",
747		"in_port": 1
748		}
749		}
750		}
751		]
752
753		stored_flow1 = {
754		"flow": {
755		"match": {"dl_vlan": "4096/4096", "in_port": 1},
756		"actions": [
757		{"action_type": "pop_vlan"},
758		{"action_type": "output", "port": 2},
759		],
760		}
761		}
762		stored_flow2 = {
763		"flow": {
764		"match": {"dl_vlan": 10, "in_port": 1},
765		"actions": [
766		{"action_type": "pop_vlan"},
767		{"action_type": "output", "port": 1},
768		],
769		}
770		}
771		stored_flow3 = {
772		"flow": {
773		"match": {"dl_vlan": "10/10", "in_port": 1},
774		"actions": [
775		{"action_type": "pop_vlan"},
776		{"action_type": "output", "port": 3},
777		],
778		}
779		}
780		stored_flow4 = {
781		"flow": {
782		"match": {"dl_vlan": "20/10", "in_port": 1},
783		"actions": [
784		{"action_type": "pop_vlan"},
785		{"action_type": "output", "port": 1},
786		],
787		}
788		}
789		mock_stored_flows.return_value = {
790		"00:00:00:00:00:00:00:01": [
791		stored_flow1,
792		stored_flow2,
793		stored_flow3,
794		stored_flow4
795		]
796		}
797
798		response = api.put(
799		url, data=json.dumps(payload), content_type="application/json"
800		)
801		current_data = json.loads(response.data)
802		result = current_data["result"]
803		assert result[0][0]["type"] == "last"
804		assert result[0][0]["out"] == {"port": 2}
805		assert result[1][0]["type"] == "incomplete"
806		assert result[1][0]["out"] is None
807
808		mock_stored_flows.return_value = {
809		"00:00:00:00:00:00:00:01": [
810		stored_flow4,
811		stored_flow3,
812		stored_flow2,
813		stored_flow1
814		]
815		}
816
817		response = api.put(
818		url, data=json.dumps(payload), content_type="application/json"
819		)
820		current_data = json.loads(response.data)
821		result = current_data["result"]
822		assert result[0][0]["type"] == "last"
823		assert result[0][0]["out"] == {"port": 3}
824		assert result[1][0]["type"] == "incomplete"
825		assert result[1][0]["out"] is None
826

kytos-ng / sdntrace_cp

Pull Request — master (#95)

circuits__no_settings() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like