build.tests.unit.test_main.TestMain.test_flows_change() - Code Metrics - Inspection of "Added flow persistence" - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#90)

unknown

created 2020-10-13 20:33 UTC

TestMain.test_flows_change() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	33
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	26
nop	2
dl	0
loc	33
rs	9.256
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_controller_mock, get_kytos_event_mock,
                               get_switch_mock, get_test_client)


# pylint: disable=protected-access
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=bad-option-value
        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)
            self.assertEqual(response_3.status_code, 404)
            self.assertEqual(response_4.status_code, 404)

        self.assertEqual(mock_install_flows.call_count, 2)

    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._flows_change")
    @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)
        mock_send_napp_event.assert_called_with(self.switch_01, flow, "add")

    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)

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

    @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", "error"]:
            self.napp._send_napp_event(switch, flow, command)

        self.assertEqual(mock_buffers_put.call_count, 3)

    @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

        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}
        )
        self.napp.handle_errors(event)

        mock_send_napp_event.assert_called_with(
            flow.switch, flow, "error", 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", "data": MagicMock()}

        flows = {"flow_list": [flow]}
        mock_event.content = {"switch": dpid}
        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.flow_manager.main.StoreHouse.save_flow")
    def test_flows_change(self, mock_save_flow):
        """Test flows change."""
        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 = {"flows": [flow]}

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

        self.napp.stored_flows = {}
        self.napp._flows_change(command, flows, [switch])
        mock_save_flow.assert_called()


1			"""Test Main methods."""
2			from unittest import TestCase
3			from unittest.mock import MagicMock, patch
4
5			from kytos.lib.helpers import (get_controller_mock, get_kytos_event_mock,
6			get_switch_mock, get_test_client)
7
8
9			# pylint: disable=protected-access
10			class TestMain(TestCase):
11			"""Tests for the Main class."""
12
13			API_URL = "http://localhost:8181/api/kytos/flow_manager"
14
15			def setUp(self):
16			patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
17			# pylint: disable=bad-option-value
18			from napps.kytos.flow_manager.main import Main
19
20			self.addCleanup(patch.stopall)
21
22			controller = get_controller_mock()
23			self.switch_01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
24			self.switch_01.is_enabled.return_value = True
25			self.switch_01.flows = []
26
27			self.switch_02 = get_switch_mock("00:00:00:00:00:00:00:02", 0x04)
28			self.switch_02.is_enabled.return_value = False
29			self.switch_02.flows = []
30
31			controller.switches = {
32			"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
36			self.napp = Main(controller)
37
38			def test_rest_list_without_dpid(self):
39			"""Test list rest method withoud dpid."""
40			flow_1 = MagicMock()
41			flow_1.as_dict.return_value = {"flow_1": "data"}
42			flow_2 = MagicMock()
43			flow_2.as_dict.return_value = {"flow_2": "data"}
44			self.switch_01.flows.append(flow_1)
45			self.switch_02.flows.append(flow_2)
46
47			api = get_test_client(self.napp.controller, self.napp)
48			url = f"{self.API_URL}/v2/flows"
49
50			response = api.get(url)
51			expected = {
52			"00:00:00:00:00:00:00:01": {"flows": [{"flow_1": "data"}]},
53			"00:00:00:00:00:00:00:02": {"flows": [{"flow_2": "data"}]},
54			}
55
56			self.assertEqual(response.json, expected)
57			self.assertEqual(response.status_code, 200)
58
59			def test_rest_list_with_dpid(self):
60			"""Test list rest method with dpid."""
61			flow_1 = MagicMock()
62			flow_1.as_dict.return_value = {"flow_1": "data"}
63			self.switch_01.flows.append(flow_1)
64
65			api = get_test_client(self.napp.controller, self.napp)
66			url = f"{self.API_URL}/v2/flows/00:00:00:00:00:00:00:01"
67
68			response = api.get(url)
69			expected = {"00:00:00:00:00:00:00:01": {"flows": [{"flow_1": "data"}]}}
70
71			self.assertEqual(response.json, expected)
72			self.assertEqual(response.status_code, 200)
73
74			@patch("napps.kytos.flow_manager.main.Main._install_flows")
75			def test_rest_add_and_delete_without_dpid(self, mock_install_flows):
76			"""Test add and delete rest method without dpid."""
77			api = get_test_client(self.napp.controller, self.napp)
78
79			for method in ["flows", "delete"]:
80			url = f"{self.API_URL}/v2/{method}"
81
82			response_1 = api.post(url, json={"data": "123"})
83			response_2 = api.post(url)
84
85			self.assertEqual(response_1.status_code, 200)
86			self.assertEqual(response_2.status_code, 404)
87
88			self.assertEqual(mock_install_flows.call_count, 2)
89
90			@patch("napps.kytos.flow_manager.main.Main._install_flows")
91			def test_rest_add_and_delete_with_dpid(self, mock_install_flows):
92			"""Test add and delete rest method with dpid."""
93			api = get_test_client(self.napp.controller, self.napp)
94
95			for method in ["flows", "delete"]:
96			url_1 = f"{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:01"
97			url_2 = f"{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:02"
98			url_3 = f"{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:03"
99
100			response_1 = api.post(url_1)
101			response_2 = api.post(url_1, json={"data": "123"})
102			response_3 = api.post(url_2, json={"data": "123"})
103			response_4 = api.post(url_3, json={"data": "123"})
104
105			self.assertEqual(response_1.status_code, 404)
106			self.assertEqual(response_2.status_code, 200)
107			self.assertEqual(response_3.status_code, 404)
108			self.assertEqual(response_4.status_code, 404)
109
110			self.assertEqual(mock_install_flows.call_count, 2)
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			@patch("napps.kytos.flow_manager.main.Main._flows_change")
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			(
126			mock_flow_factory,
127			mock_send_flow_mod,
128			mock_add_flow_mod_sent,
129			mock_send_napp_event,
130			_,
131			) = args
132			serializer = MagicMock()
133			flow = MagicMock()
134			flow_mod = MagicMock()
135
136			flow.as_of_add_flow_mod.return_value = flow_mod
137			serializer.from_dict.return_value = flow
138			mock_flow_factory.return_value = serializer
139
140			flows_dict = {"flows": [MagicMock()]}
141			switches = [self.switch_01]
142			self.napp._install_flows("add", flows_dict, switches)
143
144			mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
145			mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid, flow)
146			mock_send_napp_event.assert_called_with(self.switch_01, flow, "add")
147
148			def test_add_flow_mod_sent(self):
149			"""Test _add_flow_mod_sent method."""
150			xid = 0
151			flow = MagicMock()
152
153			self.napp._add_flow_mod_sent(xid, flow)
154
155			self.assertEqual(self.napp._flow_mods_sent[xid], flow)
156
157			@patch("kytos.core.buffers.KytosEventBuffer.put")
158			def test_send_flow_mod(self, mock_buffers_put):
159			"""Test _send_flow_mod method."""
160			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
161			flow_mod = MagicMock()
162
163			self.napp._send_flow_mod(switch, flow_mod)
164
165			mock_buffers_put.assert_called()
166
167			@patch("kytos.core.buffers.KytosEventBuffer.put")
168			def test_send_napp_event(self, mock_buffers_put):
169			"""Test _send_napp_event method."""
170			switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
171			flow = MagicMock()
172
173			for command in ["add", "delete", "error"]:
174			self.napp._send_napp_event(switch, flow, command)
175
176			self.assertEqual(mock_buffers_put.call_count, 3)
177
178			@patch("napps.kytos.flow_manager.main.Main._send_napp_event")
179			def test_handle_errors(self, mock_send_napp_event):
180			"""Test handle_errors method."""
181			flow = MagicMock()
182			self.napp._flow_mods_sent[0] = flow
183
184			message = MagicMock()
185			message.header.xid.value = 0
186			message.error_type = 2
187			message.code = 5
188			event = get_kytos_event_mock(
189			name="..of_core..ofpt_error", content={"message": message}
190			)
191			self.napp.handle_errors(event)
192
193			mock_send_napp_event.assert_called_with(
194			flow.switch, flow, "error", error_code=5, error_type=2
195			)
196
197			@patch("napps.kytos.flow_manager.main.StoreHouse.get_data")
198			def test_load_flows(self, mock_storehouse):
199			"""Test load flows."""
200			self.napp._load_flows()
201			mock_storehouse.assert_called()
202
203			@patch("napps.kytos.flow_manager.main.Main._install_flows")
204			def test_resend_stored_flows(self, mock_install_flows):
205			"""Test resend stored flows."""
206			dpid = "00:00:00:00:00:00:00:01"
207			switch = get_switch_mock(dpid, 0x04)
208			mock_event = MagicMock()
209			flow = {"command": "add", "data": MagicMock()}
210
211			flows = {"flow_list": [flow]}
212			mock_event.content = {"switch": dpid}
213			self.napp.controller.switches = {dpid: switch}
214			self.napp.stored_flows = {dpid: flows}
215			self.napp.resend_stored_flows(mock_event)
216			mock_install_flows.assert_called()
217
218			@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
219			def test_flows_change(self, mock_save_flow):
220			"""Test flows change."""
221			dpid = "00:00:00:00:00:00:00:01"
222			switch = get_switch_mock(dpid, 0x04)
223			switch.id = dpid
224			flow = {
225			"priority": 17,
226			"cookie": 84114964,
227			"command": "add",
228			"match": {"dl_dst": "00:15:af:d5:38:98"},
229			}
230			match_fields = {
231			"priority": 17,
232			"cookie": 84114964,
233			"command": "add",
234			"dl_dst": "00:15:af:d5:38:98",
235			}
236			flows = {"flows": [flow]}
237
238			command = "add"
239			flow_list = {
240			"flow_list": [
241			{"match_fields": match_fields, "command": "add", "data": flows}
242			]
243			}
244			self.napp.stored_flows = {dpid: flow_list}
245			self.napp._flows_change(command, flows, [switch])
246			mock_save_flow.assert_called()
247
248			self.napp.stored_flows = {}
249			self.napp._flows_change(command, flows, [switch])
250			mock_save_flow.assert_called()
251

kytos / flow_manager

Pull Request — master (#90)

TestMain.test_flows_change() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like