build.tests.unit.test_automate.TestAutomate.test_schedule_traces() - Code Metrics - Inspection of "Replacing scheduler napp with scheduler library" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#36)

by Vinicius

created 2022-09-13 13:15 UTC

TestAutomate.test_schedule_traces() A

↳ Parent: build.tests.unit.test_automate

Complexity

Conditions

Size

Total Lines	13
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	2
dl	0
loc	13
rs	9.8
c	0
b	0
f	0

"""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 class 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),
        ]
        switches_dict = {}
        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
            switches_dict[switch.id] = switch

        automate._tracer.controller.switches = switches_dict

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

    def test_schedule_id(self):
        """Test schedule_id with proper id type"""
        tracer = MagicMock()
        automate = Automate(tracer)
        try:
            automate.schedule_id('mock_id')
            self.assertEqual(len(automate.ids), 1)
        except AttributeError:
            self.fail('schedule_id() raised an error')

    def test_schedule_id_fail(self):
        """Test schedule_id with non-string id"""
        tracer = MagicMock()
        automate = Automate(tracer)
        with self.assertRaises(AttributeError):
            automate.schedule_id(1)

    @patch("napps.amlight.sdntrace_cp.automate.settings")
    def test_schedule_traces(self, mock_settings):
        """Test schedule_traces with the arguments from settings"""
        mock_settings.TRIGGER_SCHEDULE_TRACES = True
        mock_settings.SCHEDULE_ARGS = {'seconds': 120}
        mock_settings.SCHEDULE_TRIGGER = 'interval'
        tracer = MagicMock()
        automate = Automate(tracer)
        try:
            job = automate.schedule_traces(mock_settings)
        except AttributeError:
            self.fail("automate.schedule_traces() raised an error")
        self.assertIsNotNone(job)

    @patch("napps.amlight.sdntrace_cp.automate.settings")
    def test_schedule_traces_fail(self, mock_settings):
        """Test schedule_traces with wrong arguments from settings"""
        mock_settings.TRIGGER_SCHEDULE_TRACES = True
        mock_settings.SCHEDULE_ARGS = 120
        mock_settings.SCHEDULE_TRIGGER = {'interval'}
        tracer = MagicMock()
        automate = Automate(tracer)
        with self.assertRaises(AttributeError):
            automate.schedule_traces(mock_settings)

    @patch("napps.amlight.sdntrace_cp.automate.settings")
    def test_schedule_important_traces(self, mock_settings):
        """
        Test schedule_important_traces with the arguments from settings
        """
        mock_settings.TRIGGER_IMPORTANT_CIRCUITS = True
        mock_settings.IMPORTANT_CIRCUITS_ARGS = {'seconds': 20}
        mock_settings.IMPORTANT_CIRCUITS_TRIGGER = 'interval'
        tracer = MagicMock()
        automate = Automate(tracer)
        try:
            job = automate.schedule_important_traces(mock_settings)
        except AttributeError:
            self.fail("automate.schedule_important_traces() raised an error")
        self.assertIsNotNone(job)

    @patch("napps.amlight.sdntrace_cp.automate.settings")
    def test_schedule_important_traces_fail(self, mock_settings):
        """
        Test schedule_important_traces with wrong arguments from settings
        """
        mock_settings.TRIGGER_IMPORTANT_CIRCUITS = True
        mock_settings.IMPORTANT_CIRCUITS_ARGS = 20
        mock_settings.IMPORTANT_CIRCUITS_TRIGGER = {'interval'}
        tracer = MagicMock()
        automate = Automate(tracer)
        with self.assertRaises(AttributeError):
            automate.schedule_important_traces(mock_settings)

    def test_unschedule_ids(self):
        """Test unschedule_ids with existent ids"""
        tracer = MagicMock()
        automate = Automate(tracer)
        automate.scheduler = MagicMock()
        automate.schedule_id('mock_id')
        automate.schedule_id('id_mocked')
        self.assertEqual(len(automate.ids), 2)
        id_ = {'mock_id'}
        automate.unschedule_ids(id_set=id_)
        self.assertEqual(len(automate.ids), 1)
        automate.unschedule_ids()
        self.assertEqual(len(automate.ids), 0)


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 class 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		switches_dict = {}
456		for switch in switches:
457		flow = MagicMock()
458		in_port = MagicMock()
459		in_port.value = 1
460		flow.match = {"in_port": in_port}
461		action = MagicMock()
462		action.action_type = "output"
463		action.port = 1
464		flow.actions = [action]
465		flows = [flow]
466		switch.generic_flows = flows
467		switches_dict[switch.id] = switch
468
469		automate._tracer.controller.switches = switches_dict
470
471		automate.find_circuits()
472
473		self.assertIsNotNone(automate._circuits)
474
475		self.assertEqual(len(automate._circuits), 4)
476		for item in automate._circuits:
477		self.assertEqual(
478		item["circuit"][0]["dpid"], trace_result[0]["in"]["dpid"]
479		)
480		self.assertEqual(
481		item["circuit"][0]["in_port"], trace_result[0]["in"]["port"]
482		)
483		self.assertEqual(
484		item["circuit"][1]["dpid"], trace_result[1]["in"]["dpid"]
485		)
486		self.assertEqual(
487		item["circuit"][1]["in_port"], trace_result[1]["in"]["port"]
488		)
489		self.assertEqual(
490		item["circuit"][1]["in_vlan"], trace_result[1]["in"]["vlan"]
491		)
492		self.assertEqual(
493		item["circuit"][1]["out_port"], trace_result[1]["out"]["port"]
494		)
495		self.assertEqual(
496		item["circuit"][1]["out_vlan"], trace_result[1]["out"]["vlan"]
497		)
498
499		self.assertEqual(
500		automate._circuits[0]["entries"]["trace"]["switch"]["dpid"],
501		"00:00:00:00:00:00:00:01",
502		)
503		self.assertEqual(
504		automate._circuits[1]["entries"]["trace"]["switch"]["dpid"],
505		"00:00:00:00:00:00:00:02",
506		)
507		self.assertEqual(
508		automate._circuits[2]["entries"]["trace"]["switch"]["dpid"],
509		"00:00:00:00:00:00:00:03",
510		)
511		self.assertEqual(
512		automate._circuits[3]["entries"]["trace"]["switch"]["dpid"],
513		"00:00:00:00:00:00:00:04",
514		)
515
516		@patch("napps.amlight.sdntrace_cp.automate.requests")
517		# pylint: disable=no-self-use
518		def test_run_important_traces__empty(self, mock_requests):
519		"""Test run_important_traces with empty circuits to run."""
520		tracer = MagicMock()
521
522		automate = Automate(tracer)
523		automate.run_important_traces()
524
525		mock_requests.assert_not_called()
526
527		@patch("napps.amlight.sdntrace_cp.automate.requests")
528		@patch("napps.amlight.sdntrace_cp.automate.settings")
529		def test_run_important_traces(self, mock_settings, mock_requests):
530		"""Test run_important_traces if control plane trace result is
531		different from the data plane trace."""
532		mock_settings.IMPORTANT_CIRCUITS = [
533		{"dpid_a": "00:00:00:00:00:00:00:01", "port_a": 1, "vlan_a": 100}
534		]
535
536		mock_json = MagicMock()
537		mock_json.json.return_value = {
538		"result": [{"type": "starting"}, {"type": "last"}]
539		}
540		mock_requests.get.return_value = mock_json
541
542		def side_effect(event):
543		self.assertTrue(
544		event.content["message"]["source"], "amlight/sdntrace_cp"
545		)
546		self.assertTrue(
547		event.content["message"]["m_body"],
548		"Trace in data plane different from trace in control plane "
549		"for circuit {'dpid_a': '00:00:00:00:00:00:00:01', "
550		"'port_a': 1, 'vlan_a': 100}"
551		)
552
553		tracer = MagicMock()
554		tracer.controller.buffers.app.put.side_effect = side_effect
555
556		automate = Automate(tracer)
557		automate.run_important_traces()
558
559		@patch("napps.amlight.sdntrace_cp.automate.requests")
560		@patch("napps.amlight.sdntrace_cp.automate.settings")
561		@patch("napps.amlight.sdntrace_cp.automate.Automate._check_trace")
562		# pylint: disable=no-self-use
563		def test_run_important_traces__success(
564		self, mock_check_trace, mock_settings, mock_requests
565		):
566		"""Verify run_important_traces if control plane trace result
567		is the same result from the data plane trace."""
568		# Patch to check trace with stored circuit
569		mock_check_trace.return_value = True
570
571		mock_settings.IMPORTANT_CIRCUITS = [
572		{
573		"dpid_a": "00:00:00:00:00:00:00:01",
574		"port_a": 1,
575		"vlan_a": 100,
576		"dpid_z": "00:00:00:00:00:00:00:02",
577		"port_z": 2,
578		"vlan_z": 100,
579		},
580		]
581
582		mock_json = MagicMock()
583		mock_json.json.return_value = {
584		"result": [
585		{
586		"dpid": "00:00:00:00:00:00:00:01",
587		"port": 1,
588		"type": "starting",
589		},
590		{"dpid": "00:00:00:00:00:00:00:03", "port": 1, "type": "last"},
591		]
592		}
593		mock_requests.get.return_value = mock_json
594
595		tracer = MagicMock()
596		tracer.controller.buffers.app.put.side_effect = None
597
598		automate = Automate(tracer)
599		automate.run_important_traces()
600
601		# Check if important trace dont trigger the event
602		# It means that the CP trace is the same to the DP trace
603		tracer.controller.buffers.app.put.assert_not_called()
604
605		def test_schedule_id(self):
606		"""Test schedule_id with proper id type"""
607		tracer = MagicMock()
608		automate = Automate(tracer)
609		try:
610		automate.schedule_id('mock_id')
611		self.assertEqual(len(automate.ids), 1)
612		except AttributeError:
613		self.fail('schedule_id() raised an error')
614
615		def test_schedule_id_fail(self):
616		"""Test schedule_id with non-string id"""
617		tracer = MagicMock()
618		automate = Automate(tracer)
619		with self.assertRaises(AttributeError):
620		automate.schedule_id(1)
621
622		@patch("napps.amlight.sdntrace_cp.automate.settings")
623		def test_schedule_traces(self, mock_settings):
624		"""Test schedule_traces with the arguments from settings"""
625		mock_settings.TRIGGER_SCHEDULE_TRACES = True
626		mock_settings.SCHEDULE_ARGS = {'seconds': 120}
627		mock_settings.SCHEDULE_TRIGGER = 'interval'
628		tracer = MagicMock()
629		automate = Automate(tracer)
630		try:
631		job = automate.schedule_traces(mock_settings)
632		except AttributeError:
633		self.fail("automate.schedule_traces() raised an error")
634		self.assertIsNotNone(job)
635
636		@patch("napps.amlight.sdntrace_cp.automate.settings")
637		def test_schedule_traces_fail(self, mock_settings):
638		"""Test schedule_traces with wrong arguments from settings"""
639		mock_settings.TRIGGER_SCHEDULE_TRACES = True
640		mock_settings.SCHEDULE_ARGS = 120
641		mock_settings.SCHEDULE_TRIGGER = {'interval'}
642		tracer = MagicMock()
643		automate = Automate(tracer)
644		with self.assertRaises(AttributeError):
645		automate.schedule_traces(mock_settings)
646
647		@patch("napps.amlight.sdntrace_cp.automate.settings")
648		def test_schedule_important_traces(self, mock_settings):
649		"""
650		Test schedule_important_traces with the arguments from settings
651		"""
652		mock_settings.TRIGGER_IMPORTANT_CIRCUITS = True
653		mock_settings.IMPORTANT_CIRCUITS_ARGS = {'seconds': 20}
654		mock_settings.IMPORTANT_CIRCUITS_TRIGGER = 'interval'
655		tracer = MagicMock()
656		automate = Automate(tracer)
657		try:
658		job = automate.schedule_important_traces(mock_settings)
659		except AttributeError:
660		self.fail("automate.schedule_important_traces() raised an error")
661		self.assertIsNotNone(job)
662
663		@patch("napps.amlight.sdntrace_cp.automate.settings")
664		def test_schedule_important_traces_fail(self, mock_settings):
665		"""
666		Test schedule_important_traces with wrong arguments from settings
667		"""
668		mock_settings.TRIGGER_IMPORTANT_CIRCUITS = True
669		mock_settings.IMPORTANT_CIRCUITS_ARGS = 20
670		mock_settings.IMPORTANT_CIRCUITS_TRIGGER = {'interval'}
671		tracer = MagicMock()
672		automate = Automate(tracer)
673		with self.assertRaises(AttributeError):
674		automate.schedule_important_traces(mock_settings)
675
676		def test_unschedule_ids(self):
677		"""Test unschedule_ids with existent ids"""
678		tracer = MagicMock()
679		automate = Automate(tracer)
680		automate.scheduler = MagicMock()
681		automate.schedule_id('mock_id')
682		automate.schedule_id('id_mocked')
683		self.assertEqual(len(automate.ids), 2)
684		id_ = {'mock_id'}
685		automate.unschedule_ids(id_set=id_)
686		self.assertEqual(len(automate.ids), 1)
687		automate.unschedule_ids()
688		self.assertEqual(len(automate.ids), 0)
689

kytos-ng / sdntrace_cp

Pull Request — master (#36)

TestAutomate.test_schedule_traces() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like