build.tests.unit.test_main.TestMain.test_send_napp_event() - Code Metrics - Inspection of "Add consistency" - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#96)

by Jose

created 2020-12-18 15:40 UTC

TestMain.test_send_napp_event() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	10
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	10
rs	10
c	0
b	0
f	0
cc	2
nop	2

"""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._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')

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

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


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 = {"00:00:00:00:00:00:00:01": self.switch_01,
32		"00:00:00:00:00:00:00:02": self.switch_02}
33
34		self.napp = Main(controller)
35
36		def test_rest_list_without_dpid(self):
37		"""Test list rest method withoud dpid."""
38		flow_1 = MagicMock()
39		flow_1.as_dict.return_value = {'flow_1': 'data'}
40		flow_2 = MagicMock()
41		flow_2.as_dict.return_value = {'flow_2': 'data'}
42		self.switch_01.flows.append(flow_1)
43		self.switch_02.flows.append(flow_2)
44
45		api = get_test_client(self.napp.controller, self.napp)
46		url = f'{self.API_URL}/v2/flows'
47
48		response = api.get(url)
49		expected = {'00:00:00:00:00:00:00:01': {'flows': [{'flow_1': 'data'}]},
50		'00:00:00:00:00:00:00:02': {'flows': [{'flow_2': 'data'}]}}
51
52		self.assertEqual(response.json, expected)
53		self.assertEqual(response.status_code, 200)
54
55		def test_rest_list_with_dpid(self):
56		"""Test list rest method with dpid."""
57		flow_1 = MagicMock()
58		flow_1.as_dict.return_value = {'flow_1': 'data'}
59		self.switch_01.flows.append(flow_1)
60
61		api = get_test_client(self.napp.controller, self.napp)
62		url = f'{self.API_URL}/v2/flows/00:00:00:00:00:00:00:01'
63
64		response = api.get(url)
65		expected = {'00:00:00:00:00:00:00:01': {'flows': [{'flow_1': 'data'}]}}
66
67		self.assertEqual(response.json, expected)
68		self.assertEqual(response.status_code, 200)
69
70		@patch('napps.kytos.flow_manager.main.Main._install_flows')
71		def test_rest_add_and_delete_without_dpid(self, mock_install_flows):
72		"""Test add and delete rest method without dpid."""
73		api = get_test_client(self.napp.controller, self.napp)
74
75		for method in ['flows', 'delete']:
76		url = f'{self.API_URL}/v2/{method}'
77
78		response_1 = api.post(url, json={'data': '123'})
79		response_2 = api.post(url)
80
81		self.assertEqual(response_1.status_code, 200)
82		self.assertEqual(response_2.status_code, 404)
83
84		self.assertEqual(mock_install_flows.call_count, 2)
85
86		@patch('napps.kytos.flow_manager.main.Main._install_flows')
87		def test_rest_add_and_delete_with_dpid(self, mock_install_flows):
88		"""Test add and delete rest method with dpid."""
89		api = get_test_client(self.napp.controller, self.napp)
90
91		for method in ['flows', 'delete']:
92		url_1 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:01'
93		url_2 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:02'
94		url_3 = f'{self.API_URL}/v2/{method}/00:00:00:00:00:00:00:03'
95
96		response_1 = api.post(url_1)
97		response_2 = api.post(url_1, json={'data': '123'})
98		response_3 = api.post(url_2, json={'data': '123'})
99		response_4 = api.post(url_3, json={'data': '123'})
100
101		self.assertEqual(response_1.status_code, 404)
102		self.assertEqual(response_2.status_code, 200)
103		self.assertEqual(response_3.status_code, 404)
104		self.assertEqual(response_4.status_code, 404)
105
106		self.assertEqual(mock_install_flows.call_count, 2)
107
108		def test_get_all_switches_enabled(self):
109		"""Test _get_all_switches_enabled method."""
110		switches = self.napp._get_all_switches_enabled()
111
112		self.assertEqual(switches, [self.switch_01])
113
114		@patch('napps.kytos.flow_manager.main.Main._store_changed_flows')
115		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
116		@patch('napps.kytos.flow_manager.main.Main._add_flow_mod_sent')
117		@patch('napps.kytos.flow_manager.main.Main._send_flow_mod')
118		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
119		def test_install_flows(self, *args):
120		"""Test _install_flows method."""
121		(mock_flow_factory, mock_send_flow_mod, mock_add_flow_mod_sent,
122		mock_send_napp_event, _) = args
123		serializer = MagicMock()
124		flow = MagicMock()
125		flow_mod = MagicMock()
126
127		flow.as_of_add_flow_mod.return_value = flow_mod
128		serializer.from_dict.return_value = flow
129		mock_flow_factory.return_value = serializer
130
131		flows_dict = {'flows': [MagicMock()]}
132		switches = [self.switch_01]
133		self.napp._install_flows('add', flows_dict, switches)
134
135		mock_send_flow_mod.assert_called_with(flow.switch, flow_mod)
136		mock_add_flow_mod_sent.assert_called_with(flow_mod.header.xid,
137		flow, 'add')
138		mock_send_napp_event.assert_called_with(self.switch_01, flow, 'add')
139
140		def test_add_flow_mod_sent(self):
141		"""Test _add_flow_mod_sent method."""
142		xid = 0
143		flow = MagicMock()
144
145		self.napp._add_flow_mod_sent(xid, flow, 'add')
146
147		self.assertEqual(self.napp._flow_mods_sent[xid], (flow, 'add'))
148
149		@patch('kytos.core.buffers.KytosEventBuffer.put')
150		def test_send_flow_mod(self, mock_buffers_put):
151		"""Test _send_flow_mod method."""
152		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
153		flow_mod = MagicMock()
154
155		self.napp._send_flow_mod(switch, flow_mod)
156
157		mock_buffers_put.assert_called()
158
159		@patch('kytos.core.buffers.KytosEventBuffer.put')
160		def test_send_napp_event(self, mock_buffers_put):
161		"""Test _send_napp_event method."""
162		switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
163		flow = MagicMock()
164
165		for command in ['add', 'delete', 'error']:
166		self.napp._send_napp_event(switch, flow, command)
167
168		self.assertEqual(mock_buffers_put.call_count, 3)
169
170		@patch('napps.kytos.flow_manager.main.Main._send_napp_event')
171		def test_handle_errors(self, mock_send_napp_event):
172		"""Test handle_errors method."""
173		flow = MagicMock()
174		self.napp._flow_mods_sent[0] = (flow, 'add')
175
176		message = MagicMock()
177		message.header.xid.value = 0
178		message.error_type = 2
179		message.code = 5
180		event = get_kytos_event_mock(name='..of_core..ofpt_error',
181		content={'message': message})
182		self.napp.handle_errors(event)
183
184		mock_send_napp_event.assert_called_with(flow.switch, flow, 'error',
185		error_command='add',
186		error_code=5, error_type=2)
187
188		@patch("napps.kytos.flow_manager.main.StoreHouse.get_data")
189		def test_load_flows(self, mock_storehouse):
190		"""Test load flows."""
191		self.napp._load_flows()
192		mock_storehouse.assert_called()
193
194		@patch("napps.kytos.of_core.flow.FlowFactory.get_class")
195		@patch("napps.kytos.flow_manager.main.StoreHouse.save_flow")
196		def test_store_changed_flows(self, mock_save_flow, _):
197		"""Test store changed flows."""
198		dpid = "00:00:00:00:00:00:00:01"
199		switch = get_switch_mock(dpid, 0x04)
200		switch.id = dpid
201		flow = {
202		"priority": 17,
203		"cookie": 84114964,
204		"command": "add",
205		"match": {"dl_dst": "00:15:af:d5:38:98"},
206		}
207		match_fields = {
208		"priority": 17,
209		"cookie": 84114964,
210		"command": "add",
211		"dl_dst": "00:15:af:d5:38:98",
212		}
213		flows = {"flow": flow}
214
215		command = "add"
216		flow_list = {
217		"flow_list": [
218		{"match_fields": match_fields, "command": "delete",
219		"flow": flow}
220		]
221		}
222		self.napp.stored_flows = {dpid: flow_list}
223		self.napp._store_changed_flows(command, flows, switch)
224		mock_save_flow.assert_called()
225
226		self.napp.stored_flows = {}
227		self.napp._store_changed_flows(command, flows, switch)
228		mock_save_flow.assert_called()
229
230	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...
231		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
232		def test_check_switch_consistency_add(self, *args):
233		"""Test check_switch_consistency method.
234
235		This test checks the case when a flow is missing in switch and have the
236		ADD command.
237		"""
238		(mock_flow_factory, mock_install_flows) = args
239		dpid = "00:00:00:00:00:00:00:01"
240		switch = get_switch_mock(dpid, 0x04)
241		switch.flows = []
242
243		flow_1 = MagicMock()
244		flow_1.as_dict.return_value = {'flow_1': 'data'}
245
246		flow_list = [{"command": "add",
247		"flow": {'flow_1': 'data'}
248		}]
249		serializer = MagicMock()
250
251		mock_flow_factory.return_value = serializer
252		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
253		self.napp.check_switch_consistency(switch)
254		mock_install_flows.assert_called()
255
256	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...
257		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
258		def test_check_switch_consistency_delete(self, *args):
259		"""Test check_switch_consistency method.
260
261		This test checks the case when a flow is missing in switch and have the
262		DELETE command.
263		"""
264		(mock_flow_factory, mock_install_flows) = args
265		dpid = "00:00:00:00:00:00:00:01"
266		switch = get_switch_mock(dpid, 0x04)
267
268		flow_1 = MagicMock()
269		flow_1.as_dict.return_value = {'flow_1': 'data'}
270
271		flow_list = [{"command": "delete",
272		"flow": {'flow_1': 'data'}
273		}]
274		serializer = MagicMock()
275		serializer.from_dict.return_value = flow_1
276
277		switch.flows = [flow_1]
278
279		mock_flow_factory.return_value = serializer
280		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
281		self.napp.check_switch_consistency(switch)
282		mock_install_flows.assert_called()
283
284	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...
285		@patch('napps.kytos.flow_manager.main.FlowFactory.get_class')
286		def test_check_storehouse_consistency(self, *args):
287		"""Test check_storehouse_consistency method.
288
289		This test checks the case when a flow is missing in storehouse.
290		"""
291		(mock_flow_factory, mock_install_flows) = args
292		dpid = "00:00:00:00:00:00:00:01"
293		switch = get_switch_mock(dpid, 0x04)
294
295		flow_1 = MagicMock()
296		flow_1.as_dict.return_value = {'flow_1': 'data'}
297
298		switch.flows = [flow_1]
299
300		flow_list = [{"command": "add",
301		"flow": {'flow_2': 'data'}
302		}]
303		serializer = MagicMock()
304
305		mock_flow_factory.return_value = serializer
306		self.napp.stored_flows = {dpid: {"flow_list": flow_list}}
307		self.napp.check_storehouse_consistency(switch)
308		mock_install_flows.assert_called()
309

kytos / flow_manager

Pull Request — master (#96)

TestMain.test_send_napp_event() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like