build.tests.unit.test_automate.TestAutomate.test_run_traces() - Code Metrics - Inspection of "Improve coverage" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#30)

by Rogerio

created 2022-07-12 02:45 UTC

TestAutomate.test_run_traces() B

↳ Parent: build.tests.unit.test_automate

Complexity

Conditions

Size

Total Lines	61
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	40
nop	1
dl	0
loc	61
rs	8.92
c	0
b	0
f	0

How to fix Long Method

"""Module to test the automate.py."""
from unittest import TestCase
from unittest.mock import patch, MagicMock

from napps.amlight.sdntrace_cp.automate import Automate

from kytos.lib.helpers import get_switch_mock


# pylint: disable=too-many-public-methods, duplicate-code, protected-access
class TestAutomate(TestCase):
    """Test claas Automate."""

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")

    def test_get_circuit(self, mock_find_circuits):
        """Verify get circuit success."""
        formatted = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 3,
                "in_vlan": 100,
                "out_port": 2,
                "out_vlan": 200,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "in_port": 3,
                "in_vlan": 200,
                "out_port": 2,
                "out_vlan": 100,
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "in_port": 3,
                "in_vlan": 100,
                "out_port": 2,
                "out_vlan": 200,
            },
        ]

        circuits = []
        circuits.append({"circuit": formatted, "entries": []})

        circuit = {
            "dpid_a": circuits[0]["circuit"][0]["dpid"],
            "port_a": circuits[0]["circuit"][0]["in_port"],
            "vlan_a": circuits[0]["circuit"][0]["in_vlan"],
            "dpid_z": circuits[0]["circuit"][2]["dpid"],
            "port_z": circuits[0]["circuit"][2]["out_port"],
            "vlan_z": circuits[0]["circuit"][2]["out_vlan"],
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = circuits

        result = automate.get_circuit(circuit)

        mock_find_circuits.assert_called_once()
        self.assertEqual(result, formatted)

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")

    def test_get_circuit_not_found(self, mock_find_circuits):
        """Verify get circuit not finding a circuit."""

        formatted = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 3,
                "in_vlan": 100,
                "out_port": 2,
                "out_vlan": 200,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "in_port": 3,
                "in_vlan": 200,
                "out_port": 2,
                "out_vlan": 100,
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "in_port": 3,
                "in_vlan": 100,
                "out_port": 2,
                "out_vlan": 200,
            },
        ]

        circuits = []
        circuits.append({"circuit": formatted, "entries": []})

        circuit = {
            "dpid_a": circuits[0]["circuit"][0]["dpid"],
            "port_a": circuits[0]["circuit"][0]["in_port"],
            "vlan_a": circuits[0]["circuit"][0]["in_vlan"],
            "dpid_z": circuits[0]["circuit"][0]["dpid"],
            "port_z": circuits[0]["circuit"][0]["out_port"],
            "vlan_z": circuits[0]["circuit"][0]["out_vlan"],
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = circuits

        result = automate.get_circuit(circuit)

        mock_find_circuits.assert_called_once()
        self.assertIsNone(result)

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
    def test_get_circuit_empty(self, mock_find_circuits):
        """Verify get circuit with empty circuits"""
        circuit = {
            "dpid_a": "00:00:00:00:00:00:00:01",
            "port_a": 1,
            "vlan_a": 100,
            "dpid_z": "00:00:00:00:00:00:00:03",
            "port_z": 2,
            "vlan_z": 200,
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []

        result = automate.get_circuit(circuit)

        mock_find_circuits.assert_called_once()
        self.assertIsNone(result)

    def test_check_step(self):
        """Verify check_step success."""
        trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
        circuit_step = {}
        circuit_step["dpid"] = trace_step["dpid"]
        circuit_step["in_port"] = trace_step["port"]

        tracer = MagicMock()
        automate = Automate(tracer)
        result = automate.check_step(circuit_step, trace_step)

        self.assertTrue(result)

    def test_check_step_wront_dpid(self):

        """Verify if check_step fail with different dpid."""
        trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
        circuit_step = {}
        circuit_step["dpid"] = "00:00:00:00:00:00:00:02"
        circuit_step["in_port"] = trace_step["port"]

        tracer = MagicMock()
        automate = Automate(tracer)
        result = automate.check_step(circuit_step, trace_step)

        self.assertFalse(result)

    def test_check_step_wrong_port(self):

        """Verify if check_step fail with different port."""
        trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
        circuit_step = {}
        circuit_step["dpid"] = trace_step["dpid"]
        circuit_step["in_port"] = 2

        tracer = MagicMock()
        automate = Automate(tracer)
        result = automate.check_step(circuit_step, trace_step)

        self.assertFalse(result)

    @patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
    def test_check_trace(self, mock_find_circuits, mock_get_circuit):
        """Verify _check_trace with trace finding a valid circuit."""
        circuit_steps = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "in_port": 3,
            },
        ]
        mock_get_circuit.return_value = circuit_steps

        trace = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
            },
        ]

        circuit = {
            "dpid_a": "00:00:00:00:00:00:00:01",
            "port_a": 1,
            "dpid_z": "00:00:00:00:00:00:00:03",
            "port_z": 2,
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []
        result = automate._check_trace(circuit, trace)

        mock_find_circuits.assert_called_once()
        mock_get_circuit.assert_called_once()
        self.assertTrue(result)

    # pylint: disable=duplicate-code
    @patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
    def test_check_trace__short_trace(
        self, mock_find_circuits, mock_get_circuit
    ):
        """Verify _check_trace if lenght of circuit steps is different from
        lenght of trace steps"""
        circuit_steps = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "in_port": 3,
            },
        ]
        mock_get_circuit.return_value = circuit_steps

        trace = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
            },
        ]

        circuit = {
            "dpid_a": "00:00:00:00:00:00:00:01",
            "port_a": 1,
            "dpid_z": "00:00:00:00:00:00:00:03",
            "port_z": 2,
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []
        result = automate._check_trace(circuit, trace)

        mock_find_circuits.assert_called_once()
        mock_get_circuit.assert_called_once()
        self.assertFalse(result)

    @patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")

    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
    def test_check_trace__wrong_steps(
        self, mock_find_circuits, mock_get_circuit
    ):
        """Verify _check_trace with circuit steps different
        from trace steps"""
        circuit_steps = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "in_port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:02",
                "in_port": 3,
            },
        ]
        mock_get_circuit.return_value = circuit_steps

        trace = [
            {
                "dpid": "00:00:00:00:00:00:00:01",
                "port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:05",
                "port": 3,
            },
            {
                "dpid": "00:00:00:00:00:00:00:03",
                "port": 3,
            },
        ]

        circuit = {
            "dpid_a": "00:00:00:00:00:00:00:01",
            "port_a": 1,
            "dpid_z": "00:00:00:00:00:00:00:03",
            "port_z": 2,
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []
        result = automate._check_trace(circuit, trace)

        mock_find_circuits.assert_called_once()
        mock_get_circuit.assert_called_once()
        self.assertFalse(result)

    @patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")
    @patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
    def test_check_trace__no_steps(self, mock_find_circuits, mock_get_circuit):
        """Verify _check_trace with empty circuit steps."""
        circuit_steps = []
        mock_get_circuit.return_value = circuit_steps

        trace = MagicMock()

        circuit = {
            "dpid_a": "00:00:00:00:00:00:00:01",
            "port_a": 1,
            "dpid_z": "00:00:00:00:00:00:00:03",
            "port_z": 2,
        }

        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []
        result = automate._check_trace(circuit, trace)

        mock_find_circuits.assert_called_once()
        mock_get_circuit.assert_called_once()
        self.assertFalse(result)

    def test_run_traces__empty(self):
        """Test run_traces with empty circuits."""
        tracer = MagicMock()
        automate = Automate(tracer)
        automate._circuits = []

        result = automate.run_traces()

        tracer.tracepath.assert_not_called()
        self.assertEqual(result, [])

    def test_run_traces(self):
        """Test run_traces runnin tracepaths for all circuits."""
        trace_result = [
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "port": 1,
                    "time": "2022-06-21 21:32:14.420100",
                    "type": "starting",
                }
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "port": 3,
                    "time": "2022-06-21 21:32:14.420200",
                    "type": "trace",
                    "vlan": 100,
                },
                "out": {"port": 2, "vlan": 200},
            },
        ]

        tracer = MagicMock()
        tracer.tracepath.return_value = trace_result

        automate = Automate(tracer)
        automate._circuits = [
            {
                "circuit": {
                    "dpid_a": "00:00:00:00:00:00:00:01",
                    "port_a": 1,
                    "dpid_z": "00:00:00:00:00:00:00:03",
                    "port_z": 2,
                },
                "entries": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "in_port": 1,
                    "vlan_vid": [100],
                },
            },
            {
                "circuit": {
                    "dpid_a": "00:00:00:00:00:00:00:02",
                    "port_a": 1,
                    "dpid_z": "00:00:00:00:00:00:00:04",
                    "port_z": 2,
                },
                "entries": {
                    "dpid": "00:00:00:00:00:00:00:02",
                    "in_port": 1,
                    "vlan_vid": [100],
                },
            },
        ]

        result = automate.run_traces()

        self.assertEqual(tracer.tracepath.call_count, 2)
        self.assertEqual(result[0], automate._circuits[0])
        self.assertEqual(result[1], automate._circuits[1])

    def test_find_circuits__empty(self):
        """Test find_circuits without switches."""
        tracer = MagicMock()

        automate = Automate(tracer)
        automate.find_circuits()

        self.assertEqual(automate._circuits, [])

    def test_find_circuits(self):
        """Test find_circuits successfully finding circuits
        for all switches."""
        trace_result = [
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "port": 1,
                    "time": "2022-06-21 21:32:14.420100",
                    "type": "starting",
                }
            },
            {
                "in": {
                    "dpid": "00:00:00:00:00:00:00:03",
                    "port": 3,
                    "time": "2022-06-21 21:32:14.420200",
                    "type": "trace",
                    "vlan": 100,
                },
                "out": {"port": 2, "vlan": 200},
            },
        ]
        tracer = MagicMock()
        tracer.tracepath.return_value = trace_result

        automate = Automate(tracer)
        switches = [
            get_switch_mock("00:00:00:00:00:00:00:01", 0x04),
            get_switch_mock("00:00:00:00:00:00:00:02", 0x04),
            get_switch_mock("00:00:00:00:00:00:00:03", 0x04),
            get_switch_mock("00:00:00:00:00:00:00:04", 0x04),
        ]
        for switch in switches:
            flow = MagicMock()
            in_port = MagicMock()
            in_port.value = 1
            flow.match = {"in_port": in_port}
            action = MagicMock()
            action.action_type = "output"
            action.port = 1
            flow.actions = [action]
            flows = [flow]
            switch.generic_flows = flows

        automate._tracer.controller.switches.values.return_value = switches

        automate.find_circuits()

        self.assertIsNotNone(automate._circuits)

        self.assertEqual(len(automate._circuits), 4)
        for item in automate._circuits:
            self.assertEqual(
                item["circuit"][0]["dpid"], trace_result[0]["in"]["dpid"]
            )
            self.assertEqual(
                item["circuit"][0]["in_port"], trace_result[0]["in"]["port"]
            )
            self.assertEqual(
                item["circuit"][1]["dpid"], trace_result[1]["in"]["dpid"]
            )
            self.assertEqual(
                item["circuit"][1]["in_port"], trace_result[1]["in"]["port"]
            )
            self.assertEqual(
                item["circuit"][1]["in_vlan"], trace_result[1]["in"]["vlan"]
            )
            self.assertEqual(
                item["circuit"][1]["out_port"], trace_result[1]["out"]["port"]
            )
            self.assertEqual(
                item["circuit"][1]["out_vlan"], trace_result[1]["out"]["vlan"]
            )

        self.assertEqual(
            automate._circuits[0]["entries"]["trace"]["switch"]["dpid"],
            "00:00:00:00:00:00:00:01",
        )
        self.assertEqual(
            automate._circuits[1]["entries"]["trace"]["switch"]["dpid"],
            "00:00:00:00:00:00:00:02",
        )
        self.assertEqual(
            automate._circuits[2]["entries"]["trace"]["switch"]["dpid"],
            "00:00:00:00:00:00:00:03",
        )
        self.assertEqual(
            automate._circuits[3]["entries"]["trace"]["switch"]["dpid"],
            "00:00:00:00:00:00:00:04",
        )

    @patch("napps.amlight.sdntrace_cp.automate.requests")
    # pylint: disable=no-self-use
    def test_run_important_traces__empty(self, mock_requests):
        """Test run_important_traces with empty circuits to run."""
        tracer = MagicMock()

        automate = Automate(tracer)
        automate.run_important_traces()

        mock_requests.assert_not_called()

    @patch("napps.amlight.sdntrace_cp.automate.requests")
    @patch("napps.amlight.sdntrace_cp.automate.settings")
    def test_run_important_traces(self, mock_settings, mock_requests):
        """Test run_important_traces if control plane trace result is
        different from the data plane trace."""
        mock_settings.IMPORTANT_CIRCUITS = [
            {"dpid_a": "00:00:00:00:00:00:00:01", "port_a": 1, "vlan_a": 100}
        ]

        mock_json = MagicMock()
        mock_json.json.return_value = {
            "result": [{"type": "starting"}, {"type": "last"}]
        }
        mock_requests.get.return_value = mock_json

        def side_effect(event):
            self.assertTrue(
                event.content["message"]["source"], "amlight/sdntrace_cp"
            )
            self.assertTrue(
                event.content["message"]["m_body"],
                "Trace in data plane different from trace in control plane "
                "for circuit {'dpid_a': '00:00:00:00:00:00:00:01', "
                "'port_a': 1, 'vlan_a': 100}"
            )

        tracer = MagicMock()
        tracer.controller.buffers.app.put.side_effect = side_effect

        automate = Automate(tracer)
        automate.run_important_traces()

    @patch("napps.amlight.sdntrace_cp.automate.requests")
    @patch("napps.amlight.sdntrace_cp.automate.settings")
    @patch("napps.amlight.sdntrace_cp.automate.Automate._check_trace")
    # pylint: disable=no-self-use
    def test_run_important_traces__success(
        self, mock_check_trace, mock_settings, mock_requests
    ):
        """Verify run_important_traces if control plane trace result
        is the same result from the data plane trace."""
        # Patch to check trace with stored circuit
        mock_check_trace.return_value = True

        mock_settings.IMPORTANT_CIRCUITS = [
            {
                "dpid_a": "00:00:00:00:00:00:00:01",
                "port_a": 1,
                "vlan_a": 100,
                "dpid_z": "00:00:00:00:00:00:00:02",
                "port_z": 2,
                "vlan_z": 100,
            },
        ]

        mock_json = MagicMock()
        mock_json.json.return_value = {
            "result": [
                {
                    "dpid": "00:00:00:00:00:00:00:01",
                    "port": 1,
                    "type": "starting",
                },
                {"dpid": "00:00:00:00:00:00:00:03", "port": 1, "type": "last"},
            ]
        }
        mock_requests.get.return_value = mock_json

        tracer = MagicMock()
        tracer.controller.buffers.app.put.side_effect = None

        automate = Automate(tracer)
        automate.run_important_traces()

        # Check if important trace dont trigger the event
        # It means that the CP trace is the same to the DP trace
        tracer.controller.buffers.app.put.assert_not_called()


1		"""Module to test the automate.py."""
2		from unittest import TestCase
3		from unittest.mock import patch, MagicMock
4
5		from napps.amlight.sdntrace_cp.automate import Automate
6
7		from kytos.lib.helpers import get_switch_mock
8
9
10		# pylint: disable=too-many-public-methods, duplicate-code, protected-access
11		class TestAutomate(TestCase):
12		"""Test claas Automate."""
13
14	View Code Duplication	@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
15		def test_get_circuit(self, mock_find_circuits):
16		"""Verify get circuit success."""
17		formatted = [
18		{
19		"dpid": "00:00:00:00:00:00:00:01",
20		"in_port": 3,
21		"in_vlan": 100,
22		"out_port": 2,
23		"out_vlan": 200,
24		},
25		{
26		"dpid": "00:00:00:00:00:00:00:02",
27		"in_port": 3,
28		"in_vlan": 200,
29		"out_port": 2,
30		"out_vlan": 100,
31		},
32		{
33		"dpid": "00:00:00:00:00:00:00:03",
34		"in_port": 3,
35		"in_vlan": 100,
36		"out_port": 2,
37		"out_vlan": 200,
38		},
39		]
40
41		circuits = []
42		circuits.append({"circuit": formatted, "entries": []})
43
44		circuit = {
45		"dpid_a": circuits[0]["circuit"][0]["dpid"],
46		"port_a": circuits[0]["circuit"][0]["in_port"],
47		"vlan_a": circuits[0]["circuit"][0]["in_vlan"],
48		"dpid_z": circuits[0]["circuit"][2]["dpid"],
49		"port_z": circuits[0]["circuit"][2]["out_port"],
50		"vlan_z": circuits[0]["circuit"][2]["out_vlan"],
51		}
52
53		tracer = MagicMock()
54		automate = Automate(tracer)
55		automate._circuits = circuits
56
57		result = automate.get_circuit(circuit)
58
59		mock_find_circuits.assert_called_once()
60		self.assertEqual(result, formatted)
61
62	View Code Duplication	@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
63		def test_get_circuit_not_found(self, mock_find_circuits):
64		"""Verify get circuit not finding a circuit."""
65
66		formatted = [
67		{
68		"dpid": "00:00:00:00:00:00:00:01",
69		"in_port": 3,
70		"in_vlan": 100,
71		"out_port": 2,
72		"out_vlan": 200,
73		},
74		{
75		"dpid": "00:00:00:00:00:00:00:02",
76		"in_port": 3,
77		"in_vlan": 200,
78		"out_port": 2,
79		"out_vlan": 100,
80		},
81		{
82		"dpid": "00:00:00:00:00:00:00:03",
83		"in_port": 3,
84		"in_vlan": 100,
85		"out_port": 2,
86		"out_vlan": 200,
87		},
88		]
89
90		circuits = []
91		circuits.append({"circuit": formatted, "entries": []})
92
93		circuit = {
94		"dpid_a": circuits[0]["circuit"][0]["dpid"],
95		"port_a": circuits[0]["circuit"][0]["in_port"],
96		"vlan_a": circuits[0]["circuit"][0]["in_vlan"],
97		"dpid_z": circuits[0]["circuit"][0]["dpid"],
98		"port_z": circuits[0]["circuit"][0]["out_port"],
99		"vlan_z": circuits[0]["circuit"][0]["out_vlan"],
100		}
101
102		tracer = MagicMock()
103		automate = Automate(tracer)
104		automate._circuits = circuits
105
106		result = automate.get_circuit(circuit)
107
108		mock_find_circuits.assert_called_once()
109		self.assertIsNone(result)
110
111		@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
112		def test_get_circuit_empty(self, mock_find_circuits):
113		"""Verify get circuit with empty circuits"""
114		circuit = {
115		"dpid_a": "00:00:00:00:00:00:00:01",
116		"port_a": 1,
117		"vlan_a": 100,
118		"dpid_z": "00:00:00:00:00:00:00:03",
119		"port_z": 2,
120		"vlan_z": 200,
121		}
122
123		tracer = MagicMock()
124		automate = Automate(tracer)
125		automate._circuits = []
126
127		result = automate.get_circuit(circuit)
128
129		mock_find_circuits.assert_called_once()
130		self.assertIsNone(result)
131
132		def test_check_step(self):
133		"""Verify check_step success."""
134		trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
135		circuit_step = {}
136		circuit_step["dpid"] = trace_step["dpid"]
137		circuit_step["in_port"] = trace_step["port"]
138
139		tracer = MagicMock()
140		automate = Automate(tracer)
141		result = automate.check_step(circuit_step, trace_step)
142
143		self.assertTrue(result)
144
145	View Code Duplication	def test_check_step_wront_dpid(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
146		"""Verify if check_step fail with different dpid."""
147		trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
148		circuit_step = {}
149		circuit_step["dpid"] = "00:00:00:00:00:00:00:02"
150		circuit_step["in_port"] = trace_step["port"]
151
152		tracer = MagicMock()
153		automate = Automate(tracer)
154		result = automate.check_step(circuit_step, trace_step)
155
156		self.assertFalse(result)
157
158	View Code Duplication	def test_check_step_wrong_port(self):
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
159		"""Verify if check_step fail with different port."""
160		trace_step = {"dpid": "00:00:00:00:00:00:00:01", "port": 1}
161		circuit_step = {}
162		circuit_step["dpid"] = trace_step["dpid"]
163		circuit_step["in_port"] = 2
164
165		tracer = MagicMock()
166		automate = Automate(tracer)
167		result = automate.check_step(circuit_step, trace_step)
168
169		self.assertFalse(result)
170
171	View Code Duplication	@patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
172		@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
173		def test_check_trace(self, mock_find_circuits, mock_get_circuit):
174		"""Verify _check_trace with trace finding a valid circuit."""
175		circuit_steps = [
176		{
177		"dpid": "00:00:00:00:00:00:00:01",
178		"in_port": 3,
179		},
180		{
181		"dpid": "00:00:00:00:00:00:00:02",
182		"in_port": 3,
183		},
184		]
185		mock_get_circuit.return_value = circuit_steps
186
187		trace = [
188		{
189		"dpid": "00:00:00:00:00:00:00:01",
190		"port": 3,
191		},
192		{
193		"dpid": "00:00:00:00:00:00:00:02",
194		"port": 3,
195		},
196		{
197		"dpid": "00:00:00:00:00:00:00:03",
198		"port": 3,
199		},
200		]
201
202		circuit = {
203		"dpid_a": "00:00:00:00:00:00:00:01",
204		"port_a": 1,
205		"dpid_z": "00:00:00:00:00:00:00:03",
206		"port_z": 2,
207		}
208
209		tracer = MagicMock()
210		automate = Automate(tracer)
211		automate._circuits = []
212		result = automate._check_trace(circuit, trace)
213
214		mock_find_circuits.assert_called_once()
215		mock_get_circuit.assert_called_once()
216		self.assertTrue(result)
217
218		# pylint: disable=duplicate-code
219	View Code Duplication	@patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
220		@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
221		def test_check_trace__short_trace(
222		self, mock_find_circuits, mock_get_circuit
223		):
224		"""Verify _check_trace if lenght of circuit steps is different from
225		lenght of trace steps"""
226		circuit_steps = [
227		{
228		"dpid": "00:00:00:00:00:00:00:01",
229		"in_port": 3,
230		},
231		{
232		"dpid": "00:00:00:00:00:00:00:02",
233		"in_port": 3,
234		},
235		]
236		mock_get_circuit.return_value = circuit_steps
237
238		trace = [
239		{
240		"dpid": "00:00:00:00:00:00:00:01",
241		"port": 3,
242		},
243		{
244		"dpid": "00:00:00:00:00:00:00:03",
245		"port": 3,
246		},
247		]
248
249		circuit = {
250		"dpid_a": "00:00:00:00:00:00:00:01",
251		"port_a": 1,
252		"dpid_z": "00:00:00:00:00:00:00:03",
253		"port_z": 2,
254		}
255
256		tracer = MagicMock()
257		automate = Automate(tracer)
258		automate._circuits = []
259		result = automate._check_trace(circuit, trace)
260
261		mock_find_circuits.assert_called_once()
262		mock_get_circuit.assert_called_once()
263		self.assertFalse(result)
264
265	View Code Duplication	@patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")
		0 ignored issues – show Duplication introduced 2022-07-12 02:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
266		@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
267		def test_check_trace__wrong_steps(
268		self, mock_find_circuits, mock_get_circuit
269		):
270		"""Verify _check_trace with circuit steps different
271		from trace steps"""
272		circuit_steps = [
273		{
274		"dpid": "00:00:00:00:00:00:00:01",
275		"in_port": 3,
276		},
277		{
278		"dpid": "00:00:00:00:00:00:00:02",
279		"in_port": 3,
280		},
281		]
282		mock_get_circuit.return_value = circuit_steps
283
284		trace = [
285		{
286		"dpid": "00:00:00:00:00:00:00:01",
287		"port": 3,
288		},
289		{
290		"dpid": "00:00:00:00:00:00:00:05",
291		"port": 3,
292		},
293		{
294		"dpid": "00:00:00:00:00:00:00:03",
295		"port": 3,
296		},
297		]
298
299		circuit = {
300		"dpid_a": "00:00:00:00:00:00:00:01",
301		"port_a": 1,
302		"dpid_z": "00:00:00:00:00:00:00:03",
303		"port_z": 2,
304		}
305
306		tracer = MagicMock()
307		automate = Automate(tracer)
308		automate._circuits = []
309		result = automate._check_trace(circuit, trace)
310
311		mock_find_circuits.assert_called_once()
312		mock_get_circuit.assert_called_once()
313		self.assertFalse(result)
314
315		@patch("napps.amlight.sdntrace_cp.automate.Automate.get_circuit")
316		@patch("napps.amlight.sdntrace_cp.automate.Automate.find_circuits")
317		def test_check_trace__no_steps(self, mock_find_circuits, mock_get_circuit):
318		"""Verify _check_trace with empty circuit steps."""
319		circuit_steps = []
320		mock_get_circuit.return_value = circuit_steps
321
322		trace = MagicMock()
323
324		circuit = {
325		"dpid_a": "00:00:00:00:00:00:00:01",
326		"port_a": 1,
327		"dpid_z": "00:00:00:00:00:00:00:03",
328		"port_z": 2,
329		}
330
331		tracer = MagicMock()
332		automate = Automate(tracer)
333		automate._circuits = []
334		result = automate._check_trace(circuit, trace)
335
336		mock_find_circuits.assert_called_once()
337		mock_get_circuit.assert_called_once()
338		self.assertFalse(result)
339
340		def test_run_traces__empty(self):
341		"""Test run_traces with empty circuits."""
342		tracer = MagicMock()
343		automate = Automate(tracer)
344		automate._circuits = []
345
346		result = automate.run_traces()
347
348		tracer.tracepath.assert_not_called()
349		self.assertEqual(result, [])
350
351		def test_run_traces(self):
352		"""Test run_traces runnin tracepaths for all circuits."""
353		trace_result = [
354		{
355		"in": {
356		"dpid": "00:00:00:00:00:00:00:01",
357		"port": 1,
358		"time": "2022-06-21 21:32:14.420100",
359		"type": "starting",
360		}
361		},
362		{
363		"in": {
364		"dpid": "00:00:00:00:00:00:00:03",
365		"port": 3,
366		"time": "2022-06-21 21:32:14.420200",
367		"type": "trace",
368		"vlan": 100,
369		},
370		"out": {"port": 2, "vlan": 200},
371		},
372		]
373
374		tracer = MagicMock()
375		tracer.tracepath.return_value = trace_result
376
377		automate = Automate(tracer)
378		automate._circuits = [
379		{
380		"circuit": {
381		"dpid_a": "00:00:00:00:00:00:00:01",
382		"port_a": 1,
383		"dpid_z": "00:00:00:00:00:00:00:03",
384		"port_z": 2,
385		},
386		"entries": {
387		"dpid": "00:00:00:00:00:00:00:01",
388		"in_port": 1,
389		"vlan_vid": [100],
390		},
391		},
392		{
393		"circuit": {
394		"dpid_a": "00:00:00:00:00:00:00:02",
395		"port_a": 1,
396		"dpid_z": "00:00:00:00:00:00:00:04",
397		"port_z": 2,
398		},
399		"entries": {
400		"dpid": "00:00:00:00:00:00:00:02",
401		"in_port": 1,
402		"vlan_vid": [100],
403		},
404		},
405		]
406
407		result = automate.run_traces()
408
409		self.assertEqual(tracer.tracepath.call_count, 2)
410		self.assertEqual(result[0], automate._circuits[0])
411		self.assertEqual(result[1], automate._circuits[1])
412
413		def test_find_circuits__empty(self):
414		"""Test find_circuits without switches."""
415		tracer = MagicMock()
416
417		automate = Automate(tracer)
418		automate.find_circuits()
419
420		self.assertEqual(automate._circuits, [])
421
422		def test_find_circuits(self):
423		"""Test find_circuits successfully finding circuits
424		for all switches."""
425		trace_result = [
426		{
427		"in": {
428		"dpid": "00:00:00:00:00:00:00:01",
429		"port": 1,
430		"time": "2022-06-21 21:32:14.420100",
431		"type": "starting",
432		}
433		},
434		{
435		"in": {
436		"dpid": "00:00:00:00:00:00:00:03",
437		"port": 3,
438		"time": "2022-06-21 21:32:14.420200",
439		"type": "trace",
440		"vlan": 100,
441		},
442		"out": {"port": 2, "vlan": 200},
443		},
444		]
445		tracer = MagicMock()
446		tracer.tracepath.return_value = trace_result
447
448		automate = Automate(tracer)
449		switches = [
450		get_switch_mock("00:00:00:00:00:00:00:01", 0x04),
451		get_switch_mock("00:00:00:00:00:00:00:02", 0x04),
452		get_switch_mock("00:00:00:00:00:00:00:03", 0x04),
453		get_switch_mock("00:00:00:00:00:00:00:04", 0x04),
454		]
455		for switch in switches:
456		flow = MagicMock()
457		in_port = MagicMock()
458		in_port.value = 1
459		flow.match = {"in_port": in_port}
460		action = MagicMock()
461		action.action_type = "output"
462		action.port = 1
463		flow.actions = [action]
464		flows = [flow]
465		switch.generic_flows = flows
466
467		automate._tracer.controller.switches.values.return_value = switches
468
469		automate.find_circuits()
470
471		self.assertIsNotNone(automate._circuits)
472
473		self.assertEqual(len(automate._circuits), 4)
474		for item in automate._circuits:
475		self.assertEqual(
476		item["circuit"][0]["dpid"], trace_result[0]["in"]["dpid"]
477		)
478		self.assertEqual(
479		item["circuit"][0]["in_port"], trace_result[0]["in"]["port"]
480		)
481		self.assertEqual(
482		item["circuit"][1]["dpid"], trace_result[1]["in"]["dpid"]
483		)
484		self.assertEqual(
485		item["circuit"][1]["in_port"], trace_result[1]["in"]["port"]
486		)
487		self.assertEqual(
488		item["circuit"][1]["in_vlan"], trace_result[1]["in"]["vlan"]
489		)
490		self.assertEqual(
491		item["circuit"][1]["out_port"], trace_result[1]["out"]["port"]
492		)
493		self.assertEqual(
494		item["circuit"][1]["out_vlan"], trace_result[1]["out"]["vlan"]
495		)
496
497		self.assertEqual(
498		automate._circuits[0]["entries"]["trace"]["switch"]["dpid"],
499		"00:00:00:00:00:00:00:01",
500		)
501		self.assertEqual(
502		automate._circuits[1]["entries"]["trace"]["switch"]["dpid"],
503		"00:00:00:00:00:00:00:02",
504		)
505		self.assertEqual(
506		automate._circuits[2]["entries"]["trace"]["switch"]["dpid"],
507		"00:00:00:00:00:00:00:03",
508		)
509		self.assertEqual(
510		automate._circuits[3]["entries"]["trace"]["switch"]["dpid"],
511		"00:00:00:00:00:00:00:04",
512		)
513
514		@patch("napps.amlight.sdntrace_cp.automate.requests")
515		# pylint: disable=no-self-use
516		def test_run_important_traces__empty(self, mock_requests):
517		"""Test run_important_traces with empty circuits to run."""
518		tracer = MagicMock()
519
520		automate = Automate(tracer)
521		automate.run_important_traces()
522
523		mock_requests.assert_not_called()
524
525		@patch("napps.amlight.sdntrace_cp.automate.requests")
526		@patch("napps.amlight.sdntrace_cp.automate.settings")
527		def test_run_important_traces(self, mock_settings, mock_requests):
528		"""Test run_important_traces if control plane trace result is
529		different from the data plane trace."""
530		mock_settings.IMPORTANT_CIRCUITS = [
531		{"dpid_a": "00:00:00:00:00:00:00:01", "port_a": 1, "vlan_a": 100}
532		]
533
534		mock_json = MagicMock()
535		mock_json.json.return_value = {
536		"result": [{"type": "starting"}, {"type": "last"}]
537		}
538		mock_requests.get.return_value = mock_json
539
540		def side_effect(event):
541		self.assertTrue(
542		event.content["message"]["source"], "amlight/sdntrace_cp"
543		)
544		self.assertTrue(
545		event.content["message"]["m_body"],
546		"Trace in data plane different from trace in control plane "
547		"for circuit {'dpid_a': '00:00:00:00:00:00:00:01', "
548		"'port_a': 1, 'vlan_a': 100}"
549		)
550
551		tracer = MagicMock()
552		tracer.controller.buffers.app.put.side_effect = side_effect
553
554		automate = Automate(tracer)
555		automate.run_important_traces()
556
557		@patch("napps.amlight.sdntrace_cp.automate.requests")
558		@patch("napps.amlight.sdntrace_cp.automate.settings")
559		@patch("napps.amlight.sdntrace_cp.automate.Automate._check_trace")
560		# pylint: disable=no-self-use
561		def test_run_important_traces__success(
562		self, mock_check_trace, mock_settings, mock_requests
563		):
564		"""Verify run_important_traces if control plane trace result
565		is the same result from the data plane trace."""
566		# Patch to check trace with stored circuit
567		mock_check_trace.return_value = True
568
569		mock_settings.IMPORTANT_CIRCUITS = [
570		{
571		"dpid_a": "00:00:00:00:00:00:00:01",
572		"port_a": 1,
573		"vlan_a": 100,
574		"dpid_z": "00:00:00:00:00:00:00:02",
575		"port_z": 2,
576		"vlan_z": 100,
577		},
578		]
579
580		mock_json = MagicMock()
581		mock_json.json.return_value = {
582		"result": [
583		{
584		"dpid": "00:00:00:00:00:00:00:01",
585		"port": 1,
586		"type": "starting",
587		},
588		{"dpid": "00:00:00:00:00:00:00:03", "port": 1, "type": "last"},
589		]
590		}
591		mock_requests.get.return_value = mock_json
592
593		tracer = MagicMock()
594		tracer.controller.buffers.app.put.side_effect = None
595
596		automate = Automate(tracer)
597		automate.run_important_traces()
598
599		# Check if important trace dont trigger the event
600		# It means that the CP trace is the same to the DP trace
601		tracer.controller.buffers.app.put.assert_not_called()
602

kytos-ng / sdntrace_cp

Pull Request — master (#30)

TestAutomate.test_run_traces() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like