Passed
Push — master ( 3b7d73...e388a5 )
by Humberto
02:56
created

TestController.test_debug_no_name()   A

Complexity

Conditions 1

Size

Total Lines 8
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 2
dl 0
loc 8
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 warnings
6
from copy import copy
7
from unittest import TestCase
8
from unittest.mock import Mock, patch
9
10
from kytos.core import Controller
11
from kytos.core.config import KytosConfig
12
from kytos.core.logs import LogManager
13
14
15
class TestController(TestCase):
16
    """Controller tests."""
17
18
    def setUp(self):
19
        """Instantiate a controller."""
20
21
        self.loop = asyncio.new_event_loop()
22
        asyncio.set_event_loop(None)
23
24
        self.options = KytosConfig().options['daemon']
25
        self.controller = Controller(self.options, loop=self.loop)
26
        self.controller.log = Mock()
27
28
    def test_configuration_endpoint(self):
29
        """Should return the attribute options as json."""
30
        serializable_options = vars(self.options)
31
        expected = json.dumps(serializable_options)
32
        actual = self.controller.configuration_endpoint()
33
        self.assertEqual(expected, actual)
34
35
    @staticmethod
36
    @patch('kytos.core.controller.LogManager')
37
    @patch('kytos.core.logs.Path')
38
    def test_websocket_log_usage(path, log_manager):
39
        """Assert that the web socket log is used."""
40
        loop = asyncio.new_event_loop()
41
        asyncio.set_event_loop(None)
42
43
        # Save original state
44
        handlers_bak = copy(logging.root.handlers)
45
46
        # Minimum to instantiate Controller
47
        options = Mock(napps='')
48
        path.return_value.exists.return_value = False
49
        controller = Controller(options, loop=loop)
50
51
        # The test
52
        controller.enable_logs()
53
        log_manager.enable_websocket.assert_called_once()
54
55
        # Restore original state
56
        logging.root.handlers = handlers_bak
57
58
    def test_unload_napp_listener(self):
59
        """Call NApp shutdown listener on unload."""
60
        username, napp_name = 'test', 'napp'
61
        listener = self._add_napp(username, napp_name)
62
63
        listener.assert_not_called()
64
        self.controller.unload_napp(username, napp_name)
65
        listener.assert_called()
66
67
    def test_unload_napp_other_listener(self):
68
        """Should not call other NApps' shutdown listener on unload."""
69
        username, napp_name = 'test', 'napp1'
70
        self._add_napp(username, napp_name)
71
        other_listener = self._add_napp('test', 'napp2')
72
73
        self.controller.unload_napp(username, napp_name)
74
        other_listener.assert_not_called()
75
76
    def _add_napp(self, username, napp_name):
77
        """Add a mocked NApp to the controller."""
78
        napp_id = f'{username}/{napp_name}'
79
        event_name = f'kytos/core.shutdown.{napp_id}'
80
        listener = Mock()
81
        self.controller.events_listeners[event_name] = [listener]
82
        napp = Mock(_listeners={})
83
        self.controller.napps[(username, napp_name)] = napp
84
        return listener
85
86
    def test_deprecation_warning(self):
87
        """Deprecated method should suggest @rest decorator."""
88
        with warnings.catch_warnings(record=True) as wrngs:
89
            warnings.simplefilter("always")  # trigger all warnings
90
            self.controller.register_rest_endpoint('x', lambda x: x, ['GET'])
91
            self.assertEqual(1, len(wrngs))
92
            warning = wrngs[0]
93
            self.assertEqual(warning.category, DeprecationWarning)
94
            self.assertIn('@rest', str(warning.message))
95
96
    def test_loggers(self):
97
        """Test that all controller loggers are under kytos
98
        hierarchy logger.
99
        """
100
        loggers = self.controller.loggers()
101
        for logger in loggers:
102
            self.assertTrue(logger.name.startswith("kytos"))
103
104
    def test_debug_on(self):
105
        """Test the enable debug feature."""
106
        # Enable debug for kytos.core
107
        self.controller.toggle_debug("kytos.core")
108
        self._test_debug_result()
109
110
    def test_debug_on_defaults(self):
111
        """Test the enable debug feature. Test the default parameter"""
112
        # Enable debug for kytos.core
113
        self.controller.toggle_debug("kytos.core")
114
        self._test_debug_result()
115
116
    def _test_debug_result(self):
117
        """Verify if the loggers have level debug."""
118
        loggers = self.controller.loggers()
119
        for logger in loggers:
120
            # Check if all kytos.core loggers are in DEBUG mode.
121
            # All the rest must remain the same.
122
            if logger.name.startswith("kytos.core"):
123
                self.assertTrue(logger.getEffectiveLevel(), logging.DEBUG)
124
            else:
125
                self.assertTrue(logger.getEffectiveLevel(), logging.CRITICAL)
126
127
    def test_debug_off(self):
128
        """Test the disable debug feature"""
129
        # Fist we enable the debug
130
        self.controller.toggle_debug("kytos.core")
131
        # ... then we disable the debug for the test
132
        self.controller.toggle_debug("kytos.core")
133
        loggers = self.controller.loggers()
134
        for logger in loggers:
135
            self.assertTrue(logger.getEffectiveLevel(), logging.CRITICAL)
136
137
    @patch.object(LogManager, 'load_config_file')
138
    def test_debug_no_name(self, mock_load_config_file):
139
        """Test the enable debug logger with default levels."""
140
        # Mock the LogManager that loads the default Loggers
141
        self.controller.toggle_debug()
142
        self._test_debug_result()
143
144
        mock_load_config_file.assert_called_once()
145
146
    @patch.object(LogManager, 'load_config_file')
147
    def test_debug_empty_name(self, mock_load_config_file):
148
        """Test the enable debug logger with default levels."""
149
        # Mock the LogManager that loads the default Loggers
150
        self.controller.toggle_debug('')
151
        self._test_debug_result()
152
153
        mock_load_config_file.assert_called_once()
154
155
    def test_debug_wrong_name(self):
156
        """Test the enable debug logger with wrong name."""
157
        self.assertRaises(ValueError,
158
                          self.controller.toggle_debug, name="foobar")
159