Passed
Push — master ( cf72e6...2ad18d )
by Humberto
06:56 queued 03:57
created

TestController.setUp()   A

Complexity

Conditions 1

Size

Total Lines 11
Code Lines 8

Duplication

Lines 0
Ratio 0 %

Importance

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