build.tests.unit.test_main.TestMain.test_event_add_flow() - Code Metrics - Inspection of "Add support for installing flows on switches throu..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#129)

by Carlos

created 2021-05-18 14:13 UTC

TestMain.test_event_add_flow() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	2
dl	0
loc	11
rs	9.9
c	0
b	0
f	0

"""Test Main methods."""
from unittest import TestCase
from unittest.mock import MagicMock, patch

from kytos.lib.helpers import (get_connection_mock, get_controller_mock,
                               get_kytos_event_mock, get_switch_mock,
                               get_test_client)


# pylint: disable=protected-access, too-many-public-methods
class TestMain(TestCase):
    """Tests for the Main class."""

    API_URL = 'http://localhost:8181/api/kytos/flow_manager'

    def setUp(self):
        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        # pylint: disable=import-outside-toplevel
        from napps.kytos.flow_manager.main import Main

        self.addCleanup(patch.stopall)

        controller = get_controller_mock()
        self.switch_01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        self.switch_01.is_enabled.return_value = True
        self.switch_01.flows = []

        self.switch_02 = get_switch_mock("00:00:00:00:00:00:00:02", 0x04)
        self.switch_02.is_enabled.return_value = False
        self.switch_02.flows = []

        controller.switches = {"00:00:00:00:00:00:00:01": self.switch_01,
                               "00:00:00:00:00:00:00:02": self.switch_02}

        self.napp = Main(controller)

    def test_rest_list_without_dpid(self):
        """Test list rest method withoud dpid."""
        flow_dict = {
            "priority": 13,
            "cookie": 84114964,
            "command": "add",
            "match": {"dl_dst": "00:15:af:d5:38:98"},
        }
        flow_dict_2 = {
            "priority": 18,
            "cookie": 84114964,
            "command": "add",
            "match": {"dl_dst": "00:15:af:d5:38:98"},
        }
        flow_1 = MagicMock()
        flow_1.as_dict.return_value = flow_dict
        flow_2 = MagicMock()
        flow_2.as_dict.return_value = flow_dict_2
        self.switch_01.flows.append(flow_1)
        self.switch_02.flows.append(flow_2)

        api = get_test_client(self.napp.controller, self.napp)
        url = f'{self.API_URL}/v2/flows'

        response = api.get(url)
        expected = {
            '00:00:00:00:00:00:00:01': {'flows': [flow_dict]},
            '00:00:00:00:00:00:00:02': {'flows': [flow_dict_2]},
        }
        self.assertEqual(response.json, expected)
        self.assertEqual(response.status_code, 200)

    def test_rest_list_with_dpid(self):
        """Test list rest method with dpid."""
        flow_dict = {
            "priority": 13,
            "cookie": 84114964,
            "command": "add",
            "match": {"dl_dst": "00:15:af:d5:38:98"},
        }
        flow_1 = MagicMock()
        flow_1.as_dict.return_value = flow_dict
        self.switch_01.flows.append(flow_1)

        api = get_test_client(self.napp.controller, self.napp)
        url = f'{self.API_URL}/v2/flows/00:00:00:00:00:00:00:01'

        response = api.get(url)
        expected = {'00:00:00:00:00:00:00:01': {'flows': [flow_dict]}}

        self.assertEqual(response.json, expected)
        self.assertEqual(response.status_code, 200)

    @patch('napps.kytos.flow_manager.main.Main._install_flows')
    def test_rest_add_and_delete_without_dpid(self, mock_install_flows):
        """Test add and delete rest method without dpid."""
        api = get_test_client(self.napp.controller, self.napp)

        for method in ['flows', 'delete']:
            url = f'{self.API_URL}/v2/{method}'

            response_1 = api.post(url, json={'data': '123'})
            response_2 = api.post(url)

            self.assertEqual(response_1.status_code, 200)
            self.assertEqual(response_2.status_code, 404)

        self.assertEqual(mock_install_flows.call_count, 2)

    @patch('napps.kytos.flow_manager.main.Main._install_flows')
    def test_rest_add_and_delete_with_dpid(self, mock_install_flows):
        """Test add and delete rest method with dpid."""
        api = get_test_client(self.napp.controller, self.napp)

        for method in ['flows', 'delete']:
            url_1 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:01'
            url_2 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:02'
            url_3 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:03'

            response_1 = api.post(url_1)
            response_2 = api.post(url_1, json={'data': '123'})
            response_3 = api.post(url_2, json={'data': '123'})
            response_4 = api.post(url_3, json={'data': '123'})

            self.assertEqual(response_1.status_code, 404)
            self.assertEqual(response_2.status_code, 200)
            if method == 'flows':
                self.assertEqual(response_3.status_code, 404)
            else:
                self.assertEqual(response_3.status_code, 200)
            self.assertEqual(response_4.status_code, 404)

        self.assertEqual(mock_install_flows.call_count, 3)

    def test_get_all_switches_enabled(self):
        """Test _get_all_switches_enabled method."""
        switches = self.napp._get_all_switches_enabled()

        self.assertEqual(switches, [self.switch_01])

    @patch('napps.kytos.flow_manager.main.Main._store_changed_flows')

    @patch('napps.kytos.flow_manager.main.Main._send_napp_event')
    @patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
    @patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    def test_install_flows(self, *args):
        """Test _install_flows method."""
        (mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
         mock_send_napp_event, _) = args
        serializer = MagicMock()
        flow = MagicMock()
        flow_mod = MagicMock()

        flow.as_of_add_flow_mod.return_value = flow_mod
        serializer.from_dict.return_value = flow
        mock_flow_factory.return_value = serializer

        flows_dict = {'flows': [MagicMock()]}
        switches = [self.switch_01]
        self.napp._install_flows('add', flows_dict, switches)

        mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
        mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
                                                  flow, 'add')
        mock_send_napp_event.assert_called_with(self.switch_01, flow, 'add')

    @patch('napps.kytos.flow_manager.main.Main._store_changed_flows')

    @patch('napps.kytos.flow_manager.main.Main._send_napp_event')
    @patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
    @patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    def test_install_flows_with_delete_strict(self, *args):
        """Test _install_flows method with strict delete command."""
        (mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
         mock_send_napp_event, _) = args
        serializer = MagicMock()
        flow = MagicMock()
        flow_mod = MagicMock()

        flow.as_of_strict_delete_flow_mod.return_value = flow_mod
        serializer.from_dict.return_value = flow
        mock_flow_factory.return_value = serializer

        flows_dict = {'flows': [MagicMock()]}
        switches = [self.switch_01]
        self.napp._install_flows('delete_strict', flows_dict, switches)

        mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
        mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
                                                  flow, 'delete_strict')
        mock_send_napp_event.assert_called_with(self.switch_01, flow,
                                                'delete_strict')

    @patch('napps.kytos.flow_manager.main.Main._install_flows')
    def test_event_add_flow(self, mock_install_flows):
        """Test event_add_flow method."""
        dpid = "00:00:00:00:00:00:00:01"
        mock_flow_dict = MagicMock()
        event = get_kytos_event_mock(name='kytos.flow_manager.flows.install',
                                     content={'dpid': dpid,
                                              'command': 'add',
                                              'flow_dict': mock_flow_dict})
        self.napp.event_add_flow(event)
        mock_install_flows.assert_called()

    def test_add_flow_mod_sent(self):
        """Test _add_flow_mod_sent method."""
        xid = 0
        flow = MagicMock()

        self.napp._add_flow_mod_sent(xid, flow, 'add')

        self.assertEqual(self.napp._flow_mods_sent[xid], (flow, 'add'))

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_send_flow_mod(self, mock_buffers_put):
        """Test _send_flow_mod method."""
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        flow_mod = MagicMock()

        self.napp._send_flow_mod(switch, flow_mod)

        mock_buffers_put.assert_called()

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_send_napp_event(self, mock_buffers_put):
        """Test _send_napp_event method."""
        switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        flow = MagicMock()

        for command in ['add', 'delete', 'delete_strict', 'error']:
            self.napp._send_napp_event(switch, flow, command)

        self.assertEqual(mock_buffers_put.call_count, 4)

    @patch('napps.kytos.flow_manager.main.Main._send_napp_event')
    def test_handle_errors(self, mock_send_napp_event):
        """Test handle_errors method."""
        flow = MagicMock()
        self.napp._flow_mods_sent[0] = (flow, 'add')

        switch = get_switch_mock("00:00:00:00:00:00:00:01")
        switch.connection = get_connection_mock(
            0x04, get_switch_mock("00:00:00:00:00:00:00:02"))

        protocol = MagicMock()
        protocol.unpack.return_value = 'error_packet'

        switch.connection.protocol = protocol

        message = MagicMock()
        message.header.xid.value = 0
        message.error_type = 2
        message.code = 5
        event = get_kytos_event_mock(name='.*.of_core.*.ofpt_error',
                                     content={'message': message,
                                              'source': switch.connection})
        self.napp.handle_errors(event)

        mock_send_napp_event.assert_called_with(flow.switch, flow, 'error',
                                                error_command='add',
                                                error_code=5, error_type=2)

    @patch("napps.kytos.flow_manager.main.StoreHouse.get_data")
    def test_load_flows(self, mock_storehouse):
        """Test load flows."""
        self.napp._load_flows()
        mock_storehouse.assert_called()

    @patch("napps.kytos.flow_manager.main.Main._install_flows")
    def test_resend_stored_flows(self, mock_install_flows):
        """Test resend stored flows."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        mock_event = MagicMock()
        flow = {"command": "add", "flow": MagicMock()}

        flows = {"flow_list": [flow]}
        mock_event.content = {"switch": switch}
        self.napp.controller.switches = {dpid: switch}
        self.napp.stored_flows = {dpid: flows}
        self.napp.resend_stored_flows(mock_event)
        mock_install_flows.assert_called()

    @patch("napps.kytos.of_core.flow.FlowFactory.get_class")
    @patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
    def test_store_changed_flows(self, mock_save_flow, _):
        """Test store changed flows."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        flow = {
            "priority": 17,
            "cookie": 84114964,
            "command": "add",
            "match": {"dl_dst": "00:15:af:d5:38:98"},
        }
        match_fields = {
            "priority": 17,
            "cookie": 84114964,
            "command": "add",
            "dl_dst": "00:15:af:d5:38:98",
        }
        flows = {"flow": flow}

        command = "add"
        flow_list = {
            "flow_list": [
                {"match_fields": match_fields, "command": "delete",
                 "flow": flow}
            ]
        }
        self.napp.stored_flows = {dpid: flow_list}
        self.napp._store_changed_flows(command, flows, switch)
        mock_save_flow.assert_called()

        self.napp.stored_flows = {}
        self.napp._store_changed_flows(command, flows, switch)
        mock_save_flow.assert_called()

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    def test_check_switch_consistency_add(self, *args):
        """Test check_switch_consistency method.

        This test checks the case when a flow is missing in switch and have the
        ADD command.
        """
        (mock_flow_factory, mock_install_flows) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.flows = []

        flow_1 = MagicMock()
        flow_1.as_dict.return_value = {'flow_1': 'data'}

        flow_list = [{"command": "add",
                      "flow": {'flow_1': 'data'}
                      }]
        serializer = MagicMock()

        mock_flow_factory.return_value = serializer
        self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
        self.napp.check_switch_consistency(switch)
        mock_install_flows.assert_called()

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    def test_check_switch_consistency_delete(self, *args):
        """Test check_switch_consistency method.

        This test checks the case when a flow is missing in switch and have the
        DELETE command.
        """
        (mock_flow_factory, mock_install_flows) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)

        flow_1 = MagicMock()
        flow_1.as_dict.return_value = {'flow_1': 'data'}

        flow_list = [{"command": "delete",
                      "flow": {'flow_1': 'data'}
                      }]
        serializer = MagicMock()
        serializer.from_dict.return_value = flow_1

        switch.flows = [flow_1]

        mock_flow_factory.return_value = serializer
        self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
        self.napp.check_switch_consistency(switch)
        mock_install_flows.assert_called()

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    def test_check_storehouse_consistency(self, *args):
        """Test check_storehouse_consistency method.

        This test checks the case when a flow is missing in storehouse.
        """
        (mock_flow_factory, mock_install_flows) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)

        flow_1 = MagicMock()
        flow_1.as_dict.return_value = {'flow_1': 'data'}

        switch.flows = [flow_1]

        flow_list = [{"command": "add",
                      "flow": {'flow_2': 'data'}
                      }]
        serializer = MagicMock()

        mock_flow_factory.return_value = serializer
        self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
        self.napp.check_storehouse_consistency(switch)
        mock_install_flows.assert_called()

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    @patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
    def test_no_strict_delete(self, *args):
        """Test the non-strict matching method.

        Test non-strict matching to delete a Flow using a cookie.
        """
        (mock_save_flow, _, _) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        stored_flow = {
            "command": "add",
            "flow": {
                "actions": [{"action_type": "set_vlan", "vlan_id": 300}],
                "cookie": 6191162389751548793,
                "match": {"dl_vlan": 300, "in_port": 1},
            },
        }
        stored_flow2 = {
            "command": "add",
            "flow": {
                "actions": [],
                "cookie": 4961162389751548787,
                "match": {"in_port": 2},
            },
        }
        flow_to_install = {
            "cookie": 6191162389751548793,
            "cookie_mask": 18446744073709551615,
        }
        flow_list = {"flow_list": [stored_flow, stored_flow2]}
        command = "delete"
        self.napp.stored_flows = {dpid: flow_list}

        self.napp._store_changed_flows(command, flow_to_install, switch)
        mock_save_flow.assert_called()
        self.assertEqual(len(self.napp.stored_flows), 1)

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    @patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
    def test_no_strict_delete_with_ipv4(self, *args):
        """Test the non-strict matching method.

        Test non-strict matching to delete a Flow using IPv4.
        """
        (mock_save_flow, _, _) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        stored_flow = {
            "command": "add",
            "flow": {
                "priority": 10,
                "cookie": 84114904,
                "match": {
                    "ipv4_src": "192.168.1.120",
                    "ipv4_dst": "192.168.0.2",
                },
                "actions": [],
            },
        }
        stored_flow2 = {
            "command": "add",
            "flow": {
                "actions": [],
                "cookie": 4961162389751548787,
                "match": {"in_port": 2},
            },
        }
        flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
        flow_list = {"flow_list": [stored_flow, stored_flow2]}
        command = "delete"
        self.napp.stored_flows = {dpid: flow_list}

        self.napp._store_changed_flows(command, flow_to_install, switch)
        mock_save_flow.assert_called()
        self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 2)

    @patch('napps.kytos.flow_manager.main.Main._install_flows')

    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    @patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
    def test_no_strict_delete_with_ipv4_fail(self, *args):
        """Test the non-strict matching method.

        Test non-strict Fail case matching to delete a Flow using IPv4.
        """
        (mock_save_flow, _, _) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        stored_flow = {
            "command": "add",
            "flow": {
                "priority": 10,
                "cookie": 84114904,
                "match": {
                    "ipv4_src": "192.168.2.1",
                    "ipv4_dst": "192.168.0.2",
                },
                "actions": [],
            },
        }
        stored_flow2 = {
            "command": "add",
            "flow": {
                "actions": [],
                "cookie": 4961162389751548787,
                "match": {"in_port": 2},
            },
        }
        flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
        flow_list = {"flow_list": [stored_flow, stored_flow2]}
        command = "delete"
        self.napp.stored_flows = {dpid: flow_list}

        self.napp._store_changed_flows(command, flow_to_install, switch)
        mock_save_flow.assert_called()
        self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 3)

    @patch('napps.kytos.flow_manager.main.Main._install_flows')
    @patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
    @patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
    def test_no_strict_delete_of10(self, *args):
        """Test the non-strict matching method.

        Test non-strict matching to delete a Flow using OF10.
        """
        (mock_save_flow, _, _) = args
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x01)
        switch.id = dpid
        stored_flow = {
            "command": "add",
            "flow": {
                "actions": [{"max_len": 65535, "port": 6}],
                "cookie": 4961162389751548787,
                "match": {
                    "in_port": 80,
                    "dl_src": "00:00:00:00:00:00",
                    "dl_dst": "f2:0b:a4:7d:f8:ea",
                    "dl_vlan": 0,
                    "dl_vlan_pcp": 0,
                    "dl_type": 0,
                    "nw_tos": 0,
                    "nw_proto": 0,
                    "nw_src": "192.168.0.1",
                    "nw_dst": "0.0.0.0",
                    "tp_src": 0,
                    "tp_dst": 0,
                },
                "out_port": 65532,
                "priority": 123,
            },
        }
        stored_flow2 = {
            "command": "add",
            "flow": {
                "actions": [],
                "cookie": 4961162389751654,
                "match": {
                    "in_port": 2,
                    "dl_src": "00:00:00:00:00:00",
                    "dl_dst": "f2:0b:a4:7d:f8:ea",
                    "dl_vlan": 0,
                    "dl_vlan_pcp": 0,
                    "dl_type": 0,
                    "nw_tos": 0,
                    "nw_proto": 0,
                    "nw_src": "192.168.0.1",
                    "nw_dst": "0.0.0.0",
                    "tp_src": 0,
                    "tp_dst": 0,
                },
                "out_port": 655,
                "priority": 1,
            },
        }
        flow_to_install = {"match": {"in_port": 80, "wildcards": 4194303}}
        flow_list = {"flow_list": [stored_flow, stored_flow2]}
        command = "delete"
        self.napp.stored_flows = {dpid: flow_list}

        self.napp._store_changed_flows(command, flow_to_install, switch)
        mock_save_flow.assert_called()
        self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 1)


1		"""Test Main methods."""
2		from unittest import TestCase
3		from unittest.mock import MagicMock, patch
4
5		from kytos.lib.helpers import (get_connection_mock, get_controller_mock,
6		get_kytos_event_mock, get_switch_mock,
7		get_test_client)
8
9
10		# pylint: disable=protected-access, too-many-public-methods
11		class TestMain(TestCase):
12		"""Tests for the Main class."""
13
14		API_URL = 'http://localhost:8181/api/kytos/flow_manager'
15
16		def setUp(self):
17		patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
18		# pylint: disable=import-outside-toplevel
19		from napps.kytos.flow_manager.main import Main
20
21		self.addCleanup(patch.stopall)
22
23		controller = get_controller_mock()
24		self.switch_01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
25		self.switch_01.is_enabled.return_value = True
26		self.switch_01.flows = []
27
28		self.switch_02 = get_switch_mock("00:00:00:00:00:00:00:02", 0x04)
29		self.switch_02.is_enabled.return_value = False
30		self.switch_02.flows = []
31
32		controller.switches = {"00:00:00:00:00:00:00:01": self.switch_01,
33		"00:00:00:00:00:00:00:02": self.switch_02}
34
35		self.napp = Main(controller)
36
37		def test_rest_list_without_dpid(self):
38		"""Test list rest method withoud dpid."""
39		flow_dict = {
40		"priority": 13,
41		"cookie": 84114964,
42		"command": "add",
43		"match": {"dl_dst": "00:15:af:d5:38:98"},
44		}
45		flow_dict_2 = {
46		"priority": 18,
47		"cookie": 84114964,
48		"command": "add",
49		"match": {"dl_dst": "00:15:af:d5:38:98"},
50		}
51		flow_1 = MagicMock()
52		flow_1.as_dict.return_value = flow_dict
53		flow_2 = MagicMock()
54		flow_2.as_dict.return_value = flow_dict_2
55		self.switch_01.flows.append(flow_1)
56		self.switch_02.flows.append(flow_2)
57
58		api = get_test_client(self.napp.controller, self.napp)
59		url = f'{self.API_URL}/v2/flows'
60
61		response = api.get(url)
62		expected = {
63		'00:00:00:00:00:00:00:01': {'flows': [flow_dict]},
64		'00:00:00:00:00:00:00:02': {'flows': [flow_dict_2]},
65		}
66		self.assertEqual(response.json, expected)
67		self.assertEqual(response.status_code, 200)
68
69		def test_rest_list_with_dpid(self):
70		"""Test list rest method with dpid."""
71		flow_dict = {
72		"priority": 13,
73		"cookie": 84114964,
74		"command": "add",
75		"match": {"dl_dst": "00:15:af:d5:38:98"},
76		}
77		flow_1 = MagicMock()
78		flow_1.as_dict.return_value = flow_dict
79		self.switch_01.flows.append(flow_1)
80
81		api = get_test_client(self.napp.controller, self.napp)
82		url = f'{self.API_URL}/v2/flows/00:00:00:00:00:00:00:01'
83
84		response = api.get(url)
85		expected = {'00:00:00:00:00:00:00:01': {'flows': [flow_dict]}}
86
87		self.assertEqual(response.json, expected)
88		self.assertEqual(response.status_code, 200)
89
90		@patch('napps.kytos.flow_manager.main.Main._install_flows')
91		def test_rest_add_and_delete_without_dpid(self, mock_install_flows):
92		"""Test add and delete rest method without dpid."""
93		api = get_test_client(self.napp.controller, self.napp)
94
95		for method in ['flows', 'delete']:
96		url = f'{self.API_URL}/v2/{method}'
97
98		response_1 = api.post(url, json={'data': '123'})
99		response_2 = api.post(url)
100
101		self.assertEqual(response_1.status_code, 200)
102		self.assertEqual(response_2.status_code, 404)
103
104		self.assertEqual(mock_install_flows.call_count, 2)
105
106		@patch('napps.kytos.flow_manager.main.Main._install_flows')
107		def test_rest_add_and_delete_with_dpid(self, mock_install_flows):
108		"""Test add and delete rest method with dpid."""
109		api = get_test_client(self.napp.controller, self.napp)
110
111		for method in ['flows', 'delete']:
112		url_1 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:01'
113		url_2 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:02'
114		url_3 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:03'
115
116		response_1 = api.post(url_1)
117		response_2 = api.post(url_1, json={'data': '123'})
118		response_3 = api.post(url_2, json={'data': '123'})
119		response_4 = api.post(url_3, json={'data': '123'})
120
121		self.assertEqual(response_1.status_code, 404)
122		self.assertEqual(response_2.status_code, 200)
123		if method == 'flows':
124		self.assertEqual(response_3.status_code, 404)
125		else:
126		self.assertEqual(response_3.status_code, 200)
127		self.assertEqual(response_4.status_code, 404)
128
129		self.assertEqual(mock_install_flows.call_count, 3)
130
131		def test_get_all_switches_enabled(self):
132		"""Test _get_all_switches_enabled method."""
133		switches = self.napp._get_all_switches_enabled()
134
135		self.assertEqual(switches, [self.switch_01])
136
137	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._store_changed_flows')
		0 ignored issues – show Duplication introduced 2020-12-18 15:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
138		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
139		@patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
140		@patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
141		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
142		def test_install_flows(self, *args):
143		"""Test _install_flows method."""
144		(mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
145		mock_send_napp_event, _) = args
146		serializer = MagicMock()
147		flow = MagicMock()
148		flow_mod = MagicMock()
149
150		flow.as_of_add_flow_mod.return_value = flow_mod
151		serializer.from_dict.return_value = flow
152		mock_flow_factory.return_value = serializer
153
154		flows_dict = {'flows': [MagicMock()]}
155		switches = [self.switch_01]
156		self.napp._install_flows('add', flows_dict, switches)
157
158		mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
159		mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
160		flow, 'add')
161		mock_send_napp_event.assert_called_with(self.switch_01, flow, 'add')
162
163	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._store_changed_flows')
		0 ignored issues – show Duplication introduced 2020-12-18 15:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
164		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
165		@patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
166		@patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
167		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
168		def test_install_flows_with_delete_strict(self, *args):
169		"""Test _install_flows method with strict delete command."""
170		(mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
171		mock_send_napp_event, _) = args
172		serializer = MagicMock()
173		flow = MagicMock()
174		flow_mod = MagicMock()
175
176		flow.as_of_strict_delete_flow_mod.return_value = flow_mod
177		serializer.from_dict.return_value = flow
178		mock_flow_factory.return_value = serializer
179
180		flows_dict = {'flows': [MagicMock()]}
181		switches = [self.switch_01]
182		self.napp._install_flows('delete_strict', flows_dict, switches)
183
184		mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
185		mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
186		flow, 'delete_strict')
187		mock_send_napp_event.assert_called_with(self.switch_01, flow,
188		'delete_strict')
189
190		@patch('napps.kytos.flow_manager.main.Main._install_flows')
191		def test_event_add_flow(self, mock_install_flows):
192		"""Test event_add_flow method."""
193		dpid = "00:00:00:00:00:00:00:01"
194		mock_flow_dict = MagicMock()
195		event = get_kytos_event_mock(name='kytos.flow_manager.flows.install',
196		content={'dpid': dpid,
197		'command': 'add',
198		'flow_dict': mock_flow_dict})
199		self.napp.event_add_flow(event)
200		mock_install_flows.assert_called()
201
202		def test_add_flow_mod_sent(self):
203		"""Test _add_flow_mod_sent method."""
204		xid = 0
205		flow = MagicMock()
206
207		self.napp._add_flow_mod_sent(xid, flow, 'add')
208
209		self.assertEqual(self.napp._flow_mods_sent[xid], (flow, 'add'))
210
211		@patch('kytos.core.buffers.KytosEventBuffer.put')
212		def test_send_flow_mod(self, mock_buffers_put):
213		"""Test _send_flow_mod method."""
214		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
215		flow_mod = MagicMock()
216
217		self.napp._send_flow_mod(switch, flow_mod)
218
219		mock_buffers_put.assert_called()
220
221		@patch('kytos.core.buffers.KytosEventBuffer.put')
222		def test_send_napp_event(self, mock_buffers_put):
223		"""Test _send_napp_event method."""
224		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
225		flow = MagicMock()
226
227		for command in ['add', 'delete', 'delete_strict', 'error']:
228		self.napp._send_napp_event(switch, flow, command)
229
230		self.assertEqual(mock_buffers_put.call_count, 4)
231
232		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
233		def test_handle_errors(self, mock_send_napp_event):
234		"""Test handle_errors method."""
235		flow = MagicMock()
236		self.napp._flow_mods_sent[0] = (flow, 'add')
237
238		switch = get_switch_mock("00:00:00:00:00:00:00:01")
239		switch.connection = get_connection_mock(
240		0x04, get_switch_mock("00:00:00:00:00:00:00:02"))
241
242		protocol = MagicMock()
243		protocol.unpack.return_value = 'error_packet'
244
245		switch.connection.protocol = protocol
246
247		message = MagicMock()
248		message.header.xid.value = 0
249		message.error_type = 2
250		message.code = 5
251		event = get_kytos_event_mock(name='..of_core..ofpt_error',
252		content={'message': message,
253		'source': switch.connection})
254		self.napp.handle_errors(event)
255
256		mock_send_napp_event.assert_called_with(flow.switch, flow, 'error',
257		error_command='add',
258		error_code=5, error_type=2)
259
260		@patch("napps.kytos.flow_manager.main.StoreHouse.get_data")
261		def test_load_flows(self, mock_storehouse):
262		"""Test load flows."""
263		self.napp._load_flows()
264		mock_storehouse.assert_called()
265
266		@patch("napps.kytos.flow_manager.main.Main._install_flows")
267		def test_resend_stored_flows(self, mock_install_flows):
268		"""Test resend stored flows."""
269		dpid = "00:00:00:00:00:00:00:01"
270		switch = get_switch_mock(dpid, 0x04)
271		mock_event = MagicMock()
272		flow = {"command": "add", "flow": MagicMock()}
273
274		flows = {"flow_list": [flow]}
275		mock_event.content = {"switch": switch}
276		self.napp.controller.switches = {dpid: switch}
277		self.napp.stored_flows = {dpid: flows}
278		self.napp.resend_stored_flows(mock_event)
279		mock_install_flows.assert_called()
280
281		@patch("napps.kytos.of_core.flow.FlowFactory.get_class")
282		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
283		def test_store_changed_flows(self, mock_save_flow, _):
284		"""Test store changed flows."""
285		dpid = "00:00:00:00:00:00:00:01"
286		switch = get_switch_mock(dpid, 0x04)
287		switch.id = dpid
288		flow = {
289		"priority": 17,
290		"cookie": 84114964,
291		"command": "add",
292		"match": {"dl_dst": "00:15:af:d5:38:98"},
293		}
294		match_fields = {
295		"priority": 17,
296		"cookie": 84114964,
297		"command": "add",
298		"dl_dst": "00:15:af:d5:38:98",
299		}
300		flows = {"flow": flow}
301
302		command = "add"
303		flow_list = {
304		"flow_list": [
305		{"match_fields": match_fields, "command": "delete",
306		"flow": flow}
307		]
308		}
309		self.napp.stored_flows = {dpid: flow_list}
310		self.napp._store_changed_flows(command, flows, switch)
311		mock_save_flow.assert_called()
312
313		self.napp.stored_flows = {}
314		self.napp._store_changed_flows(command, flows, switch)
315		mock_save_flow.assert_called()
316
317	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2020-12-18 15:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
318		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
319		def test_check_switch_consistency_add(self, *args):
320		"""Test check_switch_consistency method.
321
322		This test checks the case when a flow is missing in switch and have the
323		ADD command.
324		"""
325		(mock_flow_factory, mock_install_flows) = args
326		dpid = "00:00:00:00:00:00:00:01"
327		switch = get_switch_mock(dpid, 0x04)
328		switch.flows = []
329
330		flow_1 = MagicMock()
331		flow_1.as_dict.return_value = {'flow_1': 'data'}
332
333		flow_list = [{"command": "add",
334		"flow": {'flow_1': 'data'}
335		}]
336		serializer = MagicMock()
337
338		mock_flow_factory.return_value = serializer
339		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
340		self.napp.check_switch_consistency(switch)
341		mock_install_flows.assert_called()
342
343	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2021-02-25 15:26 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
344		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
345		def test_check_switch_consistency_delete(self, *args):
346		"""Test check_switch_consistency method.
347
348		This test checks the case when a flow is missing in switch and have the
349		DELETE command.
350		"""
351		(mock_flow_factory, mock_install_flows) = args
352		dpid = "00:00:00:00:00:00:00:01"
353		switch = get_switch_mock(dpid, 0x04)
354
355		flow_1 = MagicMock()
356		flow_1.as_dict.return_value = {'flow_1': 'data'}
357
358		flow_list = [{"command": "delete",
359		"flow": {'flow_1': 'data'}
360		}]
361		serializer = MagicMock()
362		serializer.from_dict.return_value = flow_1
363
364		switch.flows = [flow_1]
365
366		mock_flow_factory.return_value = serializer
367		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
368		self.napp.check_switch_consistency(switch)
369		mock_install_flows.assert_called()
370
371	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2021-04-08 02:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
372		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
373		def test_check_storehouse_consistency(self, *args):
374		"""Test check_storehouse_consistency method.
375
376		This test checks the case when a flow is missing in storehouse.
377		"""
378		(mock_flow_factory, mock_install_flows) = args
379		dpid = "00:00:00:00:00:00:00:01"
380		switch = get_switch_mock(dpid, 0x04)
381
382		flow_1 = MagicMock()
383		flow_1.as_dict.return_value = {'flow_1': 'data'}
384
385		switch.flows = [flow_1]
386
387		flow_list = [{"command": "add",
388		"flow": {'flow_2': 'data'}
389		}]
390		serializer = MagicMock()
391
392		mock_flow_factory.return_value = serializer
393		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
394		self.napp.check_storehouse_consistency(switch)
395		mock_install_flows.assert_called()
396
397	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2021-02-25 15:26 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
398		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
399		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
400		def test_no_strict_delete(self, *args):
401		"""Test the non-strict matching method.
402
403		Test non-strict matching to delete a Flow using a cookie.
404		"""
405		(mock_save_flow, _, _) = args
406		dpid = "00:00:00:00:00:00:00:01"
407		switch = get_switch_mock(dpid, 0x04)
408		switch.id = dpid
409		stored_flow = {
410		"command": "add",
411		"flow": {
412		"actions": [{"action_type": "set_vlan", "vlan_id": 300}],
413		"cookie": 6191162389751548793,
414		"match": {"dl_vlan": 300, "in_port": 1},
415		},
416		}
417		stored_flow2 = {
418		"command": "add",
419		"flow": {
420		"actions": [],
421		"cookie": 4961162389751548787,
422		"match": {"in_port": 2},
423		},
424		}
425		flow_to_install = {
426		"cookie": 6191162389751548793,
427		"cookie_mask": 18446744073709551615,
428		}
429		flow_list = {"flow_list": [stored_flow, stored_flow2]}
430		command = "delete"
431		self.napp.stored_flows = {dpid: flow_list}
432
433		self.napp._store_changed_flows(command, flow_to_install, switch)
434		mock_save_flow.assert_called()
435		self.assertEqual(len(self.napp.stored_flows), 1)
436
437	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2021-04-22 14:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
438		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
439		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
440		def test_no_strict_delete_with_ipv4(self, *args):
441		"""Test the non-strict matching method.
442
443		Test non-strict matching to delete a Flow using IPv4.
444		"""
445		(mock_save_flow, _, _) = args
446		dpid = "00:00:00:00:00:00:00:01"
447		switch = get_switch_mock(dpid, 0x04)
448		switch.id = dpid
449		stored_flow = {
450		"command": "add",
451		"flow": {
452		"priority": 10,
453		"cookie": 84114904,
454		"match": {
455		"ipv4_src": "192.168.1.120",
456		"ipv4_dst": "192.168.0.2",
457		},
458		"actions": [],
459		},
460		}
461		stored_flow2 = {
462		"command": "add",
463		"flow": {
464		"actions": [],
465		"cookie": 4961162389751548787,
466		"match": {"in_port": 2},
467		},
468		}
469		flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
470		flow_list = {"flow_list": [stored_flow, stored_flow2]}
471		command = "delete"
472		self.napp.stored_flows = {dpid: flow_list}
473
474		self.napp._store_changed_flows(command, flow_to_install, switch)
475		mock_save_flow.assert_called()
476		self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 2)
477
478	View Code Duplication	@patch('napps.kytos.flow_manager.main.Main._install_flows')
		0 ignored issues – show Duplication introduced 2021-04-22 14:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
479		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
480		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
481		def test_no_strict_delete_with_ipv4_fail(self, *args):
482		"""Test the non-strict matching method.
483
484		Test non-strict Fail case matching to delete a Flow using IPv4.
485		"""
486		(mock_save_flow, _, _) = args
487		dpid = "00:00:00:00:00:00:00:01"
488		switch = get_switch_mock(dpid, 0x04)
489		switch.id = dpid
490		stored_flow = {
491		"command": "add",
492		"flow": {
493		"priority": 10,
494		"cookie": 84114904,
495		"match": {
496		"ipv4_src": "192.168.2.1",
497		"ipv4_dst": "192.168.0.2",
498		},
499		"actions": [],
500		},
501		}
502		stored_flow2 = {
503		"command": "add",
504		"flow": {
505		"actions": [],
506		"cookie": 4961162389751548787,
507		"match": {"in_port": 2},
508		},
509		}
510		flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
511		flow_list = {"flow_list": [stored_flow, stored_flow2]}
512		command = "delete"
513		self.napp.stored_flows = {dpid: flow_list}
514
515		self.napp._store_changed_flows(command, flow_to_install, switch)
516		mock_save_flow.assert_called()
517		self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 3)
518
519		@patch('napps.kytos.flow_manager.main.Main._install_flows')
520		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
521		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
522		def test_no_strict_delete_of10(self, *args):
523		"""Test the non-strict matching method.
524
525		Test non-strict matching to delete a Flow using OF10.
526		"""
527		(mock_save_flow, _, _) = args
528		dpid = "00:00:00:00:00:00:00:01"
529		switch = get_switch_mock(dpid, 0x01)
530		switch.id = dpid
531		stored_flow = {
532		"command": "add",
533		"flow": {
534		"actions": [{"max_len": 65535, "port": 6}],
535		"cookie": 4961162389751548787,
536		"match": {
537		"in_port": 80,
538		"dl_src": "00:00:00:00:00:00",
539		"dl_dst": "f2:0b:a4:7d:f8:ea",
540		"dl_vlan": 0,
541		"dl_vlan_pcp": 0,
542		"dl_type": 0,
543		"nw_tos": 0,
544		"nw_proto": 0,
545		"nw_src": "192.168.0.1",
546		"nw_dst": "0.0.0.0",
547		"tp_src": 0,
548		"tp_dst": 0,
549		},
550		"out_port": 65532,
551		"priority": 123,
552		},
553		}
554		stored_flow2 = {
555		"command": "add",
556		"flow": {
557		"actions": [],
558		"cookie": 4961162389751654,
559		"match": {
560		"in_port": 2,
561		"dl_src": "00:00:00:00:00:00",
562		"dl_dst": "f2:0b:a4:7d:f8:ea",
563		"dl_vlan": 0,
564		"dl_vlan_pcp": 0,
565		"dl_type": 0,
566		"nw_tos": 0,
567		"nw_proto": 0,
568		"nw_src": "192.168.0.1",
569		"nw_dst": "0.0.0.0",
570		"tp_src": 0,
571		"tp_dst": 0,
572		},
573		"out_port": 655,
574		"priority": 1,
575		},
576		}
577		flow_to_install = {"match": {"in_port": 80, "wildcards": 4194303}}
578		flow_list = {"flow_list": [stored_flow, stored_flow2]}
579		command = "delete"
580		self.napp.stored_flows = {dpid: flow_list}
581
582		self.napp._store_changed_flows(command, flow_to_install, switch)
583		mock_save_flow.assert_called()
584		self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 1)
585

kytos / flow_manager

Pull Request — master (#129)

TestMain.test_event_add_flow() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like