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

TestAPIServer.test_list_installed_napps()   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
"""APIServer tests."""
2
import json
3
import unittest
4
import warnings
5
# Disable not-grouped imports that conflicts with isort
6
from unittest.mock import (MagicMock, Mock, patch,  # pylint: disable=C0412
7
                           sentinel)
8
from urllib.error import HTTPError
9
10
from kytos.core.api_server import APIServer
11
from kytos.core.napps import rest
12
13
KYTOS_CORE_API = "http://127.0.0.1:8181/api/kytos/"
14
API_URI = KYTOS_CORE_API+"core"
15
16
17
# pylint: disable=protected-access, too-many-public-methods
18
class TestAPIServer(unittest.TestCase):
19
    """Test the class APIServer."""
20
21
    def setUp(self):
22
        """Instantiate a APIServer."""
23
        self.api_server = APIServer('CustomName', False)
24
        self.napps_manager = MagicMock()
25
        self.api_server.server = MagicMock()
26
        self.api_server.napps_manager = self.napps_manager
27
        self.api_server.napps_dir = 'napps_dir'
28
        self.api_server.flask_dir = 'flask_dir'
29
30
    def test_deprecation_warning(self):
31
        """Deprecated method should suggest @rest decorator."""
32
        with warnings.catch_warnings(record=True) as wrngs:
33
            warnings.simplefilter("always")  # trigger all warnings
34
            self.api_server.register_rest_endpoint(
35
                'rule', lambda x: x, ['POST'])
36
            self.assertEqual(1, len(wrngs))
37
            warning = wrngs[0]
38
            self.assertEqual(warning.category, DeprecationWarning)
39
            self.assertIn('@rest', str(warning.message))
40
41
    def test_run(self):
42
        """Test run method."""
43
        self.api_server.run()
44
45
        self.api_server.server.run.assert_called_with(self.api_server.app,
46
                                                      self.api_server.listen,
47
                                                      self.api_server.port)
48
49
    @patch('sys.exit')
50
    def test_run_error(self, mock_exit):
51
        """Test run method to error case."""
52
        self.api_server.server.run.side_effect = OSError
53
        self.api_server.run()
54
55
        mock_exit.assert_called()
56
57
    @patch('kytos.core.api_server.request')
58
    def test_shutdown_api(self, mock_request):
59
        """Test shutdown_api method."""
60
        mock_request.host = 'localhost:8181'
61
62
        self.api_server.shutdown_api()
63
64
        self.api_server.server.stop.assert_called()
65
66
    @patch('kytos.core.api_server.request')
67
    def test_shutdown_api__error(self, mock_request):
68
        """Test shutdown_api method to error case."""
69
        mock_request.host = 'any:port'
70
71
        self.api_server.shutdown_api()
72
73
        self.api_server.server.stop.assert_not_called()
74
75
    def test_status_api(self):
76
        """Test status_api method."""
77
        status = self.api_server.status_api()
78
        self.assertEqual(status, ('{"response": "running"}', 200))
79
80
    @patch('kytos.core.api_server.urlopen')
81
    def test_stop_api_server(self, mock_urlopen):
82
        """Test stop_api_server method."""
83
        self.api_server.stop_api_server()
84
85
        url = "%s/shutdown" % API_URI
86
        mock_urlopen.assert_called_with(url)
87
88
    @patch('kytos.core.api_server.send_file')
89
    @patch('os.path.exists', return_value=True)
90
    def test_static_web_ui__success(self, *args):
91
        """Test static_web_ui method to success case."""
92
        (_, mock_send_file) = args
93
        self.api_server.static_web_ui('kytos', 'napp', 'filename')
94
95
        mock_send_file.assert_called_with('napps_dir/kytos/napp/ui/filename')
96
97
    @patch('os.path.exists', return_value=False)
98
    def test_static_web_ui__error(self, _):
99
        """Test static_web_ui method to error case."""
100
        resp, code = self.api_server.static_web_ui('kytos', 'napp', 'filename')
101
102
        self.assertEqual(resp, '')
103
        self.assertEqual(code, 404)
104
105
    @patch('kytos.core.api_server.glob')
106
    def test_get_ui_components(self, mock_glob):
107
        """Test get_ui_components method."""
108
        with self.api_server.app.app_context():
109
            mock_glob.return_value = ['napps_dir/*/*/ui/*/*.kytos']
110
            response = self.api_server.get_ui_components('all')
111
112
            expected_json = [{'name': '*-*-*-*', 'url': 'ui/*/*/*/*.kytos'}]
113
            self.assertEqual(response.json, expected_json)
114
            self.assertEqual(response.status_code, 200)
115
116
    @patch('kytos.core.api_server.send_file')
117
    def test_web_ui(self, mock_send_file):
118
        """Test web_ui method."""
119
        self.api_server.web_ui()
120
121
        mock_send_file.assert_called_with('flask_dir/index.html')
122
123
    @patch('kytos.core.api_server.urlretrieve')
124
    @patch('kytos.core.api_server.urlopen')
125
    @patch('zipfile.ZipFile')
126
    @patch('os.path.exists')
127
    @patch('os.mkdir')
128
    @patch('shutil.move')
129
    def test_update_web_ui(self, *args):
130
        """Test update_web_ui method."""
131
        (_, _, mock_exists, mock_zipfile, mock_urlopen,
132
         mock_urlretrieve) = args
133
        zipfile = MagicMock()
134
        zipfile.testzip.return_value = None
135
        mock_zipfile.return_value = zipfile
136
137
        data = json.dumps({'tag_name': 1.0})
138
        url_response = MagicMock()
139
        url_response.readlines.return_value = [data]
140
        mock_urlopen.return_value = url_response
141
142
        mock_exists.side_effect = [False, True]
143
144
        response = self.api_server.update_web_ui()
145
146
        url = 'https://github.com/kytos/ui/releases/download/1.0/latest.zip'
147
        mock_urlretrieve.assert_called_with(url)
148
        self.assertEqual(response, 'updated the web ui')
149
150
    @patch('kytos.core.api_server.urlretrieve')
151
    @patch('kytos.core.api_server.urlopen')
152
    @patch('os.path.exists')
153
    def test_update_web_ui__http_error(self, *args):
154
        """Test update_web_ui method to http error case."""
155
        (mock_exists, mock_urlopen, mock_urlretrieve) = args
156
157
        data = json.dumps({'tag_name': 1.0})
158
        url_response = MagicMock()
159
        url_response.readlines.return_value = [data]
160
        mock_urlopen.return_value = url_response
161
        mock_urlretrieve.side_effect = HTTPError('url', 123, 'msg', 'hdrs',
162
                                                 MagicMock())
163
164
        mock_exists.return_value = False
165
166
        response = self.api_server.update_web_ui()
167
168
        expected_response = 'Uri not found https://github.com/kytos/ui/' + \
169
                            'releases/download/1.0/latest.zip.'
170
        self.assertEqual(response, expected_response)
171
172
    @patch('kytos.core.api_server.urlretrieve')
173
    @patch('kytos.core.api_server.urlopen')
174
    @patch('zipfile.ZipFile')
175
    @patch('os.path.exists')
176
    def test_update_web_ui__zip_error(self, *args):
177
        """Test update_web_ui method to error case in zip file."""
178
        (mock_exists, mock_zipfile, mock_urlopen, _) = args
179
        zipfile = MagicMock()
180
        zipfile.testzip.return_value = 'any'
181
        mock_zipfile.return_value = zipfile
182
183
        data = json.dumps({'tag_name': 1.0})
184
        url_response = MagicMock()
185
        url_response.readlines.return_value = [data]
186
        mock_urlopen.return_value = url_response
187
188
        mock_exists.return_value = False
189
190
        response = self.api_server.update_web_ui()
191
192
        expected_response = 'Zip file from https://github.com/kytos/ui/' + \
193
                            'releases/download/1.0/latest.zip is corrupted.'
194
        self.assertEqual(response, expected_response)
195
196
    def test_enable_napp__error_not_installed(self):
197
        """Test _enable_napp method error case when napp is not installed."""
198
        self.napps_manager.is_installed.return_value = False
199
200
        resp, code = self.api_server._enable_napp('kytos', 'napp')
201
202
        self.assertEqual(resp, '{"response": "not installed"}')
203
        self.assertEqual(code, 400)
204
205
    def test_enable_napp__error_not_enabling(self):
206
        """Test _enable_napp method error case when napp is not enabling."""
207
        self.napps_manager.is_installed.return_value = True
208
        self.napps_manager.is_enabled.side_effect = [False, False]
209
210
        resp, code = self.api_server._enable_napp('kytos', 'napp')
211
212
        self.assertEqual(resp, '{"response": "error"}')
213
        self.assertEqual(code, 500)
214
215
    def test_enable_napp__success(self):
216
        """Test _enable_napp method success case."""
217
        self.napps_manager.is_installed.return_value = True
218
        self.napps_manager.is_enabled.side_effect = [False, True]
219
220
        resp, code = self.api_server._enable_napp('kytos', 'napp')
221
222
        self.assertEqual(resp, '{"response": "enabled"}')
223
        self.assertEqual(code, 200)
224
225
    def test_disable_napp__error_not_installed(self):
226
        """Test _disable_napp method error case when napp is not installed."""
227
        self.napps_manager.is_installed.return_value = False
228
229
        resp, code = self.api_server._disable_napp('kytos', 'napp')
230
231
        self.assertEqual(resp, '{"response": "not installed"}')
232
        self.assertEqual(code, 400)
233
234
    def test_disable_napp__error_not_enabling(self):
235
        """Test _disable_napp method error case when napp is not enabling."""
236
        self.napps_manager.is_installed.return_value = True
237
        self.napps_manager.is_enabled.side_effect = [True, True]
238
239
        resp, code = self.api_server._disable_napp('kytos', 'napp')
240
241
        self.assertEqual(resp, '{"response": "error"}')
242
        self.assertEqual(code, 500)
243
244
    def test_disable_napp__success(self):
245
        """Test _disable_napp method success case."""
246
        self.napps_manager.is_installed.return_value = True
247
        self.napps_manager.is_enabled.side_effect = [True, False]
248
249
        resp, code = self.api_server._disable_napp('kytos', 'napp')
250
251
        self.assertEqual(resp, '{"response": "disabled"}')
252
        self.assertEqual(code, 200)
253
254
    def test_install_napp__error_not_installing(self):
255
        """Test _install_napp method error case when napp is not installing."""
256
        self.napps_manager.is_installed.return_value = False
257
        self.napps_manager.install.return_value = False
258
259
        resp, code = self.api_server._install_napp('kytos', 'napp')
260
261
        self.assertEqual(resp, '{"response": "error"}')
262
        self.assertEqual(code, 500)
263
264
    def test_install_napp__http_error(self):
265
        """Test _install_napp method to http error case."""
266
        self.napps_manager.is_installed.return_value = False
267
        self.napps_manager.install.side_effect = HTTPError('url', 123, 'msg',
268
                                                           'hdrs', MagicMock())
269
270
        resp, code = self.api_server._install_napp('kytos', 'napp')
271
272
        self.assertEqual(resp, '{"response": "error"}')
273
        self.assertEqual(code, 123)
274
275
    def test_install_napp__success_is_installed(self):
276
        """Test _install_napp method success case when napp is installed."""
277
        self.napps_manager.is_installed.return_value = True
278
279
        resp, code = self.api_server._install_napp('kytos', 'napp')
280
281
        self.assertEqual(resp, '{"response": "installed"}')
282
        self.assertEqual(code, 200)
283
284
    def test_install_napp__success(self):
285
        """Test _install_napp method success case."""
286
        self.napps_manager.is_installed.return_value = False
287
        self.napps_manager.install.return_value = True
288
289
        resp, code = self.api_server._install_napp('kytos', 'napp')
290
291
        self.assertEqual(resp, '{"response": "installed"}')
292
        self.assertEqual(code, 200)
293
294
    def test_uninstall_napp__error_not_uninstalling(self):
295
        """Test _uninstall_napp method error case when napp is not
296
           uninstalling.
297
        """
298
        self.napps_manager.is_installed.return_value = True
299
        self.napps_manager.uninstall.return_value = False
300
301
        resp, code = self.api_server._uninstall_napp('kytos', 'napp')
302
303
        self.assertEqual(resp, '{"response": "error"}')
304
        self.assertEqual(code, 500)
305
306
    def test_uninstall_napp__success_not_installed(self):
307
        """Test _uninstall_napp method success case when napp is not
308
           installed.
309
        """
310
        self.napps_manager.is_installed.return_value = False
311
312
        resp, code = self.api_server._uninstall_napp('kytos', 'napp')
313
314
        self.assertEqual(resp, '{"response": "uninstalled"}')
315
        self.assertEqual(code, 200)
316
317
    def test_uninstall_napp__success(self):
318
        """Test _uninstall_napp method success case."""
319
        self.napps_manager.is_installed.return_value = True
320
        self.napps_manager.uninstall.return_value = True
321
322
        resp, code = self.api_server._uninstall_napp('kytos', 'napp')
323
324
        self.assertEqual(resp, '{"response": "uninstalled"}')
325
        self.assertEqual(code, 200)
326
327
    def test_list_enabled_napps(self):
328
        """Test _list_enabled_napps method."""
329
        napp = MagicMock()
330
        napp.username = 'kytos'
331
        napp.name = 'name'
332
        self.napps_manager.get_enabled_napps.return_value = [napp]
333
334
        enabled_napps, code = self.api_server._list_enabled_napps()
335
336
        self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
337
        self.assertEqual(code, 200)
338
339
    def test_list_installed_napps(self):
340
        """Test _list_installed_napps method."""
341
        napp = MagicMock()
342
        napp.username = 'kytos'
343
        napp.name = 'name'
344
        self.napps_manager.get_installed_napps.return_value = [napp]
345
346
        enabled_napps, code = self.api_server._list_installed_napps()
347
348
        self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
349
        self.assertEqual(code, 200)
350
351
    def test_get_napp_metadata__not_installed(self):
352
        """Test _get_napp_metadata method to error case when napp is not
353
           installed."""
354
        self.napps_manager.is_installed.return_value = False
355
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
356
                                                        'version')
357
358
        self.assertEqual(resp, 'NApp is not installed.')
359
        self.assertEqual(code, 400)
360
361
    def test_get_napp_metadata__invalid_key(self):
362
        """Test _get_napp_metadata method to error case when key is invalid."""
363
        self.napps_manager.is_installed.return_value = True
364
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
365
                                                        'any')
366
367
        self.assertEqual(resp, 'Invalid key.')
368
        self.assertEqual(code, 400)
369
370
    def test_get_napp_metadata(self):
371
        """Test _get_napp_metadata method."""
372
        data = '{"username": "kytos", \
373
                 "name": "napp", \
374
                 "version": "1.0"}'
375
        self.napps_manager.is_installed.return_value = True
376
        self.napps_manager.get_napp_metadata.return_value = data
377
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
378
                                                        'version')
379
380
        expected_metadata = json.dumps({'version': data})
381
        self.assertEqual(resp, expected_metadata)
382
        self.assertEqual(code, 200)
383
384
    @staticmethod
385
    def __custom_endpoint():
386
        """Custom method used by APIServer."""
387
        return "Custom Endpoint"
388
389
390
class RESTNApp:  # pylint: disable=too-few-public-methods
391
    """Bare minimum for the decorator to work. Not a functional NApp."""
392
393
    def __init__(self):
394
        self.username = 'test'
395
        self.name = 'MyNApp'
396
397
398
class TestAPIDecorator(unittest.TestCase):
399
    """@rest should have the same effect as ``Flask.route``."""
400
401
    @classmethod
402
    def test_flask_call(cls):
403
        """@rest params should be forwarded to Flask."""
404
        rule = 'rule'
405
        # Use sentinels to be sure they are not changed.
406
        options = dict(param1=sentinel.val1, param2=sentinel.val2)
407
408
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
409
            """API decorator example usage."""
410
411
            @rest(rule, **options)
412
            def my_endpoint(self):
413
                """Do nothing."""
414
415
        napp = MyNApp()
416
        server = cls._mock_api_server(napp)
417
        server.app.add_url_rule.assert_called_once_with(
418
            '/api/test/MyNApp/' + rule, None, napp.my_endpoint, **options)
419
420
    @classmethod
421
    def test_remove_napp_endpoints(cls):
422
        """Test remove napp endpoints"""
423
        class MyNApp:  # pylint: disable=too-few-public-methods
424
            """API decorator example usage."""
425
426
            def __init__(self):
427
                self.username = 'test'
428
                self.name = 'MyNApp'
429
430
        napp = MyNApp()
431
        server = cls._mock_api_server(napp)
432
433
        rule = Mock()
434
        rule.methods = ['GET', 'POST']
435
        rule.rule.startswith.return_value = True
436
        endpoint = 'username/napp_name'
437
        rule.endpoint = endpoint
438
439
        server.app.url_map.iter_rules.return_value = [rule]
440
        server.app.view_functions.pop.return_value = rule
441
        # pylint: disable=protected-access
442
        server.app.url_map._rules.pop.return_value = rule
443
        # pylint: enable=protected-access
444
445
        server.remove_napp_endpoints(napp)
446
        server.app.view_functions.pop.assert_called_once_with(endpoint)
447
        # pylint: disable=protected-access
448
        server.app.url_map._rules.pop.assert_called_once_with(0)
449
        # pylint: enable=protected-access
450
451
    @classmethod
452
    def test_rule_with_slash(cls):
453
        """There should be no double slashes in a rule."""
454
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
455
            """API decorator example usage."""
456
457
            @rest('/rule')
458
            def my_endpoint(self):
459
                """Do nothing."""
460
461
        cls._assert_rule_is_added(MyNApp)
462
463
    @classmethod
464
    def test_rule_from_classmethod(cls):
465
        """Use class methods as endpoints as well."""
466
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
467
            """API decorator example usage."""
468
469
            @rest('/rule')
470
            @classmethod
471
            def my_endpoint(cls):
472
                """Do nothing."""
473
474
        cls._assert_rule_is_added(MyNApp)
475
476
    @classmethod
477
    def test_rule_from_staticmethod(cls):
478
        """Use static methods as endpoints as well."""
479
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
480
            """API decorator example usage."""
481
482
            @rest('/rule')
483
            @staticmethod
484
            def my_endpoint():
485
                """Do nothing."""
486
487
        cls._assert_rule_is_added(MyNApp)
488
489
    @classmethod
490
    def _assert_rule_is_added(cls, napp_class):
491
        """Assert Flask's add_url_rule was called with the right parameters."""
492
        napp = napp_class()
493
        server = cls._mock_api_server(napp)
494
        server.app.add_url_rule.assert_called_once_with(
495
            '/api/test/MyNApp/rule', None, napp.my_endpoint)
496
497
    @staticmethod
498
    def _mock_api_server(napp):
499
        """Instantiate APIServer, mock ``.app`` and start ``napp`` API."""
500
        server = APIServer('test')
501
        server.app = Mock()  # Flask app
502
        server.app.url_map.iter_rules.return_value = []
503
504
        server.register_napp_endpoints(napp)
505
        return server
506