build.tests.unit.test_main.TestMain.test_traces_fail() - 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-05-03 21:05 UTC

TestMain.test_traces_fail() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	16
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	1

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	16
ccs	7
cts	7
cp	1
rs	9.8
c	0
b	0
f	0
cc	1
nop	3
crap	1

"""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,
)
from kytos.core.rest_api import HTTPException


# 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.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_trace_fail(self, mock_stored_flows, event_loop):
        """Test trace with a failed dependency."""
        self.napp.controller.loop = event_loop
        mock_stored_flows.side_effect = HTTPException(424, "failed")
        payload = {
            "trace": {
                "switch": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1
                    },
                "eth": {"dl_vlan": 100},
            }
        }
        response = await self.api_client.put(self.trace_endpoint, json=payload)
        assert response.status_code == 424

    @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_fail(self, mock_stored_flows, event_loop):
        """Test traces with a failed dependency."""
        self.napp.controller.loop = event_loop
        mock_stored_flows.side_effect = HTTPException(424, "failed")
        payload = [{
            "trace": {
                "switch": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1
                    },
                "eth": {"dl_vlan": 100},
            }
        }]
        resp = await self.api_client.put(self.traces_endpoint, json=payload)
        assert resp.status_code == 424

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

kytos-ng / sdntrace_cp

Pull Request — master (#95)

TestMain.test_traces_fail() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like