build.tests.unit.test_main.TestMain.test_trace() - Code Metrics - Inspection of "Merge pull request #94 from kytos-ng/chore/starlet..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 58121b...21a39b )

by Vinicius

created 2023-05-03 18:05 UTC

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

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	46
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	16
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	36
nop	3
dl	0
loc	46
ccs	16
cts	16
cp	1
crap	1
rs	9.016
c	0
b	0
f	0

"""Module to test the main napp file."""
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:
    """Test the Main class."""

    def setup_method(self):
        """Execute steps before each tests."""
        # 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

        controller = get_controller_mock()
        self.napp = Main(controller)

        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}
        self.api_client = get_test_client(controller, self.napp)
        self.base_endpoint = "amlight/sdntrace_cp/v1"
        self.trace_endpoint = f"{self.base_endpoint}/trace"
        self.traces_endpoint = f"{self.base_endpoint}/traces"

    @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()
        assert 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()
        assert 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)
        assert 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)
        assert not 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")
    async def test_trace(self, mock_stored_flows, event_loop):
        """Test trace rest call."""
        self.napp.controller.loop = event_loop
        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]
        }

        resp = await self.api_client.put(self.trace_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")

    async def test_get_traces(self, mock_stored_flows, event_loop):
        """Test traces rest call."""
        self.napp.controller.loop = event_loop
        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]
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")
    async def test_traces(self, mock_stored_flows, event_loop):
        """Test traces rest call"""
        self.napp.controller.loop = event_loop
        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],
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")

    async def test_traces_with_loop(self, mock_stored_flows, event_loop):
        """Test traces rest call"""
        self.napp.controller.loop = event_loop
        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],
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")
    async def test_traces_no_action(self, mock_stored_flows, event_loop):
        """Test traces rest call for two traces with different switches."""
        self.napp.controller.loop = event_loop
        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]
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")
    async def test_get_traces_untagged(self, mock_stored_flows, event_loop):
        """Test traces rest call."""
        self.napp.controller.loop = event_loop
        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
            ]
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()

        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
            ]
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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")
    async def test_get_traces_any(self, mock_stored_flows, event_loop):
        """Test traces rest call."""
        self.napp.controller.loop = event_loop
        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
            ]
        }

        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 200
        current_data = resp.json()
        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
            ]
        }

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

kytos-ng / sdntrace_cp

Push — master ( 58121b...21a39b )

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like