build.tests.unit.tracing.test_tracer

Complexity

Total Complexity

14

Size/Duplication

Total Lines	632
Duplicated Lines	23.58 %

Test Coverage

Coverage

99.41%

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
B	TestTracePath.test_check_loop_port_different()	0	54	1
A	TestTracePath.setup_method()	0	4	1
B	TestTracePath.test_send_trace_probe()	0	60	1
A	TestTracePath.test_tracepath_loop_timeout()	0	44	1
B	TestTracePath.test_tracepath_loop_with_loop()	67	67	1
A	TestTracePath.test_send_trace_probe_timeout()	0	50	1
A	TestTracePath.test_check_loop_false()	0	46	1
A	TestTracePath.create_basic_switches()	21	21	1
B	TestTracePath.test_tracepath_loop()	61	61	1
A	TestTracePath.commomn_patches()	0	19	2
A	TestTracePath.test_tracepath()	0	45	1
B	TestTracePath.test_tracepath_result()	0	66	1
B	TestTracePath.test_check_loop()	0	58	1

"""
    Test tracing.trace_entries
"""
from unittest.mock import MagicMock, patch
import pytest
from napps.amlight.sdntrace.tracing.trace_msg import TraceMsg
from napps.amlight.sdntrace.tracing.trace_manager import TraceManager
from napps.amlight.sdntrace.tracing.tracer import TracePath
from napps.amlight.sdntrace.shared.switches import Switches
from napps.amlight.sdntrace.tracing.rest import FormatRest

from kytos.lib.helpers import (
    get_interface_mock,
    get_link_mock,
    get_switch_mock,
    get_controller_mock,
)


# pylint: disable=too-many-public-methods, too-many-lines,
# pylint: disable=protected-access, too-many-locals
class TestTracePath:
    """Test all combinations for DPID"""

    @pytest.fixture(autouse=True)
    def commomn_patches(self, request):
        """This function handles setup and cleanup for patches"""
        # This fixture sets up the common patches,
        # and a finalizer is added using addfinalizer to stop
        # the common patches after each test. This ensures that the cleanup
        # is performed after each test, and additional patch decorators
        # can be used within individual test functions.

        patcher = patch("kytos.core.helpers.run_on_thread", lambda x: x)
        mock_patch = patcher.start()

        _ = request.function.__name__

        def cleanup():
            patcher.stop()

        request.addfinalizer(cleanup)
        return mock_patch

    def setup_method(self):
        """Set up before each test method"""
        self.create_basic_switches(get_controller_mock())
        self.trace_manager = TraceManager(controller=get_controller_mock())

    @classmethod

This code seems to be duplicated in your project.

    def create_basic_switches(cls, controller):
        """Create basic mock switches for Kytos controller."""
        dpid_a = "00:00:00:00:00:00:00:01"
        dpid_b = "00:00:00:00:00:00:00:02"

        mock_switch_a = get_switch_mock(dpid_a, 0x04)
        mock_switch_b = get_switch_mock(dpid_b, 0x04)
        mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
        mock_interface_b = get_interface_mock("s2-eth1", 1, mock_switch_b)

        mock_link = get_link_mock(mock_interface_a, mock_interface_b)
        mock_link.id = "cf0f4071be4"
        mock_switch_a.id = dpid_a
        mock_switch_a.as_dict.return_value = {"metadata": {}}
        mock_switch_b.id = dpid_b
        mock_switch_b.as_dict.return_value = {"metadata": {}}

        controller.switches = {dpid_a: mock_switch_a, dpid_b: mock_switch_b}

        Switches(controller.switches)

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    def test_tracepath(
        self, mock_trace_loop, mock_get_switch, mock_color, mock_switch_colors
    ):
        """Test tracepath initial result item. Mocking tracepath loop
        to get just one step."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        switch = MagicMock()
        switch.dpid = "00:00:00:00:00:00:00:01"
        mock_get_switch.return_value = switch

        # Mock tracepath loop to create only the initial result item
        mock_trace_loop.return_value = True

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        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}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)
        tracer.tracepath()

        # Retrieve trace result created from tracepath
        result = self.trace_manager.get_result(trace_id)

        assert result["request_id"] == 111
        assert len(result["result"]) == 1
        assert result["result"][0]["type"] == "starting"
        assert result["result"][0]["dpid"] == dpid["dpid"]
        assert result["result"][0]["port"] == dpid["in_port"]

        assert result["request"]["trace"]["switch"]["dpid"] == dpid["dpid"]
        assert result["request"]["trace"]["switch"]["in_port"] == dpid["in_port"]
        assert result["request"]["trace"]["eth"]["dl_vlan"] == eth["dl_vlan"]
        assert mock_color.call_count == 2
        assert mock_switch_colors.call_count == 2

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    def test_tracepath_result(
        self, mock_trace_loop, mock_get_switch, mock_color, mock_switch_colors
    ):
        """Test tracepath initial result item. Patching the tracepath loop
        to test multiple steps."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        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}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Patch tracepath loop to create a second result item
        # pylint: disable=unused-argument
        def wrap_tracepath_loop(entries, color, switch):
            rest = FormatRest()
            rest.add_trace_step(
                tracer.trace_result,
                trace_type="trace",
                dpid="00:00:00:00:00:00:00:03",
                port=3,
            )
            return True

        mock_trace_loop.side_effect = wrap_tracepath_loop

        # Execute tracepath
        tracer.tracepath()

        # Retrieve trace result created from tracepath
        result = self.trace_manager.get_result(trace_id)

        assert result["request_id"] == 111
        assert len(result["result"]) == 2
        assert result["result"][0]["type"] == "starting"
        assert result["result"][0]["dpid"] == dpid["dpid"]
        assert result["result"][0]["port"] == dpid["in_port"]

        assert result["result"][1]["type"] == "trace"
        assert result["result"][1]["dpid"] == "00:00:00:00:00:00:00:03"
        assert result["result"][1]["port"] == 3

        assert result["request"]["trace"]["switch"]["dpid"] == dpid["dpid"]
        assert result["request"]["trace"]["switch"]["in_port"] == dpid["in_port"]
        assert result["request"]["trace"]["eth"]["dl_vlan"] == eth["dl_vlan"]
        assert mock_color.call_count == 2
        assert mock_switch_colors.call_count == 2

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    @patch("napps.amlight.sdntrace.tracing.tracer.send_packet_out")
    def test_send_trace_probe(
        self,
        mock_send_packet_out,
        mock_trace_loop,
        mock_get_switch,
        mock_color,
        mock_switch_colors,
    ):
        """Test send_trace_probe send and receive."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        mock_send_packet_out.return_value = True

        switch_obj = MagicMock()
        switch_obj.dpid = "00:00:00:00:00:00:00:01"
        mock_get_switch.return_value = switch_obj

        # Mock tracepath loop to create only the initial result item
        mock_trace_loop.return_value = True

        # Creating a fake packet in
        msg = TraceMsg()
        msg.request_id = 111
        pkt_in = {}
        pkt_in["dpid"] = "00:00:00:00:00:00:00:01"
        pkt_in["in_port"] = 1
        pkt_in["msg"] = msg
        pkt_in["ethernet"] = "fake_ethernet_object"
        pkt_in["event"] = "fake_event_object"
        self.trace_manager.trace_pkt_in = [pkt_in]

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        eth = {"dl_vlan": 100, "dl_type": 2048}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        in_port = 1
        probe_pkt = MagicMock()

        result = tracer.send_trace_probe(switch_obj, in_port, probe_pkt)

        mock_send_packet_out.assert_called_once()

        assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"
        assert result[0]["port"] == 1
        assert result[1] == "fake_event_object"
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    @patch("napps.amlight.sdntrace.tracing.tracer.send_packet_out")
    def test_send_trace_probe_timeout(
        self,
        mock_send_packet_out,
        mock_trace_loop,
        mock_get_switch,
        mock_color,
        mock_switch_colors,
    ):
        """Test send_trace_probe with timeout."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"
        mock_send_packet_out.return_value = True

        switch_obj = MagicMock()
        switch_obj.dpid = "00:00:00:00:00:00:00:01"
        mock_get_switch.return_value = switch_obj

        # Mock tracepath loop to create only the initial result item
        mock_trace_loop.return_value = True

        # Creating a fake packet in
        self.trace_manager.trace_pkt_in = []

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        eth = {"dl_vlan": 100, "dl_type": 2048}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        in_port = 1
        probe_pkt = MagicMock()

        result = tracer.send_trace_probe(switch_obj, in_port, probe_pkt)

        assert mock_send_packet_out.call_count == 3

        assert result[0] == "timeout"
        assert result[1] is False
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    def test_check_loop(self, mock_get_switch, mock_color, mock_switch_colors):
        """Test check_loop with loop detection."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        eth = {"dl_vlan": 100, "dl_type": 2048}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Patch tracepath loop to create a second result item
        rest = FormatRest()
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:01",
            port=1,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:02",
            port=2,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:03",
            port=3,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:01",
            port=1,
        )

        result = tracer.check_loop()
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()
        assert result is True

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    def test_check_loop_false(self, mock_get_switch, mock_color, mock_switch_colors):
        """Test check_loop with no loop detection."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        eth = {"dl_vlan": 100, "dl_type": 2048}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Patch tracepath loop to create a second result item
        rest = FormatRest()
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:01",
            port=1,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:02",
            port=2,
        )

        result = tracer.check_loop()
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()
        assert result == 0

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    def test_check_loop_port_different(
        self, mock_get_switch, mock_color, mock_switch_colors
    ):
        """Test check_loop with same switch and different port."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        eth = {"dl_vlan": 100, "dl_type": 2048}
        dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Patch tracepath loop to create a second result item
        rest = FormatRest()
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:01",
            port=1,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:02",
            port=2,
        )
        rest.add_trace_step(
            tracer.trace_result,
            trace_type="trace",
            dpid="00:00:00:00:00:00:00:01",
            port=10,
        )

        result = tracer.check_loop()
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()
        assert result == 0

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")

This code seems to be duplicated in your project.

    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    @patch("napps.amlight.sdntrace.tracing.tracer.prepare_next_packet")
    def test_tracepath_loop(
        self,
        mock_next_packet,
        mock_probe,
        mock_get_switch,
        mock_color,
        mock_switch_colors,
    ):
        """Test tracepath loop method. This test force the return
        after one normal trace."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        mock_probe.return_value = [
            {"dpid": "00:00:00:00:00:00:00:01", "port": 1},
            "fake_event_object",
        ]

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        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}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Mock the next packt to stop the trace loop
        # pylint: disable=unused-argument
        def wrap_next_packet(entries, result, packet_in):
            tracer.trace_ended = True
            return "", "", ""

        mock_next_packet.side_effect = wrap_next_packet

        color = {"color_field": "dl_src", "color_value": "ee:ee:ee:ee:01:2c"}

        # Execute tracepath
        tracer.tracepath_loop(trace_entries, color, switch)
        result = tracer.trace_result

        mock_probe.assert_called_once()
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()
        assert result[0]["type"] == "trace"
        assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    def test_tracepath_loop_timeout(
        self, mock_probe, mock_get_switch, mock_color, mock_switch_colors
    ):
        """Test tracepath loop method finishing with timeout."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        mock_probe.return_value = ["timeout", "fake_event_object"]

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        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}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        color = {"color_field": "dl_src", "color_value": "ee:ee:ee:ee:01:2c"}

        # Execute tracepath
        tracer = TracePath(self.trace_manager, trace_id, trace_entries)
        tracer.tracepath_loop(trace_entries, color, switch)

        result = tracer.trace_result

        mock_probe.assert_called_once()
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()
        assert result[0]["type"] == "last"
        assert result[0]["reason"] == "done"
        assert result[0]["msg"] == "none"

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")

This code seems to be duplicated in your project.

    @patch("napps.amlight.sdntrace.shared.colors.Colors._get_colors")
    @patch("napps.amlight.sdntrace.shared.switches.Switches.get_switch")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    @patch("napps.amlight.sdntrace.tracing.tracer.prepare_next_packet")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.check_loop")
    def test_tracepath_loop_with_loop(
        self,
        mock_check_loop,
        mock_next_packet,
        mock_probe,
        mock_get_switch,
        mock_color,
        mock_switch_colors,
    ):
        """Test tracepath loop method finishing with a loop."""
        mock_switch_colors.return_value = "ee:ee:ee:ee:ee:01"
        mock_check_loop.return_value = True

        # Patch Switches.get_switch
        def wrap_get_switch(dpid):
            switch = MagicMock()
            switch.dpid = dpid
            return switch

        mock_get_switch.side_effect = wrap_get_switch

        mock_probe.return_value = [
            {"dpid": "00:00:00:00:00:00:00:01", "port": 1},
            "fake_event_object",
        ]

        # Trace id to recover the result
        trace_id = 111

        # Creating trace entries
        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}
        trace_entries = self.trace_manager.is_entry_valid(entries)

        tracer = TracePath(self.trace_manager, trace_id, trace_entries)

        # Mock the next packt to stop the trace loop
        # pylint: disable=unused-argument
        def wrap_next_packet(entries, result, packet_in):
            tracer.trace_ended = True
            return "", "", ""

        mock_next_packet.side_effect = wrap_next_packet

        color = {"color_field": "dl_src", "color_value": "ee:ee:ee:ee:01:2c"}

        # Execute tracepath
        tracer.tracepath_loop(trace_entries, color, switch)
        result = tracer.trace_result

        mock_check_loop.assert_called_once()
        mock_next_packet.assert_not_called()
        mock_probe.assert_called_once()
        assert mock_get_switch.call_count == 3
        mock_color.assert_called_once()
        mock_switch_colors.assert_called_once()

        assert result[0]["type"] == "trace"
        assert result[0]["dpid"] == "00:00:00:00:00:00:00:01"