build.tests.unit.test_main.TestMain.setup_method() - Code Metrics - Inspection of "chore: `@rest` endpoints are now run by `starlette..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#94)

by Vinicius

created 2023-04-25 13:49 UTC

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

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	26
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	18
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	18
nop	1
dl	0
loc	26
ccs	18
cts	18
cp	1
crap	2
rs	9.5
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):
        """Test trace rest call."""
        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):
        """Test traces rest call."""
        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):
        """Test traces rest call"""
        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):
        """Test traces rest call"""
        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):
        """Test traces rest call for two traces with different switches."""
        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):
        """Test traces rest call."""
        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):
        """Test traces rest call."""
        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):
320			"""Test trace rest call."""
321	1		payload = {
322			"trace": {
323			"switch": {
324			"dpid": "00:00:00:00:00:00:00:01",
325			"in_port": 1
326			},
327			"eth": {"dl_vlan": 100},
328			}
329			}
330	1		stored_flows = {
331			"flow": {
332			"table_id": 0,
333			"cookie": 84114964,
334			"hard_timeout": 0,
335			"idle_timeout": 0,
336			"priority": 10,
337			"match": {"dl_vlan": 100, "in_port": 1},
338			"actions": [
339			{"action_type": "push_vlan"},
340			{"action_type": "set_vlan", "vlan_id": 200},
341			{"action_type": "output", "port": 2}
342			],
343			},
344			"flow_id": 1,
345			"state": "installed",
346			"switch": "00:00:00:00:00:00:00:01",
347			}
348	1		mock_stored_flows.return_value = {
349			"00:00:00:00:00:00:00:01": [stored_flows]
350			}
351
352	1		resp = await self.api_client.put(self.trace_endpoint, json=payload)
353	1		assert resp.status_code == 200
354	1		current_data = resp.json()
355	1		result = current_data["result"]
356
357	1		assert len(result) == 1
358	1		assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
359	1		assert result[0]["port"] == 1
360	1		assert result[0]["type"] == "last"
361	1		assert result[0]["vlan"] == 100
362	1		assert result[0]["out"] == {"port": 2, "vlan": 200}
363
364	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...
365	1		async def test_get_traces(self, mock_stored_flows):
366			"""Test traces rest call."""
367	1		payload = [{
368			"trace": {
369			"switch": {
370			"dpid": "00:00:00:00:00:00:00:01",
371			"in_port": 1
372			},
373			"eth": {"dl_vlan": 100},
374			}
375			}]
376
377	1		stored_flow = {
378			"id": 1,
379			"flow": {
380			"table_id": 0,
381			"cookie": 84114964,
382			"hard_timeout": 0,
383			"idle_timeout": 0,
384			"priority": 10,
385			"match": {"dl_vlan": 100, "in_port": 1},
386			"actions": [
387			{"action_type": "pop_vlan"},
388			{"action_type": "output", "port": 2},
389			],
390			}
391			}
392
393	1		mock_stored_flows.return_value = {
394			"00:00:00:00:00:00:00:01": [stored_flow]
395			}
396
397	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
398	1		assert resp.status_code == 200
399	1		current_data = resp.json()
400	1		result1 = current_data["result"]
401
402	1		assert len(result1) == 1
403	1		assert len(result1[0]) == 1
404	1		assert result1[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
405	1		assert result1[0][0]["port"] == 1
406	1		assert result1[0][0]["type"] == "last"
407	1		assert result1[0][0]["vlan"] == 100
408	1		assert result1[0][0]["out"] == {"port": 2}
409
410	1		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
411	1		async def test_traces(self, mock_stored_flows):
412			"""Test traces rest call"""
413	1		payload = [
414			{
415			"trace": {
416			"switch": {
417			"dpid": "00:00:00:00:00:00:00:01",
418			"in_port": 1
419			},
420			"eth": {
421			"dl_vlan": 100
422			}
423			}
424			},
425			{
426			"trace": {
427			"switch": {
428			"dpid": "00:00:00:00:00:00:00:0a",
429			"in_port": 1
430			},
431			"eth": {
432			"dl_vlan": 100
433			}
434			}
435			},
436			{
437			"trace": {
438			"switch": {
439			"dpid": "00:00:00:00:00:00:00:02",
440			"in_port": 2
441			},
442			"eth": {
443			"dl_vlan": 100
444			}
445			}
446			}
447			]
448
449	1		stored_flow = {
450			"id": 1,
451			"flow": {
452			"table_id": 0,
453			"cookie": 84114964,
454			"hard_timeout": 0,
455			"idle_timeout": 0,
456			"priority": 10,
457			"match": {"dl_vlan": 100, "in_port": 1},
458			"actions": [{"action_type": "output", "port": 2}],
459			}
460			}
461
462	1		mock_stored_flows.return_value = {
463			"00:00:00:00:00:00:00:01": [stored_flow],
464			"00:00:00:00:00:00:00:02": [stored_flow],
465			}
466
467	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
468	1		assert resp.status_code == 200
469	1		current_data = resp.json()
470	1		result = current_data["result"]
471	1		assert len(result) == 3
472
473	1		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
474	1		assert result[0][0]["port"] == 1
475	1		assert result[0][0]["type"] == "last"
476	1		assert result[0][0]["vlan"] == 100
477	1		assert result[0][0]["out"] == {"port": 2, "vlan": 100}
478
479	1		assert result[1][0]["dpid"] == "00:00:00:00:00:00:00:0a"
480	1		assert result[1][0]["port"] == 1
481	1		assert result[1][0]["type"] == "last"
482	1		assert result[1][0]["vlan"] == 100
483	1		assert result[1][0]["out"] is None
484
485	1		assert result[2][0]["dpid"] == "00:00:00:00:00:00:00:02"
486	1		assert result[2][0]["port"] == 2
487	1		assert result[2][0]["type"] == "incomplete"
488	1		assert result[2][0]["vlan"] == 100
489	1		assert result[2][0]["out"] is None
490
491	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...
492	1		async def test_traces_with_loop(self, mock_stored_flows):
493			"""Test traces rest call"""
494	1		payload = [
495			{
496			"trace": {
497			"switch": {
498			"dpid": "00:00:00:00:00:00:00:01",
499			"in_port": 1
500			},
501			"eth": {
502			"dl_vlan": 100
503			}
504			}
505			}
506			]
507
508	1		stored_flow = {
509			"id": 1,
510			"flow": {
511			"table_id": 0,
512			"cookie": 84114964,
513			"hard_timeout": 0,
514			"idle_timeout": 0,
515			"priority": 10,
516			"match": {"dl_vlan": 100, "in_port": 1},
517			"actions": [{"action_type": "output", "port": 1}],
518			}
519			}
520
521	1		mock_stored_flows.return_value = {
522			"00:00:00:00:00:00:00:01": [stored_flow],
523			}
524
525	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
526	1		assert resp.status_code == 200
527	1		current_data = resp.json()
528	1		result = current_data["result"]
529	1		assert len(result) == 1
530	1		assert result[0][0]["dpid"] == "00:00:00:00:00:00:00:01"
531	1		assert result[0][0]["port"] == 1
532	1		assert result[0][0]["type"] == "loop"
533	1		assert result[0][0]["vlan"] == 100
534	1		assert result[0][0]["out"] == {"port": 1, "vlan": 100}
535
536	1		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
537	1		async def test_traces_no_action(self, mock_stored_flows):
538			"""Test traces rest call for two traces with different switches."""
539	1		payload = [
540			{
541			"trace": {
542			"switch": {
543			"dpid": "00:00:00:00:00:00:00:01",
544			"in_port": 1
545			},
546			"eth": {"dl_vlan": 100},
547			}
548			},
549			{
550			"trace": {
551			"switch": {
552			"dpid": "00:00:00:00:00:00:00:02",
553			"in_port": 1},
554			"eth": {"dl_vlan": 100},
555			}
556			}
557			]
558
559	1		stored_flow1 = {
560			"id": 1,
561			"flow": {
562			"table_id": 0,
563			"cookie": 84114964,
564			"hard_timeout": 0,
565			"idle_timeout": 0,
566			"priority": 10,
567			"match": {"dl_vlan": 100, "in_port": 1}
568			}
569			}
570	1		stored_flow2 = {
571			"id": 1,
572			"flow": {
573			"table_id": 0,
574			"cookie": 84114964,
575			"hard_timeout": 0,
576			"idle_timeout": 0,
577			"priority": 10,
578			"match": {"dl_vlan": 100, "in_port": 1},
579			"actions": []
580			}
581			}
582
583	1		mock_stored_flows.return_value = {
584			"00:00:00:00:00:00:00:01": [stored_flow1],
585			"00:00:00:00:00:00:00:02": [stored_flow2]
586			}
587
588	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
589	1		assert resp.status_code == 200
590	1		current_data = resp.json()
591	1		result = current_data["result"]
592	1		assert len(result) == 2
593	1		assert result[0][-1]['type'] == "incomplete"
594	1		assert result[1][-1]['type'] == "incomplete"
595
596	1		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
597	1		async def test_get_traces_untagged(self, mock_stored_flows):
598			"""Test traces rest call."""
599	1		payload = [{
600			"trace": {
601			"switch": {
602			"dpid": "00:00:00:00:00:00:00:01",
603			"in_port": 1
604			},
605			"eth": {"dl_vlan": 10},
606			}
607			}, {
608			"trace": {
609			"switch": {
610			"dpid": "00:00:00:00:00:00:00:01",
611			"in_port": 1
612			}
613			}
614			}
615			]
616
617	1		stored_flow1 = {
618			"flow": {
619			"match": {"dl_vlan": 0, "in_port": 1},
620			"actions": [
621			{"action_type": "pop_vlan"},
622			{"action_type": "output", "port": 2},
623			],
624			}
625			}
626	1		stored_flow2 = {
627			"flow": {
628			"match": {"in_port": 1},
629			"actions": [
630			{"action_type": "pop_vlan"},
631			{"action_type": "output", "port": 3},
632			],
633			}
634			}
635	1		stored_flow3 = {
636			"flow": {
637			"match": {"dl_vlan": 10, "in_port": 1},
638			"actions": [
639			{"action_type": "pop_vlan"},
640			{"action_type": "output", "port": 1},
641			],
642			}
643			}
644	1		mock_stored_flows.return_value = {
645			"00:00:00:00:00:00:00:01": [
646			stored_flow1,
647			stored_flow2,
648			stored_flow3
649			]
650			}
651
652	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
653	1		assert resp.status_code == 200
654	1		current_data = resp.json()
655
656	1		result = current_data["result"]
657	1		assert result[0][0]["type"] == "last"
658	1		assert result[0][0]["out"] == {"port": 3}
659	1		assert result[1][0]["type"] == "last"
660	1		assert result[1][0]["out"] == {"port": 2}
661
662	1		mock_stored_flows.return_value = {
663			"00:00:00:00:00:00:00:01": [
664			stored_flow3,
665			stored_flow2,
666			stored_flow1
667			]
668			}
669
670	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
671	1		assert resp.status_code == 200
672	1		current_data = resp.json()
673	1		result = current_data["result"]
674	1		assert result[0][0]["type"] == "loop"
675	1		assert result[0][0]["out"] == {"port": 1}
676	1		assert result[1][0]["type"] == "last"
677	1		assert result[1][0]["out"] == {"port": 3}
678
679	1		@patch("napps.amlight.sdntrace_cp.main.get_stored_flows")
680	1		async def test_get_traces_any(self, mock_stored_flows):
681			"""Test traces rest call."""
682	1		payload = [{
683			"trace": {
684			"switch": {
685			"dpid": "00:00:00:00:00:00:00:01",
686			"in_port": 1
687			},
688			"eth": {"dl_vlan": 10}
689			}
690			}, {
691			"trace": {
692			"switch": {
693			"dpid": "00:00:00:00:00:00:00:01",
694			"in_port": 1
695			}
696			}
697			}
698			]
699
700	1		stored_flow1 = {
701			"flow": {
702			"match": {"dl_vlan": "4096/4096", "in_port": 1},
703			"actions": [
704			{"action_type": "pop_vlan"},
705			{"action_type": "output", "port": 2},
706			],
707			}
708			}
709	1		stored_flow2 = {
710			"flow": {
711			"match": {"dl_vlan": 10, "in_port": 1},
712			"actions": [
713			{"action_type": "pop_vlan"},
714			{"action_type": "output", "port": 1},
715			],
716			}
717			}
718	1		stored_flow3 = {
719			"flow": {
720			"match": {"dl_vlan": "10/10", "in_port": 1},
721			"actions": [
722			{"action_type": "pop_vlan"},
723			{"action_type": "output", "port": 3},
724			],
725			}
726			}
727	1		stored_flow4 = {
728			"flow": {
729			"match": {"dl_vlan": "20/10", "in_port": 1},
730			"actions": [
731			{"action_type": "pop_vlan"},
732			{"action_type": "output", "port": 1},
733			],
734			}
735			}
736	1		mock_stored_flows.return_value = {
737			"00:00:00:00:00:00:00:01": [
738			stored_flow1,
739			stored_flow2,
740			stored_flow3,
741			stored_flow4
742			]
743			}
744
745	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
746	1		assert resp.status_code == 200
747	1		current_data = resp.json()
748	1		result = current_data["result"]
749	1		assert result[0][0]["type"] == "last"
750	1		assert result[0][0]["out"] == {"port": 2}
751	1		assert result[1][0]["type"] == "incomplete"
752	1		assert result[1][0]["out"] is None
753
754	1		mock_stored_flows.return_value = {
755			"00:00:00:00:00:00:00:01": [
756			stored_flow4,
757			stored_flow3,
758			stored_flow2,
759			stored_flow1
760			]
761			}
762
763	1		resp = await self.api_client.put(self.traces_endpoint, json=payload)
764	1		assert resp.status_code == 200
765	1		current_data = resp.json()
766	1		result = current_data["result"]
767	1		assert result[0][0]["type"] == "last"
768	1		assert result[0][0]["out"] == {"port": 3}
769	1		assert result[1][0]["type"] == "incomplete"
770			assert result[1][0]["out"] is None
771

kytos-ng / sdntrace_cp

Pull Request — master (#94)

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like