build.tests.unit.tracing.test_trace_manager.TestTraceManager.test_is_entry_missing_dpid() - Code Metrics - Inspection of "Create unit test" - kytos-ng/sdntrace - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#20)

by Rogerio

created 2022-07-12 02:38 UTC

TestTraceManager.test_is_entry_missing_dpid() A

↳ Parent: build.tests.unit.tracing.test_trace_manager

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

"""
    Test tracing.trace_manager
"""
import time
from unittest import TestCase
from unittest.mock import patch, MagicMock

from napps.amlight.sdntrace import settings
from napps.amlight.sdntrace.tracing.trace_manager import TraceManager
from napps.amlight.sdntrace.tracing.trace_entries import TraceEntries
from napps.amlight.sdntrace.shared.switches import Switches

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


# pylint: disable=protected-access
class TestTraceManager(TestCase):
    """Unit tests for tracing.trace_manager.TraceManager"""

    def setUp(self):
        self.create_basic_switches(get_controller_mock())

        # 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()

        self.addCleanup(patch.stopall)
        self.trace_manager = TraceManager(controller=get_controller_mock())

    def tearDown(self) -> None:
        self.trace_manager.stop_traces()
        return super().tearDown()

    @classmethod

    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()._switches = controller.switches

    def test_is_entry_invalid(self):
        """Test if the entry request does not have a valid switch."""
        eth = {"dl_vlan": 100}
        dpid = {"dpid": "a", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        entry = self.trace_manager.is_entry_valid(entries)

        self.assertEqual(entry, "Unknown Switch")

    def test_is_entry_empty_dpid(self):
        """Test if the entry request does not have a valid switch."""
        eth = {"dl_vlan": 100}
        dpid = {"dpid": "", "in_port": 1}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        entry = self.trace_manager.is_entry_valid(entries)

        self.assertEqual(
            entry, "Error: dpid allows [a-f], int, and :. Lengths: 1-16 and 23"
        )

    def test_is_entry_missing_dpid(self):
        """Test if the entry request with missing dpid."""
        eth = {"dl_vlan": 100}
        dpid = {}
        switch = {"switch": dpid, "eth": eth}
        entries = {"trace": switch}
        entry = self.trace_manager.is_entry_valid(entries)

        self.assertEqual(entry, "Error: dpid not provided")

    def test_is_entry_invalid_not_colored(self):
        """Test if the entry request does not have a valid color."""
        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}
        entry = self.trace_manager.is_entry_valid(entries)
        self.assertEqual(entry, "Switch not Colored")

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    def test_is_entry_valid(self, mock_colors):
        """Test if the entry request is valid."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }

        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}
        entry = self.trace_manager.is_entry_valid(entries)
        self.assertEqual(entry.dpid, "00:00:00:00:00:00:00:01")

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

    def test_new_trace(self, mock_colors):
        """Test trace manager new trace creation."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }

        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)
        self.assertIsInstance(trace_entries, TraceEntries)

        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30001)

        # new_trace does not check duplicated request.
        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30002)

    def test_get_id(self):
        """Test trace manager ID control."""
        trace_id = self.trace_manager.get_id()
        self.assertEqual(trace_id, 30001)

        trace_id = self.trace_manager.get_id()
        self.assertEqual(trace_id, 30002)

        trace_id = self.trace_manager.get_id()
        self.assertEqual(trace_id, 30003)

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

    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    def test_trace_pending(self, mock_send_probe, mock_colors):
        """Test trace manager tracing request."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }
        mock_send_probe.return_value = {
            "dpid": "00:00:00:00:00:00:00:01",
            "port": 1,
        }, ""

        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)
        self.assertIsInstance(trace_entries, TraceEntries)

        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30001)

        pending = self.trace_manager.number_pending_requests()
        self.assertEqual(pending, 1)

        result = self.trace_manager.get_result(trace_id)
        self.assertEqual(result, {"msg": "trace pending"})

    def test_request_invalid_trace_id(self):
        """Test trace manager tracing request."""
        result = self.trace_manager.get_result("1234")
        self.assertEqual(result, {"msg": "unknown trace id"})

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

    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    def test_trace_in_process(self, mock_send_probe, mock_colors):
        """Test trace manager tracing request and processing."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }
        mock_send_probe.return_value = {
            "dpid": "00:00:00:00:00:00:00:01",
            "port": 1,
        }, ""

        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)
        self.assertIsInstance(trace_entries, TraceEntries)

        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30001)

        pending = self.trace_manager.number_pending_requests()
        self.assertEqual(pending, 1)

        while pending == 1:
            result = self.trace_manager.get_result(trace_id)
            pending = self.trace_manager.number_pending_requests()
            time.sleep(0.1)

        self.assertEqual(result, {"msg": "trace in process"})


    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    def test_get_result(self, mock_trace_loop, mock_send_probe, mock_colors):
        """Test tracemanager tracing request and resultS."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }
        mock_send_probe.return_value = {
            "dpid": "00:00:00:00:00:00:00:01",
            "port": 1,
        }, ""
        mock_trace_loop.return_value = True

        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)
        trace_id = self.trace_manager.new_trace(trace_entries)
        pending = self.trace_manager.number_pending_requests()

        # Waiting thread to start processing the request
        count = 0
        while pending == 1:
            result = self.trace_manager.get_result(trace_id)
            pending = self.trace_manager.number_pending_requests()
            time.sleep(0.1)
            count += 1
            if count > 30:
                self.fail("Timeout waiting to start processing trace")
                break

        count = 0
        result = self.trace_manager.get_result(trace_id)
        # Waiting thread to process the request
        while "msg" in result and result["msg"] == "trace in process":
            result = self.trace_manager.get_result(trace_id)
            time.sleep(0.1)
            count += 1
            if count > 30:
                self.fail("Timeout waiting to process trace")
                break

        self.assertEqual(result["request_id"], 30001)
        self.assertEqual(result["result"][0]["type"], "starting")
        self.assertEqual(
            result["result"][0]["dpid"], "00:00:00:00:00:00:00:01"
        )
        self.assertEqual(result["result"][0]["port"], 1)
        self.assertIsNotNone(result["result"][0]["time"])
        self.assertIsNotNone(result["start_time"])
        self.assertIsNotNone(result["total_time"])
        self.assertEqual(
            result["request"]["trace"]["switch"]["dpid"],
            "00:00:00:00:00:00:00:01",
        )
        self.assertEqual(result["request"]["trace"]["switch"]["in_port"], 1)

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

    def test_duplicated_request(self, mock_colors):
        """Test trace manager new trace creation."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }

        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)
        self.assertIsInstance(trace_entries, TraceEntries)

        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30001)

        duplicated = self.trace_manager.avoid_duplicated_request(trace_entries)
        self.assertEqual(duplicated, True)

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    def test_avoid_duplicated_request(self, mock_colors):
        """Test trace manager new trace creation."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }

        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)
        self.assertIsInstance(trace_entries, TraceEntries)

        trace_id = self.trace_manager.new_trace(trace_entries)
        self.assertEqual(trace_id, 30001)

        entries["trace"]["switch"]["dpid"] = "00:00:00:00:00:00:00:02"
        trace_entries = self.trace_manager.is_entry_valid(entries)
        self.assertIsInstance(trace_entries, TraceEntries)

        duplicated = self.trace_manager.avoid_duplicated_request(trace_entries)
        self.assertEqual(duplicated, False)

    def test_limit_traces_reached(self):
        """Test trace manager limit for thread processing."""
        # filling the running traces array
        for i in range(settings.PARALLEL_TRACES - 1):
            self.trace_manager._running_traces[i] = i
            is_limit = self.trace_manager.limit_traces_reached()
            self.assertFalse(is_limit)

        self.trace_manager._running_traces[settings.PARALLEL_TRACES] = 9999
        is_limit = self.trace_manager.limit_traces_reached()
        self.assertTrue(is_limit)

    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath")
    def test_spawn_trace(self, mock_tracepath):
        """Test spawn trace."""
        # mock_tracepath
        trace_id = 0
        trace_entries = MagicMock()

        self.trace_manager._running_traces[0] = 9999

        self.trace_manager._spawn_trace(trace_id, trace_entries)

        mock_tracepath.assert_called_once()
        self.assertEqual(len(self.trace_manager._running_traces), 0)


class TestTraceManagerTheadTest(TestCase):
    """Now, load all entries at once"""

    def setUp(self):

        # 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
        self.addCleanup(patch.stopall)

        self.trace_manager = TraceManager(controller=get_controller_mock())
        self.trace_manager.stop_traces()

        self.count_running_traces = 0

    def run_trace_once(self):
        """function to replace the <flag> in "while <flag>()" code
        to run the loop just once."""
        if self.count_running_traces == 0:
            self.count_running_traces = self.count_running_traces + 1
            return True
        return False

    @patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
    @patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
    def test_run_traces(self, mock_trace_loop, mock_send_probe, mock_colors):
        """Test tracemanager tracing request and results."""
        mock_colors.return_value = {
            "color_field": "dl_src",
            "color_value": "ee:ee:ee:ee:ee:01",
        }
        mock_send_probe.return_value = {
            "dpid": "00:00:00:00:00:00:00:01",
            "port": 1,
        }, ""
        mock_trace_loop.return_value = True

        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)

        self.trace_manager._request_queue[1] = trace_entries

        with patch.object(
            TraceManager, "is_tracing_running", side_effect=self.run_trace_once
        ):
            self.trace_manager._run_traces(0.5)

        self.assertEqual(len(self.trace_manager._request_queue), 0)
        self.assertEqual(self.trace_manager.number_pending_requests(), 0)


1		"""
2		Test tracing.trace_manager
3		"""
4		import time
5		from unittest import TestCase
6		from unittest.mock import patch, MagicMock
7
8		from napps.amlight.sdntrace import settings
9		from napps.amlight.sdntrace.tracing.trace_manager import TraceManager
10		from napps.amlight.sdntrace.tracing.trace_entries import TraceEntries
11		from napps.amlight.sdntrace.shared.switches import Switches
12
13		from kytos.lib.helpers import (
14		get_interface_mock,
15		get_link_mock,
16		get_switch_mock,
17		get_controller_mock,
18		)
19
20
21		# pylint: disable=protected-access
22		class TestTraceManager(TestCase):
23		"""Unit tests for tracing.trace_manager.TraceManager"""
24
25		def setUp(self):
26		self.create_basic_switches(get_controller_mock())
27
28		# The decorator run_on_thread is patched, so methods that listen
29		# for events do not run on threads while tested.
30		# Decorators have to be patched before the methods that are
31		# decorated with them are imported.
32		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
33
34		self.addCleanup(patch.stopall)
35		self.trace_manager = TraceManager(controller=get_controller_mock())
36
37		def tearDown(self) -> None:
38		self.trace_manager.stop_traces()
39		return super().tearDown()
40
41	View Code Duplication	@classmethod
		0 ignored issues – show Duplication introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
42		def create_basic_switches(cls, controller):
43		"""Create basic mock switches for Kytos controller."""
44		dpid_a = "00:00:00:00:00:00:00:01"
45		dpid_b = "00:00:00:00:00:00:00:02"
46
47		mock_switch_a = get_switch_mock(dpid_a, 0x04)
48		mock_switch_b = get_switch_mock(dpid_b, 0x04)
49		mock_interface_a = get_interface_mock("s1-eth1", 1, mock_switch_a)
50		mock_interface_b = get_interface_mock("s2-eth1", 1, mock_switch_b)
51
52		mock_link = get_link_mock(mock_interface_a, mock_interface_b)
53		mock_link.id = "cf0f4071be4"
54		mock_switch_a.id = dpid_a
55		mock_switch_a.as_dict.return_value = {"metadata": {}}
56		mock_switch_b.id = dpid_b
57		mock_switch_b.as_dict.return_value = {"metadata": {}}
58
59		controller.switches = {dpid_a: mock_switch_a, dpid_b: mock_switch_b}
60
61		Switches()._switches = controller.switches
62
63		def test_is_entry_invalid(self):
64		"""Test if the entry request does not have a valid switch."""
65		eth = {"dl_vlan": 100}
66		dpid = {"dpid": "a", "in_port": 1}
67		switch = {"switch": dpid, "eth": eth}
68		entries = {"trace": switch}
69		entry = self.trace_manager.is_entry_valid(entries)
70
71		self.assertEqual(entry, "Unknown Switch")
72
73		def test_is_entry_empty_dpid(self):
74		"""Test if the entry request does not have a valid switch."""
75		eth = {"dl_vlan": 100}
76		dpid = {"dpid": "", "in_port": 1}
77		switch = {"switch": dpid, "eth": eth}
78		entries = {"trace": switch}
79		entry = self.trace_manager.is_entry_valid(entries)
80
81		self.assertEqual(
82		entry, "Error: dpid allows [a-f], int, and :. Lengths: 1-16 and 23"
83		)
84
85		def test_is_entry_missing_dpid(self):
86		"""Test if the entry request with missing dpid."""
87		eth = {"dl_vlan": 100}
88		dpid = {}
89		switch = {"switch": dpid, "eth": eth}
90		entries = {"trace": switch}
91		entry = self.trace_manager.is_entry_valid(entries)
92
93		self.assertEqual(entry, "Error: dpid not provided")
94
95		def test_is_entry_invalid_not_colored(self):
96		"""Test if the entry request does not have a valid color."""
97		eth = {"dl_vlan": 100}
98		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
99		switch = {"switch": dpid, "eth": eth}
100		entries = {"trace": switch}
101		entry = self.trace_manager.is_entry_valid(entries)
102		self.assertEqual(entry, "Switch not Colored")
103
104		@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
105		def test_is_entry_valid(self, mock_colors):
106		"""Test if the entry request is valid."""
107		mock_colors.return_value = {
108		"color_field": "dl_src",
109		"color_value": "ee:ee:ee:ee:ee:01",
110		}
111
112		eth = {"dl_vlan": 100}
113		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
114		switch = {"switch": dpid, "eth": eth}
115		entries = {"trace": switch}
116		entry = self.trace_manager.is_entry_valid(entries)
117		self.assertEqual(entry.dpid, "00:00:00:00:00:00:00:01")
118
119	View Code Duplication	@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
		0 ignored issues – show Duplication introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
120		def test_new_trace(self, mock_colors):
121		"""Test trace manager new trace creation."""
122		mock_colors.return_value = {
123		"color_field": "dl_src",
124		"color_value": "ee:ee:ee:ee:ee:01",
125		}
126
127		eth = {"dl_vlan": 100}
128		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
129		switch = {"switch": dpid, "eth": eth}
130		entries = {"trace": switch}
131
132		trace_entries = self.trace_manager.is_entry_valid(entries)
133		self.assertIsInstance(trace_entries, TraceEntries)
134
135		trace_id = self.trace_manager.new_trace(trace_entries)
136		self.assertEqual(trace_id, 30001)
137
138		# new_trace does not check duplicated request.
139		trace_id = self.trace_manager.new_trace(trace_entries)
140		self.assertEqual(trace_id, 30002)
141
142		def test_get_id(self):
143		"""Test trace manager ID control."""
144		trace_id = self.trace_manager.get_id()
145		self.assertEqual(trace_id, 30001)
146
147		trace_id = self.trace_manager.get_id()
148		self.assertEqual(trace_id, 30002)
149
150		trace_id = self.trace_manager.get_id()
151		self.assertEqual(trace_id, 30003)
152
153	View Code Duplication	@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
		0 ignored issues – show Duplication introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
154		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
155		def test_trace_pending(self, mock_send_probe, mock_colors):
156		"""Test trace manager tracing request."""
157		mock_colors.return_value = {
158		"color_field": "dl_src",
159		"color_value": "ee:ee:ee:ee:ee:01",
160		}
161		mock_send_probe.return_value = {
162		"dpid": "00:00:00:00:00:00:00:01",
163		"port": 1,
164		}, ""
165
166		eth = {"dl_vlan": 100}
167		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
168		switch = {"switch": dpid, "eth": eth}
169		entries = {"trace": switch}
170
171		trace_entries = self.trace_manager.is_entry_valid(entries)
172		self.assertIsInstance(trace_entries, TraceEntries)
173
174		trace_id = self.trace_manager.new_trace(trace_entries)
175		self.assertEqual(trace_id, 30001)
176
177		pending = self.trace_manager.number_pending_requests()
178		self.assertEqual(pending, 1)
179
180		result = self.trace_manager.get_result(trace_id)
181		self.assertEqual(result, {"msg": "trace pending"})
182
183		def test_request_invalid_trace_id(self):
184		"""Test trace manager tracing request."""
185		result = self.trace_manager.get_result("1234")
186		self.assertEqual(result, {"msg": "unknown trace id"})
187
188	View Code Duplication	@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
		0 ignored issues – show Duplication introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
189		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
190		def test_trace_in_process(self, mock_send_probe, mock_colors):
191		"""Test trace manager tracing request and processing."""
192		mock_colors.return_value = {
193		"color_field": "dl_src",
194		"color_value": "ee:ee:ee:ee:ee:01",
195		}
196		mock_send_probe.return_value = {
197		"dpid": "00:00:00:00:00:00:00:01",
198		"port": 1,
199		}, ""
200
201		eth = {"dl_vlan": 100}
202		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
203		switch = {"switch": dpid, "eth": eth}
204		entries = {"trace": switch}
205
206		trace_entries = self.trace_manager.is_entry_valid(entries)
207		self.assertIsInstance(trace_entries, TraceEntries)
208
209		trace_id = self.trace_manager.new_trace(trace_entries)
210		self.assertEqual(trace_id, 30001)
211
212		pending = self.trace_manager.number_pending_requests()
213		self.assertEqual(pending, 1)
214
215		while pending == 1:
216		result = self.trace_manager.get_result(trace_id)
217		pending = self.trace_manager.number_pending_requests()
218		time.sleep(0.1)
219
220		self.assertEqual(result, {"msg": "trace in process"})
		0 ignored issues – show introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report The variable `result` does not seem to be defined in case the `while` loop on line `215` is not entered. Are you sure this can never be the case? Loading history...
221
222		@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
223		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
224		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
225		def test_get_result(self, mock_trace_loop, mock_send_probe, mock_colors):
226		"""Test tracemanager tracing request and resultS."""
227		mock_colors.return_value = {
228		"color_field": "dl_src",
229		"color_value": "ee:ee:ee:ee:ee:01",
230		}
231		mock_send_probe.return_value = {
232		"dpid": "00:00:00:00:00:00:00:01",
233		"port": 1,
234		}, ""
235		mock_trace_loop.return_value = True
236
237		eth = {"dl_vlan": 100}
238		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
239		switch = {"switch": dpid, "eth": eth}
240		entries = {"trace": switch}
241
242		trace_entries = self.trace_manager.is_entry_valid(entries)
243		trace_id = self.trace_manager.new_trace(trace_entries)
244		pending = self.trace_manager.number_pending_requests()
245
246		# Waiting thread to start processing the request
247		count = 0
248		while pending == 1:
249		result = self.trace_manager.get_result(trace_id)
250		pending = self.trace_manager.number_pending_requests()
251		time.sleep(0.1)
252		count += 1
253		if count > 30:
254		self.fail("Timeout waiting to start processing trace")
255		break
256
257		count = 0
258		result = self.trace_manager.get_result(trace_id)
259		# Waiting thread to process the request
260		while "msg" in result and result["msg"] == "trace in process":
261		result = self.trace_manager.get_result(trace_id)
262		time.sleep(0.1)
263		count += 1
264		if count > 30:
265		self.fail("Timeout waiting to process trace")
266		break
267
268		self.assertEqual(result["request_id"], 30001)
269		self.assertEqual(result["result"][0]["type"], "starting")
270		self.assertEqual(
271		result["result"][0]["dpid"], "00:00:00:00:00:00:00:01"
272		)
273		self.assertEqual(result["result"][0]["port"], 1)
274		self.assertIsNotNone(result["result"][0]["time"])
275		self.assertIsNotNone(result["start_time"])
276		self.assertIsNotNone(result["total_time"])
277		self.assertEqual(
278		result["request"]["trace"]["switch"]["dpid"],
279		"00:00:00:00:00:00:00:01",
280		)
281		self.assertEqual(result["request"]["trace"]["switch"]["in_port"], 1)
282
283	View Code Duplication	@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
		0 ignored issues – show Duplication introduced 2022-06-20 17:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
284		def test_duplicated_request(self, mock_colors):
285		"""Test trace manager new trace creation."""
286		mock_colors.return_value = {
287		"color_field": "dl_src",
288		"color_value": "ee:ee:ee:ee:ee:01",
289		}
290
291		eth = {"dl_vlan": 100}
292		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
293		switch = {"switch": dpid, "eth": eth}
294		entries = {"trace": switch}
295
296		trace_entries = self.trace_manager.is_entry_valid(entries)
297		self.assertIsInstance(trace_entries, TraceEntries)
298
299		trace_id = self.trace_manager.new_trace(trace_entries)
300		self.assertEqual(trace_id, 30001)
301
302		duplicated = self.trace_manager.avoid_duplicated_request(trace_entries)
303		self.assertEqual(duplicated, True)
304
305		@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
306		def test_avoid_duplicated_request(self, mock_colors):
307		"""Test trace manager new trace creation."""
308		mock_colors.return_value = {
309		"color_field": "dl_src",
310		"color_value": "ee:ee:ee:ee:ee:01",
311		}
312
313		eth = {"dl_vlan": 100}
314		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
315		switch = {"switch": dpid, "eth": eth}
316		entries = {"trace": switch}
317
318		trace_entries = self.trace_manager.is_entry_valid(entries)
319		self.assertIsInstance(trace_entries, TraceEntries)
320
321		trace_id = self.trace_manager.new_trace(trace_entries)
322		self.assertEqual(trace_id, 30001)
323
324		entries["trace"]["switch"]["dpid"] = "00:00:00:00:00:00:00:02"
325		trace_entries = self.trace_manager.is_entry_valid(entries)
326		self.assertIsInstance(trace_entries, TraceEntries)
327
328		duplicated = self.trace_manager.avoid_duplicated_request(trace_entries)
329		self.assertEqual(duplicated, False)
330
331		def test_limit_traces_reached(self):
332		"""Test trace manager limit for thread processing."""
333		# filling the running traces array
334		for i in range(settings.PARALLEL_TRACES - 1):
335		self.trace_manager._running_traces[i] = i
336		is_limit = self.trace_manager.limit_traces_reached()
337		self.assertFalse(is_limit)
338
339		self.trace_manager._running_traces[settings.PARALLEL_TRACES] = 9999
340		is_limit = self.trace_manager.limit_traces_reached()
341		self.assertTrue(is_limit)
342
343		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath")
344		def test_spawn_trace(self, mock_tracepath):
345		"""Test spawn trace."""
346		# mock_tracepath
347		trace_id = 0
348		trace_entries = MagicMock()
349
350		self.trace_manager._running_traces[0] = 9999
351
352		self.trace_manager._spawn_trace(trace_id, trace_entries)
353
354		mock_tracepath.assert_called_once()
355		self.assertEqual(len(self.trace_manager._running_traces), 0)
356
357
358		class TestTraceManagerTheadTest(TestCase):
359		"""Now, load all entries at once"""
360
361		def setUp(self):
362
363		# The decorator run_on_thread is patched, so methods that listen
364		# for events do not run on threads while tested.
365		# Decorators have to be patched before the methods that are
366		# decorated with them are imported.
367		patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
368		# pylint: disable=import-outside-toplevel
369		self.addCleanup(patch.stopall)
370
371		self.trace_manager = TraceManager(controller=get_controller_mock())
372		self.trace_manager.stop_traces()
373
374		self.count_running_traces = 0
375
376		def run_trace_once(self):
377		"""function to replace the <flag> in "while <flag>()" code
378		to run the loop just once."""
379		if self.count_running_traces == 0:
380		self.count_running_traces = self.count_running_traces + 1
381		return True
382		return False
383
384		@patch("napps.amlight.sdntrace.shared.colors.Colors.get_switch_color")
385		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.send_trace_probe")
386		@patch("napps.amlight.sdntrace.tracing.tracer.TracePath.tracepath_loop")
387		def test_run_traces(self, mock_trace_loop, mock_send_probe, mock_colors):
388		"""Test tracemanager tracing request and results."""
389		mock_colors.return_value = {
390		"color_field": "dl_src",
391		"color_value": "ee:ee:ee:ee:ee:01",
392		}
393		mock_send_probe.return_value = {
394		"dpid": "00:00:00:00:00:00:00:01",
395		"port": 1,
396		}, ""
397		mock_trace_loop.return_value = True
398
399		eth = {"dl_vlan": 100}
400		dpid = {"dpid": "00:00:00:00:00:00:00:01", "in_port": 1}
401		switch = {"switch": dpid, "eth": eth}
402		entries = {"trace": switch}
403
404		trace_entries = self.trace_manager.is_entry_valid(entries)
405
406		self.trace_manager._request_queue[1] = trace_entries
407
408		with patch.object(
409		TraceManager, "is_tracing_running", side_effect=self.run_trace_once
410		):
411		self.trace_manager._run_traces(0.5)
412
413		self.assertEqual(len(self.trace_manager._request_queue), 0)
414		self.assertEqual(self.trace_manager.number_pending_requests(), 0)
415

kytos-ng / sdntrace

Pull Request — master (#20)

TestTraceManager.test_is_entry_missing_dpid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like