build.tests.unit.test_main.TestMain.test_no_strict_delete() - Code Metrics - Inspection of "Add support for non-strict exclusion in the consis..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#117)

by Carlos

created 2021-04-28 14:37 UTC

TestMain.test_no_strict_delete() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	39
Code Lines	29

Duplication

Lines	39
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	29
nop	2
dl	39
loc	39
rs	9.184
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_1 = MagicMock()
        flow_1.as_dict.return_value = {'flow_1': 'data'}
        flow_2 = MagicMock()
        flow_2.as_dict.return_value = {'flow_2': 'data'}
        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_1': 'data'}]},
                    '00:00:00:00:00:00:00:02': {'flows': [{'flow_2': 'data'}]}}

        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_1 = MagicMock()
        flow_1.as_dict.return_value = {'flow_1': 'data'}
        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_1': 'data'}]}}

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

    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)


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_1 = MagicMock()
40		flow_1.as_dict.return_value = {'flow_1': 'data'}
41		flow_2 = MagicMock()
42		flow_2.as_dict.return_value = {'flow_2': 'data'}
43		self.switch_01.flows.append(flow_1)
44		self.switch_02.flows.append(flow_2)
45
46		api = get_test_client(self.napp.controller, self.napp)
47		url = f'{self.API_URL}/v2/flows'
48
49		response = api.get(url)
50		expected = {'00:00:00:00:00:00:00:01': {'flows': [{'flow_1': 'data'}]},
51		'00:00:00:00:00:00:00:02': {'flows': [{'flow_2': 'data'}]}}
52
53		self.assertEqual(response.json, expected)
54		self.assertEqual(response.status_code, 200)
55
56		def test_rest_list_with_dpid(self):
57		"""Test list rest method with dpid."""
58		flow_1 = MagicMock()
59		flow_1.as_dict.return_value = {'flow_1': 'data'}
60		self.switch_01.flows.append(flow_1)
61
62		api = get_test_client(self.napp.controller, self.napp)
63		url = f'{self.API_URL}/v2/flows/00:00:00:00:00:00:00:01'
64
65		response = api.get(url)
66		expected = {'00:00:00:00:00:00:00:01': {'flows': [{'flow_1': 'data'}]}}
67
68		self.assertEqual(response.json, expected)
69		self.assertEqual(response.status_code, 200)
70
71		@patch('napps.kytos.flow_manager.main.Main._install_flows')
72		def test_rest_add_and_delete_without_dpid(self, mock_install_flows):
73		"""Test add and delete rest method without dpid."""
74		api = get_test_client(self.napp.controller, self.napp)
75
76		for method in ['flows', 'delete']:
77		url = f'{self.API_URL}/v2/{method}'
78
79		response_1 = api.post(url, json={'data': '123'})
80		response_2 = api.post(url)
81
82		self.assertEqual(response_1.status_code, 200)
83		self.assertEqual(response_2.status_code, 404)
84
85		self.assertEqual(mock_install_flows.call_count, 2)
86
87		@patch('napps.kytos.flow_manager.main.Main._install_flows')
88		def test_rest_add_and_delete_with_dpid(self, mock_install_flows):
89		"""Test add and delete rest method with dpid."""
90		api = get_test_client(self.napp.controller, self.napp)
91
92		for method in ['flows', 'delete']:
93		url_1 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:01'
94		url_2 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:02'
95		url_3 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:03'
96
97		response_1 = api.post(url_1)
98		response_2 = api.post(url_1, json={'data': '123'})
99		response_3 = api.post(url_2, json={'data': '123'})
100		response_4 = api.post(url_3, json={'data': '123'})
101
102		self.assertEqual(response_1.status_code, 404)
103		self.assertEqual(response_2.status_code, 200)
104		if method == 'flows':
105		self.assertEqual(response_3.status_code, 404)
106		else:
107		self.assertEqual(response_3.status_code, 200)
108		self.assertEqual(response_4.status_code, 404)
109
110		self.assertEqual(mock_install_flows.call_count, 3)
111
112		def test_get_all_switches_enabled(self):
113		"""Test _get_all_switches_enabled method."""
114		switches = self.napp._get_all_switches_enabled()
115
116		self.assertEqual(switches, [self.switch_01])
117
118	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...
119		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
120		@patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
121		@patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
122		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
123		def test_install_flows(self, *args):
124		"""Test _install_flows method."""
125		(mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
126		mock_send_napp_event, _) = args
127		serializer = MagicMock()
128		flow = MagicMock()
129		flow_mod = MagicMock()
130
131		flow.as_of_add_flow_mod.return_value = flow_mod
132		serializer.from_dict.return_value = flow
133		mock_flow_factory.return_value = serializer
134
135		flows_dict = {'flows': [MagicMock()]}
136		switches = [self.switch_01]
137		self.napp._install_flows('add', flows_dict, switches)
138
139		mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
140		mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
141		flow, 'add')
142		mock_send_napp_event.assert_called_with(self.switch_01, flow, 'add')
143
144	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...
145		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
146		@patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
147		@patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
148		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
149		def test_install_flows_with_delete_strict(self, *args):
150		"""Test _install_flows method with strict delete command."""
151		(mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
152		mock_send_napp_event, _) = args
153		serializer = MagicMock()
154		flow = MagicMock()
155		flow_mod = MagicMock()
156
157		flow.as_of_strict_delete_flow_mod.return_value = flow_mod
158		serializer.from_dict.return_value = flow
159		mock_flow_factory.return_value = serializer
160
161		flows_dict = {'flows': [MagicMock()]}
162		switches = [self.switch_01]
163		self.napp._install_flows('delete_strict', flows_dict, switches)
164
165		mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
166		mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
167		flow, 'delete_strict')
168		mock_send_napp_event.assert_called_with(self.switch_01, flow,
169		'delete_strict')
170
171		def test_add_flow_mod_sent(self):
172		"""Test _add_flow_mod_sent method."""
173		xid = 0
174		flow = MagicMock()
175
176		self.napp._add_flow_mod_sent(xid, flow, 'add')
177
178		self.assertEqual(self.napp._flow_mods_sent[xid], (flow, 'add'))
179
180		@patch('kytos.core.buffers.KytosEventBuffer.put')
181		def test_send_flow_mod(self, mock_buffers_put):
182		"""Test _send_flow_mod method."""
183		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
184		flow_mod = MagicMock()
185
186		self.napp._send_flow_mod(switch, flow_mod)
187
188		mock_buffers_put.assert_called()
189
190		@patch('kytos.core.buffers.KytosEventBuffer.put')
191		def test_send_napp_event(self, mock_buffers_put):
192		"""Test _send_napp_event method."""
193		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
194		flow = MagicMock()
195
196		for command in ['add', 'delete', 'delete_strict', 'error']:
197		self.napp._send_napp_event(switch, flow, command)
198
199		self.assertEqual(mock_buffers_put.call_count, 4)
200
201		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
202		def test_handle_errors(self, mock_send_napp_event):
203		"""Test handle_errors method."""
204		flow = MagicMock()
205		self.napp._flow_mods_sent[0] = (flow, 'add')
206
207		switch = get_switch_mock("00:00:00:00:00:00:00:01")
208		switch.connection = get_connection_mock(
209		0x04, get_switch_mock("00:00:00:00:00:00:00:02"))
210
211		protocol = MagicMock()
212		protocol.unpack.return_value = 'error_packet'
213
214		switch.connection.protocol = protocol
215
216		message = MagicMock()
217		message.header.xid.value = 0
218		message.error_type = 2
219		message.code = 5
220		event = get_kytos_event_mock(name='..of_core..ofpt_error',
221		content={'message': message,
222		'source': switch.connection})
223		self.napp.handle_errors(event)
224
225		mock_send_napp_event.assert_called_with(flow.switch, flow, 'error',
226		error_command='add',
227		error_code=5, error_type=2)
228
229		@patch("napps.kytos.flow_manager.main.StoreHouse.get_data")
230		def test_load_flows(self, mock_storehouse):
231		"""Test load flows."""
232		self.napp._load_flows()
233		mock_storehouse.assert_called()
234
235		@patch("napps.kytos.flow_manager.main.Main._install_flows")
236		def test_resend_stored_flows(self, mock_install_flows):
237		"""Test resend stored flows."""
238		dpid = "00:00:00:00:00:00:00:01"
239		switch = get_switch_mock(dpid, 0x04)
240		mock_event = MagicMock()
241		flow = {"command": "add", "flow": MagicMock()}
242
243		flows = {"flow_list": [flow]}
244		mock_event.content = {"switch": switch}
245		self.napp.controller.switches = {dpid: switch}
246		self.napp.stored_flows = {dpid: flows}
247		self.napp.resend_stored_flows(mock_event)
248		mock_install_flows.assert_called()
249
250		@patch("napps.kytos.of_core.flow.FlowFactory.get_class")
251		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
252		def test_store_changed_flows(self, mock_save_flow, _):
253		"""Test store changed flows."""
254		dpid = "00:00:00:00:00:00:00:01"
255		switch = get_switch_mock(dpid, 0x04)
256		switch.id = dpid
257		flow = {
258		"priority": 17,
259		"cookie": 84114964,
260		"command": "add",
261		"match": {"dl_dst": "00:15:af:d5:38:98"},
262		}
263		match_fields = {
264		"priority": 17,
265		"cookie": 84114964,
266		"command": "add",
267		"dl_dst": "00:15:af:d5:38:98",
268		}
269		flows = {"flow": flow}
270
271		command = "add"
272		flow_list = {
273		"flow_list": [
274		{"match_fields": match_fields, "command": "delete",
275		"flow": flow}
276		]
277		}
278		self.napp.stored_flows = {dpid: flow_list}
279		self.napp._store_changed_flows(command, flows, switch)
280		mock_save_flow.assert_called()
281
282		self.napp.stored_flows = {}
283		self.napp._store_changed_flows(command, flows, switch)
284		mock_save_flow.assert_called()
285
286	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...
287		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
288		def test_check_switch_consistency_add(self, *args):
289		"""Test check_switch_consistency method.
290
291		This test checks the case when a flow is missing in switch and have the
292		ADD command.
293		"""
294		(mock_flow_factory, mock_install_flows) = args
295		dpid = "00:00:00:00:00:00:00:01"
296		switch = get_switch_mock(dpid, 0x04)
297		switch.flows = []
298
299		flow_1 = MagicMock()
300		flow_1.as_dict.return_value = {'flow_1': 'data'}
301
302		flow_list = [{"command": "add",
303		"flow": {'flow_1': 'data'}
304		}]
305		serializer = MagicMock()
306
307		mock_flow_factory.return_value = serializer
308		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
309		self.napp.check_switch_consistency(switch)
310		mock_install_flows.assert_called()
311
312	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...
313		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
314		def test_check_switch_consistency_delete(self, *args):
315		"""Test check_switch_consistency method.
316
317		This test checks the case when a flow is missing in switch and have the
318		DELETE command.
319		"""
320		(mock_flow_factory, mock_install_flows) = args
321		dpid = "00:00:00:00:00:00:00:01"
322		switch = get_switch_mock(dpid, 0x04)
323
324		flow_1 = MagicMock()
325		flow_1.as_dict.return_value = {'flow_1': 'data'}
326
327		flow_list = [{"command": "delete",
328		"flow": {'flow_1': 'data'}
329		}]
330		serializer = MagicMock()
331		serializer.from_dict.return_value = flow_1
332
333		switch.flows = [flow_1]
334
335		mock_flow_factory.return_value = serializer
336		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
337		self.napp.check_switch_consistency(switch)
338		mock_install_flows.assert_called()
339
340	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...
341		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
342		def test_check_storehouse_consistency(self, *args):
343		"""Test check_storehouse_consistency method.
344
345		This test checks the case when a flow is missing in storehouse.
346		"""
347		(mock_flow_factory, mock_install_flows) = args
348		dpid = "00:00:00:00:00:00:00:01"
349		switch = get_switch_mock(dpid, 0x04)
350
351		flow_1 = MagicMock()
352		flow_1.as_dict.return_value = {'flow_1': 'data'}
353
354		switch.flows = [flow_1]
355
356		flow_list = [{"command": "add",
357		"flow": {'flow_2': 'data'}
358		}]
359		serializer = MagicMock()
360
361		mock_flow_factory.return_value = serializer
362		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
363		self.napp.check_storehouse_consistency(switch)
364		mock_install_flows.assert_called()
365
366	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...
367		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
368		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
369		def test_no_strict_delete(self, *args):
370		"""Test the non-strict matching method.
371
372		Test non-strict matching to delete a Flow using a cookie.
373		"""
374		(mock_save_flow, _, _) = args
375		dpid = "00:00:00:00:00:00:00:01"
376		switch = get_switch_mock(dpid, 0x04)
377		switch.id = dpid
378		stored_flow = {
379		"command": "add",
380		"flow": {
381		"actions": [{"action_type": "set_vlan", "vlan_id": 300}],
382		"cookie": 6191162389751548793,
383		"match": {"dl_vlan": 300, "in_port": 1},
384		},
385		}
386		stored_flow2 = {
387		"command": "add",
388		"flow": {
389		"actions": [],
390		"cookie": 4961162389751548787,
391		"match": {"in_port": 2},
392		},
393		}
394		flow_to_install = {
395		"cookie": 6191162389751548793,
396		"cookie_mask": 18446744073709551615,
397		}
398		flow_list = {"flow_list": [stored_flow, stored_flow2]}
399		command = "delete"
400		self.napp.stored_flows = {dpid: flow_list}
401
402		self.napp._store_changed_flows(command, flow_to_install, switch)
403		mock_save_flow.assert_called()
404		self.assertEqual(len(self.napp.stored_flows), 1)
405
406	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...
407		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
408		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
409		def test_no_strict_delete_with_ipv4(self, *args):
410		"""Test the non-strict matching method.
411
412		Test non-strict matching to delete a Flow using IPv4.
413		"""
414		(mock_save_flow, _, _) = args
415		dpid = "00:00:00:00:00:00:00:01"
416		switch = get_switch_mock(dpid, 0x04)
417		switch.id = dpid
418		stored_flow = {
419		"command": "add",
420		"flow": {
421		"priority": 10,
422		"cookie": 84114904,
423		"match": {
424		"ipv4_src": "192.168.1.120",
425		"ipv4_dst": "192.168.0.2",
426		},
427		"actions": [],
428		},
429		}
430		stored_flow2 = {
431		"command": "add",
432		"flow": {
433		"actions": [],
434		"cookie": 4961162389751548787,
435		"match": {"in_port": 2},
436		},
437		}
438		flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
439		flow_list = {"flow_list": [stored_flow, stored_flow2]}
440		command = "delete"
441		self.napp.stored_flows = {dpid: flow_list}
442
443		self.napp._store_changed_flows(command, flow_to_install, switch)
444		mock_save_flow.assert_called()
445		self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 2)
446
447	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...
448		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
449		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
450		def test_no_strict_delete_with_ipv4_fail(self, *args):
451		"""Test the non-strict matching method.
452
453		Test non-strict Fail case matching to delete a Flow using IPv4.
454		"""
455		(mock_save_flow, _, _) = args
456		dpid = "00:00:00:00:00:00:00:01"
457		switch = get_switch_mock(dpid, 0x04)
458		switch.id = dpid
459		stored_flow = {
460		"command": "add",
461		"flow": {
462		"priority": 10,
463		"cookie": 84114904,
464		"match": {
465		"ipv4_src": "192.168.2.1",
466		"ipv4_dst": "192.168.0.2",
467		},
468		"actions": [],
469		},
470		}
471		stored_flow2 = {
472		"command": "add",
473		"flow": {
474		"actions": [],
475		"cookie": 4961162389751548787,
476		"match": {"in_port": 2},
477		},
478		}
479		flow_to_install = {"match": {"ipv4_src": '192.168.1.1/24'}}
480		flow_list = {"flow_list": [stored_flow, stored_flow2]}
481		command = "delete"
482		self.napp.stored_flows = {dpid: flow_list}
483
484		self.napp._store_changed_flows(command, flow_to_install, switch)
485		mock_save_flow.assert_called()
486		self.assertEqual(len(self.napp.stored_flows[dpid]['flow_list']), 3)
487

kytos / flow_manager

Pull Request — master (#117)

TestMain.test_no_strict_delete() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like