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

Passed

Pull Request — master (#20)

by Rogerio

created 2022-07-19 23:18 UTC

TestTraceManager.tearDown() A

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

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	3
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}

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

kytos-ng / sdntrace

Pull Request — master (#20)

TestTraceManager.tearDown() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like