tests.unit.test_core.test_controller   F
last analyzed

Complexity

Total Complexity 66

Size/Duplication

Total Lines 700
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 470
dl 0
loc 700
rs 3.12
c 0
b 0
f 0
wmc 66

57 Methods

Rating   Name   Duplication   Size   Complexity  
A TestController.test_configuration_endpoint() 0 6 1
A TestController.test_websocket_log_usage() 0 22 1
A TestController.setUp() 0 11 1
A TestController.test_stop() 0 9 1
A TestController.test_debug_on_defaults() 0 5 1
A TestController.test_reload_napp__error() 0 10 1
A TestController.test_add_new_switch() 0 10 1
A TestController.test_create_pidfile() 0 15 2
A TestController.test_app_event_handler() 0 10 1
A TestController.test_notify_listeners() 0 10 1
A TestController.test_stop_controller() 0 24 1
A TestController.test_get_switch_or_create__exists() 0 10 1
A TestController.test_rest_reload_napp() 0 8 1
A TestController._add_napp() 0 9 1
A TestController._test_debug_result() 0 10 3
A TestController.test_remove_switch() 0 9 1
A TestController.test_load_napp__file_not_found() 0 9 1
A TestController.test_remove_switch__error() 0 11 1
A TestController.test_msg_out_event_handler() 0 25 1
A TestController.test_remove_connection() 0 9 1
A TestController.test_get_connection_by_id() 0 9 1
A TestController.test_load_napp__module_not_found() 0 9 1
A TestController.test_pre_install_napps() 0 11 1
A TestController.test_debug_no_name() 0 8 1
A TestController.test_get_interface_by_id() 0 11 1
A TestController.test_reload_napp_module__module_not_found() 0 7 2
A TestController.test_get_switch_by_dpid() 0 9 1
A TestController.test_deprecation_warning() 0 9 3
A TestController.test_debug_off() 0 9 2
A TestController.test_reload_napp() 0 15 1
A TestController.test_restart() 0 14 1
A TestController.test_import_napp() 0 18 1
A TestController.test_rest_reload_all_napps() 0 9 1
A TestController.test_raw_event_handler() 0 10 1
A TestController.test_create_or_update_connection() 0 9 1
A TestController.test_load_napps() 0 12 1
A TestController.test_get_interface_by_id__not_switch() 0 11 1
A TestController.test_load_napp() 0 17 1
A TestController.test_status() 0 8 1
A TestController.test_get_switch_or_create__not_exists() 0 10 1
A TestController.test_load_napp__loaded() 0 8 1
A TestController.test_loggers() 0 7 2
A TestController.test_start() 0 11 1
A TestController.test_unload_napp_other_listener() 0 9 1
A TestController.test_debug_wrong_name() 0 4 1
A TestController.test_debug_empty_name() 0 8 1
A TestController.test_unload_napp_listener() 0 9 1
A TestController.test_start_controller() 0 31 1
A TestController.test_new_connection() 0 11 1
A TestController.test_get_interface_by_id__not_interface() 0 5 1
A TestController.test_msg_in_event_handler() 0 11 1
A TestController.test_debug_on() 0 5 1
A TestController.test_reload_napp_module() 0 12 1
A TestController.test_load_napp__error() 0 15 1
A TestController.test_metadata_endpoint() 0 8 1
A TestController.test_reload_napp_module__import_error() 0 11 2
A TestController.test_uptime() 0 11 1

How to fix   Complexity   

Complexity

Complex classes like tests.unit.test_core.test_controller often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
"""Test kytos.core.controller module."""
2
import asyncio
3
import json
4
import logging
5
import sys
6
import tempfile
7
import warnings
8
from copy import copy
9
from unittest import TestCase
10
from unittest.mock import MagicMock, Mock, call, patch
11
12
from kytos.core import Controller
13
from kytos.core.config import KytosConfig
14
from kytos.core.logs import LogManager
15
16
17
# pylint: disable=protected-access, too-many-public-methods
18
class TestController(TestCase):
19
    """Controller tests."""
20
21
    def setUp(self):
22
        """Instantiate a controller."""
23
24
        self.loop = asyncio.new_event_loop()
25
        asyncio.set_event_loop(self.loop)
26
27
        self.options = KytosConfig().options['daemon']
28
        self.napps_manager = Mock()
29
        self.controller = Controller(self.options, loop=self.loop)
30
        self.controller.napps_manager = self.napps_manager
31
        self.controller.log = Mock()
32
33
    def test_configuration_endpoint(self):
34
        """Should return the attribute options as json."""
35
        serializable_options = vars(self.options)
36
        expected = json.dumps(serializable_options)
37
        actual = self.controller.configuration_endpoint()
38
        self.assertEqual(expected, actual)
39
40
    @staticmethod
41
    @patch('kytos.core.controller.LogManager')
42
    @patch('kytos.core.logs.Path')
43
    def test_websocket_log_usage(path, log_manager):
44
        """Assert that the web socket log is used."""
45
        loop = asyncio.new_event_loop()
46
        asyncio.set_event_loop(None)
47
48
        # Save original state
49
        handlers_bak = copy(logging.root.handlers)
50
51
        # Minimum to instantiate Controller
52
        options = Mock(napps='')
53
        path.return_value.exists.return_value = False
54
        controller = Controller(options, loop=loop)
55
56
        # The test
57
        controller.enable_logs()
58
        log_manager.enable_websocket.assert_called_once()
59
60
        # Restore original state
61
        logging.root.handlers = handlers_bak
62
63
    @patch('kytos.core.api_server.APIServer.remove_napp_endpoints')
64
    def test_unload_napp_listener(self, _):
65
        """Call NApp shutdown listener on unload."""
66
        username, napp_name = 'test', 'napp'
67
        listener = self._add_napp(username, napp_name)
68
69
        listener.assert_not_called()
70
        self.controller.unload_napp(username, napp_name)
71
        listener.assert_called()
72
73
    @patch('kytos.core.api_server.APIServer.remove_napp_endpoints')
74
    def test_unload_napp_other_listener(self, _):
75
        """Should not call other NApps' shutdown listener on unload."""
76
        username, napp_name = 'test', 'napp1'
77
        self._add_napp(username, napp_name)
78
        other_listener = self._add_napp('test', 'napp2')
79
80
        self.controller.unload_napp(username, napp_name)
81
        other_listener.assert_not_called()
82
83
    def _add_napp(self, username, napp_name):
84
        """Add a mocked NApp to the controller."""
85
        napp_id = f'{username}/{napp_name}'
86
        event_name = f'kytos/core.shutdown.{napp_id}'
87
        listener = Mock()
88
        self.controller.events_listeners[event_name] = [listener]
89
        napp = Mock(_listeners={})
90
        self.controller.napps[(username, napp_name)] = napp
91
        return listener
92
93
    def test_deprecation_warning(self):
94
        """Deprecated method should suggest @rest decorator."""
95
        with warnings.catch_warnings(record=True) as wrngs:
96
            warnings.simplefilter("always")  # trigger all warnings
97
            self.controller.register_rest_endpoint('x', lambda x: x, ['GET'])
98
            self.assertEqual(1, len(wrngs))
99
            warning = wrngs[0]
100
            self.assertEqual(warning.category, DeprecationWarning)
101
            self.assertIn('@rest', str(warning.message))
102
103
    def test_loggers(self):
104
        """Test that all controller loggers are under kytos
105
        hierarchy logger.
106
        """
107
        loggers = self.controller.loggers()
108
        for logger in loggers:
109
            self.assertTrue(logger.name.startswith("kytos"))
110
111
    def test_debug_on(self):
112
        """Test the enable debug feature."""
113
        # Enable debug for kytos.core
114
        self.controller.toggle_debug("kytos.core")
115
        self._test_debug_result()
116
117
    def test_debug_on_defaults(self):
118
        """Test the enable debug feature. Test the default parameter"""
119
        # Enable debug for kytos.core
120
        self.controller.toggle_debug("kytos.core")
121
        self._test_debug_result()
122
123
    def _test_debug_result(self):
124
        """Verify if the loggers have level debug."""
125
        loggers = self.controller.loggers()
126
        for logger in loggers:
127
            # Check if all kytos.core loggers are in DEBUG mode.
128
            # All the rest must remain the same.
129
            if logger.name.startswith("kytos.core"):
130
                self.assertTrue(logger.getEffectiveLevel(), logging.DEBUG)
131
            else:
132
                self.assertTrue(logger.getEffectiveLevel(), logging.CRITICAL)
133
134
    def test_debug_off(self):
135
        """Test the disable debug feature"""
136
        # Fist we enable the debug
137
        self.controller.toggle_debug("kytos.core")
138
        # ... then we disable the debug for the test
139
        self.controller.toggle_debug("kytos.core")
140
        loggers = self.controller.loggers()
141
        for logger in loggers:
142
            self.assertTrue(logger.getEffectiveLevel(), logging.CRITICAL)
143
144
    @patch.object(LogManager, 'load_config_file')
145
    def test_debug_no_name(self, mock_load_config_file):
146
        """Test the enable debug logger with default levels."""
147
        # Mock the LogManager that loads the default Loggers
148
        self.controller.toggle_debug()
149
        self._test_debug_result()
150
151
        mock_load_config_file.assert_called_once()
152
153
    @patch.object(LogManager, 'load_config_file')
154
    def test_debug_empty_name(self, mock_load_config_file):
155
        """Test the enable debug logger with default levels."""
156
        # Mock the LogManager that loads the default Loggers
157
        self.controller.toggle_debug('')
158
        self._test_debug_result()
159
160
        mock_load_config_file.assert_called_once()
161
162
    def test_debug_wrong_name(self):
163
        """Test the enable debug logger with wrong name."""
164
        self.assertRaises(ValueError,
165
                          self.controller.toggle_debug, name="foobar")
166
167
    @patch('kytos.core.controller.Controller.start_controller')
168
    @patch('kytos.core.controller.Controller.create_pidfile')
169
    @patch('kytos.core.controller.Controller.enable_logs')
170
    def test_start(self, *args):
171
        """Test activate method."""
172
        (mock_enable_logs, mock_create_pidfile, mock_start_controller) = args
173
        self.controller.start()
174
175
        mock_enable_logs.assert_called()
176
        mock_create_pidfile.assert_called()
177
        mock_start_controller.assert_called()
178
179
    @patch('os.getpid')
180
    @patch('kytos.core.controller.atexit')
181
    def test_create_pidfile(self, *args):
182
        """Test activate method."""
183
        (_, mock_getpid) = args
184
        mock_getpid.return_value = 2
185
        with tempfile.NamedTemporaryFile() as tmp_file:
186
            tmp_file.write(b'1')
187
            tmp_file.seek(0)
188
            self.controller.options.pidfile = tmp_file.name
189
190
            self.controller.create_pidfile()
191
192
            pid = tmp_file.read()
193
            self.assertEqual(pid, b'2')
194
195
    @staticmethod
196
    @patch('kytos.core.controller.KytosServer')
197
    @patch('kytos.core.controller.Controller.app_event_handler')
198
    @patch('kytos.core.controller.Controller.msg_out_event_handler')
199
    @patch('kytos.core.controller.Controller.msg_in_event_handler')
200
    @patch('kytos.core.controller.Controller.raw_event_handler')
201
    @patch('kytos.core.controller.Controller.load_napps')
202
    @patch('kytos.core.controller.Controller.pre_install_napps')
203
    def test_start_controller(*args):
204
        """Test activate method."""
205
        (mock_pre_install_napps, mock_load_napps, mock_raw_event_handler,
206
         mock_msg_in_event_handler, mock_msg_out_event_handler,
207
         mock_app_event_handler, _) = args
208
209
        napp = MagicMock()
210
        loop = MagicMock()
211
        options = KytosConfig().options['daemon']
212
        options.napps_pre_installed = [napp]
213
        controller = Controller(options, loop=loop)
214
        controller.log = Mock()
215
216
        controller.start_controller()
217
218
        controller.server.serve_forever.assert_called()
219
        calls = [call(mock_raw_event_handler.return_value),
220
                 call(mock_msg_in_event_handler.return_value),
221
                 call(mock_msg_out_event_handler.return_value),
222
                 call(mock_app_event_handler.return_value)]
223
        loop.create_task.assert_has_calls(calls)
224
        mock_pre_install_napps.assert_called_with([napp])
225
        mock_load_napps.assert_called()
226
227
    @patch('kytos.core.controller.Controller.__init__')
228
    @patch('kytos.core.controller.Controller.start')
229
    @patch('kytos.core.controller.Controller.stop')
230
    def test_restart(self, *args):
231
        """Test restart method."""
232
        (mock_stop, mock_start, mock_init) = args
233
        self.controller.started_at = 1
234
235
        graceful = True
236
        self.controller.restart(graceful)
237
238
        mock_stop.assert_called_with(graceful)
239
        mock_init.assert_called_with(self.controller.options)
240
        mock_start.assert_called_with(restart=True)
241
242
    @patch('kytos.core.controller.Controller.stop_controller')
243
    def test_stop(self, mock_stop_controller):
244
        """Test stop method."""
245
        self.controller.started_at = 1
246
247
        graceful = True
248
        self.controller.stop(graceful)
249
250
        mock_stop_controller.assert_called_with(graceful)
251
252
    @patch('kytos.core.controller.Controller.unload_napps')
253
    @patch('kytos.core.controller.KytosBuffers')
254
    def test_stop_controller(self, *args):
255
        """Test stop_controller method."""
256
        (_, mock_unload_napps) = args
257
        server = MagicMock()
258
        buffers = MagicMock()
259
        api_server = MagicMock()
260
        napp_dir_listener = MagicMock()
261
        pool = MagicMock()
262
        self.controller.server = server
263
        self.controller.buffers = buffers
264
        self.controller.api_server = api_server
265
        self.controller.napp_dir_listener = napp_dir_listener
266
        self.controller._pool = pool
267
268
        self.controller.stop_controller()
269
270
        buffers.send_stop_signal.assert_called()
271
        api_server.stop_api_server.assert_called()
272
        napp_dir_listener.stop.assert_called()
273
        pool.shutdown.assert_called()
274
        mock_unload_napps.assert_called()
275
        server.shutdown.assert_called()
276
277
    def test_status(self):
278
        """Test status method."""
279
        status_1 = self.controller.status()
280
        self.controller.started_at = 1
281
        status_2 = self.controller.status()
282
283
        self.assertEqual(status_1, 'Stopped')
284
        self.assertEqual(status_2, 'Running since 1')
285
286
    @patch('kytos.core.controller.now')
287
    def test_uptime(self, mock_now):
288
        """Test uptime method."""
289
        mock_now.return_value = 11
290
291
        uptime_1 = self.controller.uptime()
292
        self.controller.started_at = 1
293
        uptime_2 = self.controller.uptime()
294
295
        self.assertEqual(uptime_1, 0)
296
        self.assertEqual(uptime_2, 10)
297
298
    def test_metadata_endpoint(self):
299
        """Test metadata_endpoint method."""
300
        metadata = self.controller.metadata_endpoint()
301
        json_metadata = json.loads(metadata)
302
303
        expected_keys = ['__version__', '__author__', '__license__', '__url__',
304
                         '__description__']
305
        self.assertEqual(list(json_metadata.keys()), expected_keys)
306
307
    def test_notify_listeners(self):
308
        """Test notify_listeners method."""
309
        method = MagicMock()
310
        self.controller.events_listeners = {'kytos/any': [method]}
311
312
        event = MagicMock()
313
        event.name = 'kytos/any'
314
        self.controller.notify_listeners(event)
315
316
        method.assert_called_with(event)
317
318
    def test_get_interface_by_id__not_interface(self):
319
        """Test get_interface_by_id method when interface does not exist."""
320
        resp_interface = self.controller.get_interface_by_id(None)
321
322
        self.assertIsNone(resp_interface)
323
324
    def test_get_interface_by_id__not_switch(self):
325
        """Test get_interface_by_id method when switch does not exist."""
326
        interface = MagicMock()
327
        switch = MagicMock()
328
        switch.interfaces = {123: interface}
329
        self.controller.switches = {'00:00:00:00:00:00:00:02': switch}
330
331
        interface_id = '00:00:00:00:00:00:00:01:123'
332
        resp_interface = self.controller.get_interface_by_id(interface_id)
333
334
        self.assertIsNone(resp_interface)
335
336
    def test_get_interface_by_id(self):
337
        """Test get_interface_by_id method."""
338
        interface = MagicMock()
339
        switch = MagicMock()
340
        switch.interfaces = {123: interface}
341
        self.controller.switches = {'00:00:00:00:00:00:00:01': switch}
342
343
        interface_id = '00:00:00:00:00:00:00:01:123'
344
        resp_interface = self.controller.get_interface_by_id(interface_id)
345
346
        self.assertEqual(resp_interface, interface)
347
348
    def test_get_switch_by_dpid(self):
349
        """Test get_switch_by_dpid method."""
350
        dpid = '00:00:00:00:00:00:00:01'
351
        switch = MagicMock(dpid=dpid)
352
        self.controller.switches = {dpid: switch}
353
354
        resp_switch = self.controller.get_switch_by_dpid(dpid)
355
356
        self.assertEqual(resp_switch, switch)
357
358
    def test_get_switch_or_create__exists(self):
359
        """Test status_api method when switch exists."""
360
        dpid = '00:00:00:00:00:00:00:01'
361
        switch = MagicMock(dpid=dpid)
362
        self.controller.switches = {dpid: switch}
363
364
        connection = MagicMock()
365
        resp_switch = self.controller.get_switch_or_create(dpid, connection)
366
367
        self.assertEqual(resp_switch, switch)
368
369
    def test_get_switch_or_create__not_exists(self):
370
        """Test status_api method when switch does not exist."""
371
        self.controller.switches = {}
372
373
        dpid = '00:00:00:00:00:00:00:01'
374
        connection = MagicMock()
375
        switch = self.controller.get_switch_or_create(dpid, connection)
376
377
        expected_switches = {'00:00:00:00:00:00:00:01': switch}
378
        self.assertEqual(self.controller.switches, expected_switches)
379
380
    def test_create_or_update_connection(self):
381
        """Test create_or_update_connection method."""
382
        self.controller.connections = {}
383
384
        connection = MagicMock()
385
        connection.id = '123'
386
        self.controller.create_or_update_connection(connection)
387
388
        self.assertEqual(self.controller.connections, {'123': connection})
389
390
    def test_get_connection_by_id(self):
391
        """Test get_connection_by_id method."""
392
        connection = MagicMock()
393
        connection.id = '123'
394
        self.controller.connections = {connection.id: connection}
395
396
        resp_connection = self.controller.get_connection_by_id('123')
397
398
        self.assertEqual(resp_connection, connection)
399
400
    def test_remove_connection(self):
401
        """Test remove_connection method."""
402
        connection = MagicMock()
403
        connection.id = '123'
404
        self.controller.connections = {connection.id: connection}
405
406
        self.controller.remove_connection(connection)
407
408
        self.assertEqual(self.controller.connections, {})
409
410
    def test_remove_switch(self):
411
        """Test remove_switch method."""
412
        switch = MagicMock()
413
        switch.dpid = '00:00:00:00:00:00:00:01'
414
        self.controller.switches = {switch.dpid: switch}
415
416
        self.controller.remove_switch(switch)
417
418
        self.assertEqual(self.controller.switches, {})
419
420
    def test_remove_switch__error(self):
421
        """Test remove_switch method to error case."""
422
        switch_1 = MagicMock()
423
        switch_2 = MagicMock()
424
        switch_1.dpid = '00:00:00:00:00:00:00:01'
425
        switch_2.dpid = '00:00:00:00:00:00:00:02'
426
        self.controller.switches = {switch_1.dpid: switch_1}
427
428
        self.controller.remove_switch(switch_2)
429
430
        self.assertEqual(self.controller.switches, {switch_1.dpid: switch_1})
431
432
    def test_new_connection(self):
433
        """Test new_connection method."""
434
        self.controller.connections = {}
435
436
        connection = MagicMock()
437
        connection.id = '123'
438
        event = MagicMock()
439
        event.source = connection
440
        self.controller.new_connection(event)
441
442
        self.assertEqual(self.controller.connections, {'123': connection})
443
444
    def test_add_new_switch(self):
445
        """Test add_new_switch method."""
446
        self.controller.switches = {}
447
448
        switch = MagicMock()
449
        switch.dpid = '00:00:00:00:00:00:00:01'
450
        self.controller.add_new_switch(switch)
451
452
        expected_switches = {'00:00:00:00:00:00:00:01': switch}
453
        self.assertEqual(self.controller.switches, expected_switches)
454
455
    @patch('kytos.core.controller.module_from_spec')
456
    @patch('kytos.core.controller.spec_from_file_location')
457
    def test_import_napp(self, *args):
458
        """Test _import_napp method."""
459
        (mock_spec_from_file, mock_module_from_spec) = args
460
        napp_spec = MagicMock()
461
        napp_spec.name = 'spec_name'
462
        mock_spec_from_file.return_value = napp_spec
463
        napp_module = MagicMock()
464
        mock_module_from_spec.return_value = napp_module
465
466
        self.controller.options.napps = 'napps'
467
        self.controller._import_napp('kytos', 'napp')
468
469
        self.assertEqual(sys.modules[napp_spec.name], napp_module)
470
        mock_spec_from_file.assert_called_with('napps.kytos.napp.main',
471
                                               'napps/kytos/napp/main.py')
472
        napp_spec.loader.exec_module.assert_called_with(napp_module)
473
474
    def test_load_napp__loaded(self):
475
        """Test load_napp method when napp is already loaded."""
476
        napp = MagicMock()
477
        self.controller.napps = {('kytos', 'napp'): napp}
478
479
        self.controller.load_napp('kytos', 'napp')
480
481
        self.assertEqual(self.controller.napps, {('kytos', 'napp'): napp})
482
483
    @patch('kytos.core.controller.Controller._import_napp')
484
    def test_load_napp__module_not_found(self, mock_import_napp):
485
        """Test load_napp method when module is not found."""
486
        mock_import_napp.side_effect = ModuleNotFoundError
487
        self.controller.napps = {}
488
489
        self.controller.load_napp('kytos', 'napp')
490
491
        self.assertEqual(self.controller.napps, {})
492
493
    @patch('kytos.core.controller.Controller._import_napp')
494
    def test_load_napp__file_not_found(self, mock_import_napp):
495
        """Test load_napp method when file is not found."""
496
        mock_import_napp.side_effect = FileNotFoundError
0 ignored issues
show
Comprehensibility Best Practice introduced by Gleyberson Andrade
The variable FileNotFoundError does not seem to be defined.
Loading history...
497
        self.controller.napps = {}
498
499
        self.controller.load_napp('kytos', 'napp')
500
501
        self.assertEqual(self.controller.napps, {})
502
503
    @patch('kytos.core.api_server.APIServer.register_napp_endpoints')
504
    @patch('kytos.core.controller.Controller._import_napp')
505
    def test_load_napp__error(self, *args):
506
        """Test load_napp method when an error is raised on napp module
507
           attribution."""
508
        (mock_import_napp, _) = args
509
        self.controller.napps = {}
510
511
        module = MagicMock()
512
        module.Main.side_effect = Exception
513
        mock_import_napp.return_value = module
514
515
        self.controller.load_napp('kytos', 'napp')
516
517
        self.assertEqual(self.controller.napps, {})
518
519
    @patch('kytos.core.api_server.APIServer.register_napp_endpoints')
520
    @patch('kytos.core.controller.Controller._import_napp')
521
    def test_load_napp(self, *args):
522
        """Test load_napp method."""
523
        (mock_import_napp, mock_register) = args
524
        self.controller.napps = {}
525
526
        napp = MagicMock()
527
        module = MagicMock()
528
        module.Main.return_value = napp
529
        mock_import_napp.return_value = module
530
531
        self.controller.load_napp('kytos', 'napp')
532
533
        self.assertEqual(self.controller.napps, {('kytos', 'napp'): napp})
534
        napp.start.assert_called()
535
        mock_register.assert_called_with(napp)
536
537
    def test_pre_install_napps(self):
538
        """Test pre_install_napps method."""
539
        napp_1 = MagicMock()
540
        napp_2 = MagicMock()
541
        installed_napps = [napp_1]
542
        napps = [str(napp_1), str(napp_2)]
543
        self.napps_manager.get_installed_napps.return_value = installed_napps
544
545
        self.controller.pre_install_napps(napps)
546
547
        self.napps_manager.install.assert_called_with(str(napp_2), enable=True)
548
549
    @patch('kytos.core.controller.Controller.load_napp')
550
    def test_load_napps(self, mock_load):
551
        """Test load_napps method."""
552
        napp = MagicMock()
553
        napp.username = 'kytos'
554
        napp.name = 'name'
555
        enabled_napps = [napp]
556
        self.napps_manager.get_enabled_napps.return_value = enabled_napps
557
558
        self.controller.load_napps()
559
560
        mock_load.assert_called_with('kytos', 'name')
561
562
    @patch('kytos.core.controller.import_module')
563
    def test_reload_napp_module__module_not_found(self, mock_import_module):
564
        """Test reload_napp_module method when module is not found."""
565
        mock_import_module.side_effect = ModuleNotFoundError
566
567
        with self.assertRaises(ModuleNotFoundError):
568
            self.controller.reload_napp_module('kytos', 'napp', 'napp_file')
569
570
    @patch('kytos.core.controller.reload_module')
571
    @patch('kytos.core.controller.import_module')
572
    def test_reload_napp_module__import_error(self, *args):
573
        """Test reload_napp_module method when an import error occurs."""
574
        (mock_import_module, mock_reload_module) = args
575
        napp_module = MagicMock()
576
        mock_import_module.return_value = napp_module
577
        mock_reload_module.side_effect = ImportError
578
579
        with self.assertRaises(ImportError):
580
            self.controller.reload_napp_module('kytos', 'napp', 'napp_file')
581
582
    @patch('kytos.core.controller.reload_module')
583
    @patch('kytos.core.controller.import_module')
584
    def test_reload_napp_module(self, *args):
585
        """Test reload_napp_module method."""
586
        (mock_import_module, mock_reload_module) = args
587
        napp_module = MagicMock()
588
        mock_import_module.return_value = napp_module
589
590
        self.controller.reload_napp_module('kytos', 'napp', 'napp_file')
591
592
        mock_import_module.assert_called_with('napps.kytos.napp.napp_file')
593
        mock_reload_module.assert_called_with(napp_module)
594
595
    @patch('kytos.core.controller.Controller.load_napp')
596
    @patch('kytos.core.controller.Controller.unload_napp')
597
    @patch('kytos.core.controller.Controller.reload_napp_module')
598
    def test_reload_napp(self, *args):
599
        """Test reload_napp method."""
600
        (mock_reload_napp_module, mock_unload, mock_load) = args
601
602
        code = self.controller.reload_napp('kytos', 'napp')
603
604
        mock_unload.assert_called_with('kytos', 'napp')
605
        calls = [call('kytos', 'napp', 'settings'),
606
                 call('kytos', 'napp', 'main')]
607
        mock_reload_napp_module.assert_has_calls(calls)
608
        mock_load.assert_called_with('kytos', 'napp')
609
        self.assertEqual(code, 200)
610
611
    @patch('kytos.core.controller.Controller.unload_napp')
612
    @patch('kytos.core.controller.Controller.reload_napp_module')
613
    def test_reload_napp__error(self, *args):
614
        """Test reload_napp method to error case."""
615
        (mock_reload_napp_module, _) = args
616
        mock_reload_napp_module.side_effect = ModuleNotFoundError
617
618
        code = self.controller.reload_napp('kytos', 'napp')
619
620
        self.assertEqual(code, 400)
621
622
    @patch('kytos.core.controller.Controller.reload_napp', return_value=200)
623
    def test_rest_reload_napp(self, mock_reload_napp):
624
        """Test rest_reload_napp method."""
625
        resp, code = self.controller.rest_reload_napp('kytos', 'napp')
626
627
        mock_reload_napp.assert_called_with('kytos', 'napp')
628
        self.assertEqual(resp, 'reloaded')
629
        self.assertEqual(code, 200)
630
631
    @patch('kytos.core.controller.Controller.reload_napp')
632
    def test_rest_reload_all_napps(self, mock_reload_napp):
633
        """Test rest_reload_all_napps method."""
634
        self.controller.napps = [('kytos', 'napp')]
635
        resp, code = self.controller.rest_reload_all_napps()
636
637
        mock_reload_napp.assert_called_with('kytos', 'napp')
638
        self.assertEqual(resp, 'reloaded')
639
        self.assertEqual(code, 200)
640
641
    @patch('kytos.core.controller.Controller.notify_listeners')
642
    def test_raw_event_handler(self, mock_notify_listeners):
643
        """Test raw_event_handler async method by handling a shutdown event."""
644
        event = MagicMock()
645
        event.name = 'kytos/core.shutdown'
646
        self.controller.buffers.raw._queue.sync_q.put(event)
647
648
        self.loop.run_until_complete(self.controller.raw_event_handler())
649
650
        mock_notify_listeners.assert_called_with(event)
651
652
    @patch('kytos.core.controller.Controller.notify_listeners')
653
    def test_msg_in_event_handler(self, mock_notify_listeners):
654
        """Test msg_in_event_handler async method by handling a shutdown
655
           event."""
656
        event = MagicMock()
657
        event.name = 'kytos/core.shutdown'
658
        self.controller.buffers.msg_in._queue.sync_q.put(event)
659
660
        self.loop.run_until_complete(self.controller.msg_in_event_handler())
661
662
        mock_notify_listeners.assert_called_with(event)
663
664
    @patch('kytos.core.controller.Controller.notify_listeners')
665
    def test_msg_out_event_handler(self, mock_notify_listeners):
666
        """Test msg_out_event_handler async method by handling a common and a
667
           shutdown event."""
668
        dst = MagicMock()
669
        dst.state = 0
670
        packet = MagicMock()
671
        msg = MagicMock()
672
        msg.pack.return_value = packet
673
674
        event_1 = MagicMock()
675
        event_1.name = 'kytos/core.any'
676
        event_1.destination = dst
677
        event_1.content = {"message": msg}
678
679
        event_2 = MagicMock()
680
        event_2.name = 'kytos/core.shutdown'
681
682
        self.controller.buffers.msg_out._queue.sync_q.put(event_1)
683
        self.controller.buffers.msg_out._queue.sync_q.put(event_2)
684
685
        self.loop.run_until_complete(self.controller.msg_out_event_handler())
686
687
        dst.send.assert_called_with(packet)
688
        mock_notify_listeners.assert_called_with(event_1)
689
690
    @patch('kytos.core.controller.Controller.notify_listeners')
691
    def test_app_event_handler(self, mock_notify_listeners):
692
        """Test app_event_handler async method by handling a shutdown event."""
693
        event = MagicMock()
694
        event.name = 'kytos/core.shutdown'
695
        self.controller.buffers.app._queue.sync_q.put(event)
696
697
        self.loop.run_until_complete(self.controller.app_event_handler())
698
699
        mock_notify_listeners.assert_called_with(event)
700