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)

unknown

created 2022-09-12 18:01 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.core import log
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(MagicMock, 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(MagicMock, 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(MagicMock, 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(MagicMock, mock_settings)

    def test_unschedule_id(self):
        """Test unschedule_id with an existent id"""
        tracer = MagicMock()
        automate = Automate(tracer)
        id_ = 'mock_id'
        automate.schedule_id(id_)
        self.assertEqual(len(automate.ids), 1)
        automate.unschedule_id({id_})
        self.assertEqual(len(automate.ids), 0)

    def test_unschedule_id_fail(self):
        """Test unschedule_id with a non-existent id"""
        tracer = MagicMock()
        automate = Automate(tracer)
        automate.unschedule_id({'mock_id'})
        self.assertLogs(log, level='warning')


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

kytos-ng / sdntrace_cp

Pull Request — master (#36)

TestAutomate.test_schedule_traces() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like