tests.unit.test_core.test_api_server.TestAPIServer.test_web_ui__success() - Code Metrics - Inspection of "Handle URLError when it's not possible to fetch th..." - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b4d101...11876d )

by Humberto

created 2020-10-01 19:33 UTC

TestAPIServer.test_web_ui__success() A

↳ Parent: tests.unit.test_core.test_api_server

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	3
dl	0
loc	8
rs	10
c	0
b	0
f	0

"""APIServer tests."""
import json
import unittest
import warnings
# Disable not-grouped imports that conflicts with isort
from unittest.mock import (MagicMock, Mock, patch,  # pylint: disable=C0412
                           sentinel)
from urllib.error import HTTPError

from kytos.core.api_server import APIServer
from kytos.core.napps import rest

KYTOS_CORE_API = "http://127.0.0.1:8181/api/kytos/"
API_URI = KYTOS_CORE_API+"core"


# pylint: disable=protected-access, too-many-public-methods
class TestAPIServer(unittest.TestCase):
    """Test the class APIServer."""

    def setUp(self):
        """Instantiate a APIServer."""
        self.api_server = APIServer('CustomName', False)
        self.napps_manager = MagicMock()
        self.api_server.server = MagicMock()
        self.api_server.napps_manager = self.napps_manager
        self.api_server.napps_dir = 'napps_dir'
        self.api_server.flask_dir = 'flask_dir'

    def test_deprecation_warning(self):
        """Deprecated method should suggest @rest decorator."""
        with warnings.catch_warnings(record=True) as wrngs:
            warnings.simplefilter("always")  # trigger all warnings
            self.api_server.register_rest_endpoint(
                'rule', lambda x: x, ['POST'])
            self.assertEqual(1, len(wrngs))
            warning = wrngs[0]
            self.assertEqual(warning.category, DeprecationWarning)
            self.assertIn('@rest', str(warning.message))

    def test_run(self):
        """Test run method."""
        self.api_server.run()

        self.api_server.server.run.assert_called_with(self.api_server.app,
                                                      self.api_server.listen,
                                                      self.api_server.port)

    @patch('sys.exit')
    def test_run_error(self, mock_exit):
        """Test run method to error case."""
        self.api_server.server.run.side_effect = OSError
        self.api_server.run()

        mock_exit.assert_called()

    @patch('kytos.core.api_server.request')
    def test_shutdown_api(self, mock_request):
        """Test shutdown_api method."""
        mock_request.host = 'localhost:8181'

        self.api_server.shutdown_api()

        self.api_server.server.stop.assert_called()

    @patch('kytos.core.api_server.request')
    def test_shutdown_api__error(self, mock_request):
        """Test shutdown_api method to error case."""
        mock_request.host = 'any:port'

        self.api_server.shutdown_api()

        self.api_server.server.stop.assert_not_called()

    def test_status_api(self):
        """Test status_api method."""
        status = self.api_server.status_api()
        self.assertEqual(status, ('{"response": "running"}', 200))

    @patch('kytos.core.api_server.urlopen')
    def test_stop_api_server(self, mock_urlopen):
        """Test stop_api_server method."""
        self.api_server.stop_api_server()

        url = "%s/shutdown" % API_URI
        mock_urlopen.assert_called_with(url)

    @patch('kytos.core.api_server.send_file')
    @patch('os.path.exists', return_value=True)
    def test_static_web_ui__success(self, *args):
        """Test static_web_ui method to success case."""
        (_, mock_send_file) = args
        self.api_server.static_web_ui('kytos', 'napp', 'filename')

        mock_send_file.assert_called_with('napps_dir/kytos/napp/ui/filename')

    @patch('os.path.exists', return_value=False)
    def test_static_web_ui__error(self, _):
        """Test static_web_ui method to error case."""
        resp, code = self.api_server.static_web_ui('kytos', 'napp', 'filename')

        self.assertEqual(resp, '')
        self.assertEqual(code, 404)

    @patch('kytos.core.api_server.glob')
    def test_get_ui_components(self, mock_glob):
        """Test get_ui_components method."""
        with self.api_server.app.app_context():
            mock_glob.return_value = ['napps_dir/*/*/ui/*/*.kytos']
            response = self.api_server.get_ui_components('all')

            expected_json = [{'name': '*-*-*-*', 'url': 'ui/*/*/*/*.kytos'}]
            self.assertEqual(response.json, expected_json)
            self.assertEqual(response.status_code, 200)

    @patch('os.path')
    @patch('kytos.core.api_server.send_file')
    def test_web_ui__success(self, mock_send_file, ospath_mock):
        """Test web_ui method."""
        ospath_mock.exists.return_value = True
        self.api_server.web_ui()

        mock_send_file.assert_called_with('flask_dir/index.html')

    @patch('os.path')
    def test_web_ui__error(self, ospath_mock):
        """Test web_ui method."""
        ospath_mock.exists.return_value = False
        _, error = self.api_server.web_ui()

        self.assertEqual(error, 404)

    @patch('kytos.core.api_server.urlretrieve')
    @patch('kytos.core.api_server.urlopen')
    @patch('zipfile.ZipFile')
    @patch('os.path.exists')
    @patch('os.mkdir')
    @patch('shutil.move')
    def test_update_web_ui(self, *args):
        """Test update_web_ui method."""
        (_, _, mock_exists, mock_zipfile, mock_urlopen,
         mock_urlretrieve) = args
        zipfile = MagicMock()
        zipfile.testzip.return_value = None
        mock_zipfile.return_value = zipfile

        data = json.dumps({'tag_name': 1.0})
        url_response = MagicMock()
        url_response.readlines.return_value = [data]
        mock_urlopen.return_value = url_response

        mock_exists.side_effect = [False, True]

        response = self.api_server.update_web_ui()

        url = 'https://github.com/kytos/ui/releases/download/1.0/latest.zip'
        mock_urlretrieve.assert_called_with(url)
        self.assertEqual(response, 'updated the web ui')

    @patch('kytos.core.api_server.urlretrieve')
    @patch('kytos.core.api_server.urlopen')
    @patch('os.path.exists')
    def test_update_web_ui__http_error(self, *args):
        """Test update_web_ui method to http error case."""
        (mock_exists, mock_urlopen, mock_urlretrieve) = args

        data = json.dumps({'tag_name': 1.0})
        url_response = MagicMock()
        url_response.readlines.return_value = [data]
        mock_urlopen.return_value = url_response
        mock_urlretrieve.side_effect = HTTPError('url', 123, 'msg', 'hdrs',
                                                 MagicMock())

        mock_exists.return_value = False

        response = self.api_server.update_web_ui()

        expected_response = 'Uri not found https://github.com/kytos/ui/' + \
                            'releases/download/1.0/latest.zip.'
        self.assertEqual(response, expected_response)

    @patch('kytos.core.api_server.urlretrieve')
    @patch('kytos.core.api_server.urlopen')
    @patch('zipfile.ZipFile')
    @patch('os.path.exists')
    def test_update_web_ui__zip_error(self, *args):
        """Test update_web_ui method to error case in zip file."""
        (mock_exists, mock_zipfile, mock_urlopen, _) = args
        zipfile = MagicMock()
        zipfile.testzip.return_value = 'any'
        mock_zipfile.return_value = zipfile

        data = json.dumps({'tag_name': 1.0})
        url_response = MagicMock()
        url_response.readlines.return_value = [data]
        mock_urlopen.return_value = url_response

        mock_exists.return_value = False

        response = self.api_server.update_web_ui()

        expected_response = 'Zip file from https://github.com/kytos/ui/' + \
                            'releases/download/1.0/latest.zip is corrupted.'
        self.assertEqual(response, expected_response)

    def test_enable_napp__error_not_installed(self):
        """Test _enable_napp method error case when napp is not installed."""
        self.napps_manager.is_installed.return_value = False

        resp, code = self.api_server._enable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "not installed"}')
        self.assertEqual(code, 400)

    def test_enable_napp__error_not_enabling(self):
        """Test _enable_napp method error case when napp is not enabling."""
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.is_enabled.side_effect = [False, False]

        resp, code = self.api_server._enable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "error"}')
        self.assertEqual(code, 500)

    def test_enable_napp__success(self):
        """Test _enable_napp method success case."""
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.is_enabled.side_effect = [False, True]

        resp, code = self.api_server._enable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "enabled"}')
        self.assertEqual(code, 200)

    def test_disable_napp__error_not_installed(self):
        """Test _disable_napp method error case when napp is not installed."""
        self.napps_manager.is_installed.return_value = False

        resp, code = self.api_server._disable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "not installed"}')
        self.assertEqual(code, 400)

    def test_disable_napp__error_not_enabling(self):
        """Test _disable_napp method error case when napp is not enabling."""
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.is_enabled.side_effect = [True, True]

        resp, code = self.api_server._disable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "error"}')
        self.assertEqual(code, 500)

    def test_disable_napp__success(self):
        """Test _disable_napp method success case."""
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.is_enabled.side_effect = [True, False]

        resp, code = self.api_server._disable_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "disabled"}')
        self.assertEqual(code, 200)

    def test_install_napp__error_not_installing(self):
        """Test _install_napp method error case when napp is not installing."""
        self.napps_manager.is_installed.return_value = False
        self.napps_manager.install.return_value = False

        resp, code = self.api_server._install_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "error"}')
        self.assertEqual(code, 500)

    def test_install_napp__http_error(self):
        """Test _install_napp method to http error case."""
        self.napps_manager.is_installed.return_value = False
        self.napps_manager.install.side_effect = HTTPError('url', 123, 'msg',
                                                           'hdrs', MagicMock())

        resp, code = self.api_server._install_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "error"}')
        self.assertEqual(code, 123)

    def test_install_napp__success_is_installed(self):
        """Test _install_napp method success case when napp is installed."""
        self.napps_manager.is_installed.return_value = True

        resp, code = self.api_server._install_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "installed"}')
        self.assertEqual(code, 200)

    def test_install_napp__success(self):
        """Test _install_napp method success case."""
        self.napps_manager.is_installed.return_value = False
        self.napps_manager.install.return_value = True

        resp, code = self.api_server._install_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "installed"}')
        self.assertEqual(code, 200)

    def test_uninstall_napp__error_not_uninstalling(self):
        """Test _uninstall_napp method error case when napp is not
           uninstalling.
        """
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.uninstall.return_value = False

        resp, code = self.api_server._uninstall_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "error"}')
        self.assertEqual(code, 500)

    def test_uninstall_napp__success_not_installed(self):
        """Test _uninstall_napp method success case when napp is not
           installed.
        """
        self.napps_manager.is_installed.return_value = False

        resp, code = self.api_server._uninstall_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "uninstalled"}')
        self.assertEqual(code, 200)

    def test_uninstall_napp__success(self):
        """Test _uninstall_napp method success case."""
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.uninstall.return_value = True

        resp, code = self.api_server._uninstall_napp('kytos', 'napp')

        self.assertEqual(resp, '{"response": "uninstalled"}')
        self.assertEqual(code, 200)

    def test_list_enabled_napps(self):
        """Test _list_enabled_napps method."""
        napp = MagicMock()
        napp.username = 'kytos'
        napp.name = 'name'
        self.napps_manager.get_enabled_napps.return_value = [napp]

        enabled_napps, code = self.api_server._list_enabled_napps()

        self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
        self.assertEqual(code, 200)

    def test_list_installed_napps(self):
        """Test _list_installed_napps method."""
        napp = MagicMock()
        napp.username = 'kytos'
        napp.name = 'name'
        self.napps_manager.get_installed_napps.return_value = [napp]

        enabled_napps, code = self.api_server._list_installed_napps()

        self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
        self.assertEqual(code, 200)

    def test_get_napp_metadata__not_installed(self):
        """Test _get_napp_metadata method to error case when napp is not
           installed."""
        self.napps_manager.is_installed.return_value = False
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
                                                        'version')

        self.assertEqual(resp, 'NApp is not installed.')
        self.assertEqual(code, 400)

    def test_get_napp_metadata__invalid_key(self):
        """Test _get_napp_metadata method to error case when key is invalid."""
        self.napps_manager.is_installed.return_value = True
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
                                                        'any')

        self.assertEqual(resp, 'Invalid key.')
        self.assertEqual(code, 400)

    def test_get_napp_metadata(self):
        """Test _get_napp_metadata method."""
        data = '{"username": "kytos", \
                 "name": "napp", \
                 "version": "1.0"}'
        self.napps_manager.is_installed.return_value = True
        self.napps_manager.get_napp_metadata.return_value = data
        resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
                                                        'version')

        expected_metadata = json.dumps({'version': data})
        self.assertEqual(resp, expected_metadata)
        self.assertEqual(code, 200)

    @staticmethod
    def __custom_endpoint():
        """Custom method used by APIServer."""
        return "Custom Endpoint"


class RESTNApp:  # pylint: disable=too-few-public-methods
    """Bare minimum for the decorator to work. Not a functional NApp."""

    def __init__(self):
        self.username = 'test'
        self.name = 'MyNApp'
        self.napp_id = 'test/MyNApp'


class TestAPIDecorator(unittest.TestCase):
    """@rest should have the same effect as ``Flask.route``."""

    @classmethod
    @patch('kytos.core.api_server.Blueprint')
    def test_flask_call(cls, mock_blueprint):
        """@rest params should be forwarded to Flask."""
        rule = 'rule'
        # Use sentinels to be sure they are not changed.
        options = dict(param1=sentinel.val1, param2=sentinel.val2)

        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
            """API decorator example usage."""

            @rest(rule, **options)
            def my_endpoint(self):
                """Do nothing."""

        blueprint = Mock()
        mock_blueprint.return_value = blueprint

        napp = MyNApp()
        cls._mock_api_server(napp)
        blueprint.add_url_rule.assert_called_once_with(
            '/api/test/MyNApp/' + rule, None, napp.my_endpoint, **options)

    @classmethod
    def test_remove_napp_endpoints(cls):
        """Test remove napp endpoints"""
        class MyNApp:  # pylint: disable=too-few-public-methods
            """API decorator example usage."""

            def __init__(self):
                self.username = 'test'
                self.name = 'MyNApp'
                self.napp_id = 'test/MyNApp'

        napp = MyNApp()
        server = cls._mock_api_server(napp)

        rule = Mock()
        rule.methods = ['GET', 'POST']
        rule.rule.startswith.return_value = True
        endpoint = 'username/napp_name'
        rule.endpoint = endpoint

        server.app.url_map.iter_rules.return_value = [rule]
        server.app.view_functions.pop.return_value = rule
        # pylint: disable=protected-access
        server.app.url_map._rules.pop.return_value = rule
        # pylint: enable=protected-access

        server.remove_napp_endpoints(napp)
        server.app.view_functions.pop.assert_called_once_with(endpoint)
        # pylint: disable=protected-access
        server.app.url_map._rules.pop.assert_called_once_with(0)
        # pylint: enable=protected-access
        server.app.blueprints.pop.assert_called_once_with(napp.napp_id)

    @classmethod
    @patch('kytos.core.api_server.Blueprint')
    def test_rule_with_slash(cls, mock_blueprint):
        """There should be no double slashes in a rule."""
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
            """API decorator example usage."""

            @rest('/rule')
            def my_endpoint(self):
                """Do nothing."""

        blueprint = Mock()
        mock_blueprint.return_value = blueprint
        cls._assert_rule_is_added(MyNApp, blueprint)

    @classmethod
    @patch('kytos.core.api_server.Blueprint')
    def test_rule_from_classmethod(cls, mock_blueprint):
        """Use class methods as endpoints as well."""
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
            """API decorator example usage."""

            @rest('/rule')
            @classmethod
            def my_endpoint(cls):
                """Do nothing."""

        blueprint = Mock()
        mock_blueprint.return_value = blueprint
        cls._assert_rule_is_added(MyNApp, blueprint)

    @classmethod
    @patch('kytos.core.api_server.Blueprint')
    def test_rule_from_staticmethod(cls, mock_blueprint):
        """Use static methods as endpoints as well."""
        class MyNApp(RESTNApp):  # pylint: disable=too-few-public-methods
            """API decorator example usage."""

            @rest('/rule')
            @staticmethod
            def my_endpoint():
                """Do nothing."""

        blueprint = Mock()
        mock_blueprint.return_value = blueprint
        cls._assert_rule_is_added(MyNApp, blueprint)

    @classmethod
    def _assert_rule_is_added(cls, napp_class, blueprint):
        """Assert Flask's add_url_rule was called with the right parameters."""
        napp = napp_class()
        cls._mock_api_server(napp)
        blueprint.add_url_rule.assert_called_once_with(
            '/api/test/MyNApp/rule', None, napp.my_endpoint)

    @staticmethod
    def _mock_api_server(napp):
        """Instantiate APIServer, mock ``.app`` and start ``napp`` API."""
        server = APIServer('test')
        server.app = Mock()  # Flask app
        server.app.url_map.iter_rules.return_value = []

        server.register_napp_endpoints(napp)
        return server


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('os.path')
117			@patch('kytos.core.api_server.send_file')
118			def test_web_ui__success(self, mock_send_file, ospath_mock):
119			"""Test web_ui method."""
120			ospath_mock.exists.return_value = True
121			self.api_server.web_ui()
122
123			mock_send_file.assert_called_with('flask_dir/index.html')
124
125			@patch('os.path')
126			def test_web_ui__error(self, ospath_mock):
127			"""Test web_ui method."""
128			ospath_mock.exists.return_value = False
129			_, error = self.api_server.web_ui()
130
131			self.assertEqual(error, 404)
132
133			@patch('kytos.core.api_server.urlretrieve')
134			@patch('kytos.core.api_server.urlopen')
135			@patch('zipfile.ZipFile')
136			@patch('os.path.exists')
137			@patch('os.mkdir')
138			@patch('shutil.move')
139			def test_update_web_ui(self, *args):
140			"""Test update_web_ui method."""
141			(_, _, mock_exists, mock_zipfile, mock_urlopen,
142			mock_urlretrieve) = args
143			zipfile = MagicMock()
144			zipfile.testzip.return_value = None
145			mock_zipfile.return_value = zipfile
146
147			data = json.dumps({'tag_name': 1.0})
148			url_response = MagicMock()
149			url_response.readlines.return_value = [data]
150			mock_urlopen.return_value = url_response
151
152			mock_exists.side_effect = [False, True]
153
154			response = self.api_server.update_web_ui()
155
156			url = 'https://github.com/kytos/ui/releases/download/1.0/latest.zip'
157			mock_urlretrieve.assert_called_with(url)
158			self.assertEqual(response, 'updated the web ui')
159
160			@patch('kytos.core.api_server.urlretrieve')
161			@patch('kytos.core.api_server.urlopen')
162			@patch('os.path.exists')
163			def test_update_web_ui__http_error(self, *args):
164			"""Test update_web_ui method to http error case."""
165			(mock_exists, mock_urlopen, mock_urlretrieve) = args
166
167			data = json.dumps({'tag_name': 1.0})
168			url_response = MagicMock()
169			url_response.readlines.return_value = [data]
170			mock_urlopen.return_value = url_response
171			mock_urlretrieve.side_effect = HTTPError('url', 123, 'msg', 'hdrs',
172			MagicMock())
173
174			mock_exists.return_value = False
175
176			response = self.api_server.update_web_ui()
177
178			expected_response = 'Uri not found https://github.com/kytos/ui/' + \
179			'releases/download/1.0/latest.zip.'
180			self.assertEqual(response, expected_response)
181
182			@patch('kytos.core.api_server.urlretrieve')
183			@patch('kytos.core.api_server.urlopen')
184			@patch('zipfile.ZipFile')
185			@patch('os.path.exists')
186			def test_update_web_ui__zip_error(self, *args):
187			"""Test update_web_ui method to error case in zip file."""
188			(mock_exists, mock_zipfile, mock_urlopen, _) = args
189			zipfile = MagicMock()
190			zipfile.testzip.return_value = 'any'
191			mock_zipfile.return_value = zipfile
192
193			data = json.dumps({'tag_name': 1.0})
194			url_response = MagicMock()
195			url_response.readlines.return_value = [data]
196			mock_urlopen.return_value = url_response
197
198			mock_exists.return_value = False
199
200			response = self.api_server.update_web_ui()
201
202			expected_response = 'Zip file from https://github.com/kytos/ui/' + \
203			'releases/download/1.0/latest.zip is corrupted.'
204			self.assertEqual(response, expected_response)
205
206			def test_enable_napp__error_not_installed(self):
207			"""Test _enable_napp method error case when napp is not installed."""
208			self.napps_manager.is_installed.return_value = False
209
210			resp, code = self.api_server._enable_napp('kytos', 'napp')
211
212			self.assertEqual(resp, '{"response": "not installed"}')
213			self.assertEqual(code, 400)
214
215			def test_enable_napp__error_not_enabling(self):
216			"""Test _enable_napp method error case when napp is not enabling."""
217			self.napps_manager.is_installed.return_value = True
218			self.napps_manager.is_enabled.side_effect = [False, False]
219
220			resp, code = self.api_server._enable_napp('kytos', 'napp')
221
222			self.assertEqual(resp, '{"response": "error"}')
223			self.assertEqual(code, 500)
224
225			def test_enable_napp__success(self):
226			"""Test _enable_napp method success case."""
227			self.napps_manager.is_installed.return_value = True
228			self.napps_manager.is_enabled.side_effect = [False, True]
229
230			resp, code = self.api_server._enable_napp('kytos', 'napp')
231
232			self.assertEqual(resp, '{"response": "enabled"}')
233			self.assertEqual(code, 200)
234
235			def test_disable_napp__error_not_installed(self):
236			"""Test _disable_napp method error case when napp is not installed."""
237			self.napps_manager.is_installed.return_value = False
238
239			resp, code = self.api_server._disable_napp('kytos', 'napp')
240
241			self.assertEqual(resp, '{"response": "not installed"}')
242			self.assertEqual(code, 400)
243
244			def test_disable_napp__error_not_enabling(self):
245			"""Test _disable_napp method error case when napp is not enabling."""
246			self.napps_manager.is_installed.return_value = True
247			self.napps_manager.is_enabled.side_effect = [True, True]
248
249			resp, code = self.api_server._disable_napp('kytos', 'napp')
250
251			self.assertEqual(resp, '{"response": "error"}')
252			self.assertEqual(code, 500)
253
254			def test_disable_napp__success(self):
255			"""Test _disable_napp method success case."""
256			self.napps_manager.is_installed.return_value = True
257			self.napps_manager.is_enabled.side_effect = [True, False]
258
259			resp, code = self.api_server._disable_napp('kytos', 'napp')
260
261			self.assertEqual(resp, '{"response": "disabled"}')
262			self.assertEqual(code, 200)
263
264			def test_install_napp__error_not_installing(self):
265			"""Test _install_napp method error case when napp is not installing."""
266			self.napps_manager.is_installed.return_value = False
267			self.napps_manager.install.return_value = False
268
269			resp, code = self.api_server._install_napp('kytos', 'napp')
270
271			self.assertEqual(resp, '{"response": "error"}')
272			self.assertEqual(code, 500)
273
274			def test_install_napp__http_error(self):
275			"""Test _install_napp method to http error case."""
276			self.napps_manager.is_installed.return_value = False
277			self.napps_manager.install.side_effect = HTTPError('url', 123, 'msg',
278			'hdrs', MagicMock())
279
280			resp, code = self.api_server._install_napp('kytos', 'napp')
281
282			self.assertEqual(resp, '{"response": "error"}')
283			self.assertEqual(code, 123)
284
285			def test_install_napp__success_is_installed(self):
286			"""Test _install_napp method success case when napp is installed."""
287			self.napps_manager.is_installed.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_install_napp__success(self):
295			"""Test _install_napp method success case."""
296			self.napps_manager.is_installed.return_value = False
297			self.napps_manager.install.return_value = True
298
299			resp, code = self.api_server._install_napp('kytos', 'napp')
300
301			self.assertEqual(resp, '{"response": "installed"}')
302			self.assertEqual(code, 200)
303
304			def test_uninstall_napp__error_not_uninstalling(self):
305			"""Test _uninstall_napp method error case when napp is not
306			uninstalling.
307			"""
308			self.napps_manager.is_installed.return_value = True
309			self.napps_manager.uninstall.return_value = False
310
311			resp, code = self.api_server._uninstall_napp('kytos', 'napp')
312
313			self.assertEqual(resp, '{"response": "error"}')
314			self.assertEqual(code, 500)
315
316			def test_uninstall_napp__success_not_installed(self):
317			"""Test _uninstall_napp method success case when napp is not
318			installed.
319			"""
320			self.napps_manager.is_installed.return_value = False
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_uninstall_napp__success(self):
328			"""Test _uninstall_napp method success case."""
329			self.napps_manager.is_installed.return_value = True
330			self.napps_manager.uninstall.return_value = True
331
332			resp, code = self.api_server._uninstall_napp('kytos', 'napp')
333
334			self.assertEqual(resp, '{"response": "uninstalled"}')
335			self.assertEqual(code, 200)
336
337			def test_list_enabled_napps(self):
338			"""Test _list_enabled_napps method."""
339			napp = MagicMock()
340			napp.username = 'kytos'
341			napp.name = 'name'
342			self.napps_manager.get_enabled_napps.return_value = [napp]
343
344			enabled_napps, code = self.api_server._list_enabled_napps()
345
346			self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
347			self.assertEqual(code, 200)
348
349			def test_list_installed_napps(self):
350			"""Test _list_installed_napps method."""
351			napp = MagicMock()
352			napp.username = 'kytos'
353			napp.name = 'name'
354			self.napps_manager.get_installed_napps.return_value = [napp]
355
356			enabled_napps, code = self.api_server._list_installed_napps()
357
358			self.assertEqual(enabled_napps, '{"napps": [["kytos", "name"]]}')
359			self.assertEqual(code, 200)
360
361			def test_get_napp_metadata__not_installed(self):
362			"""Test _get_napp_metadata method to error case when napp is not
363			installed."""
364			self.napps_manager.is_installed.return_value = False
365			resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
366			'version')
367
368			self.assertEqual(resp, 'NApp is not installed.')
369			self.assertEqual(code, 400)
370
371			def test_get_napp_metadata__invalid_key(self):
372			"""Test _get_napp_metadata method to error case when key is invalid."""
373			self.napps_manager.is_installed.return_value = True
374			resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
375			'any')
376
377			self.assertEqual(resp, 'Invalid key.')
378			self.assertEqual(code, 400)
379
380			def test_get_napp_metadata(self):
381			"""Test _get_napp_metadata method."""
382			data = '{"username": "kytos", \
383			"name": "napp", \
384			"version": "1.0"}'
385			self.napps_manager.is_installed.return_value = True
386			self.napps_manager.get_napp_metadata.return_value = data
387			resp, code = self.api_server._get_napp_metadata('kytos', 'napp',
388			'version')
389
390			expected_metadata = json.dumps({'version': data})
391			self.assertEqual(resp, expected_metadata)
392			self.assertEqual(code, 200)
393
394			@staticmethod
395			def __custom_endpoint():
396			"""Custom method used by APIServer."""
397			return "Custom Endpoint"
398
399
400			class RESTNApp: # pylint: disable=too-few-public-methods
401			"""Bare minimum for the decorator to work. Not a functional NApp."""
402
403			def __init__(self):
404			self.username = 'test'
405			self.name = 'MyNApp'
406			self.napp_id = 'test/MyNApp'
407
408
409			class TestAPIDecorator(unittest.TestCase):
410			"""@rest should have the same effect as ``Flask.route``."""
411
412			@classmethod
413			@patch('kytos.core.api_server.Blueprint')
414			def test_flask_call(cls, mock_blueprint):
415			"""@rest params should be forwarded to Flask."""
416			rule = 'rule'
417			# Use sentinels to be sure they are not changed.
418			options = dict(param1=sentinel.val1, param2=sentinel.val2)
419
420			class MyNApp(RESTNApp): # pylint: disable=too-few-public-methods
421			"""API decorator example usage."""
422
423			@rest(rule, **options)
424			def my_endpoint(self):
425			"""Do nothing."""
426
427			blueprint = Mock()
428			mock_blueprint.return_value = blueprint
429
430			napp = MyNApp()
431			cls._mock_api_server(napp)
432			blueprint.add_url_rule.assert_called_once_with(
433			'/api/test/MyNApp/' + rule, None, napp.my_endpoint, **options)
434
435			@classmethod
436			def test_remove_napp_endpoints(cls):
437			"""Test remove napp endpoints"""
438			class MyNApp: # pylint: disable=too-few-public-methods
439			"""API decorator example usage."""
440
441			def __init__(self):
442			self.username = 'test'
443			self.name = 'MyNApp'
444			self.napp_id = 'test/MyNApp'
445
446			napp = MyNApp()
447			server = cls._mock_api_server(napp)
448
449			rule = Mock()
450			rule.methods = ['GET', 'POST']
451			rule.rule.startswith.return_value = True
452			endpoint = 'username/napp_name'
453			rule.endpoint = endpoint
454
455			server.app.url_map.iter_rules.return_value = [rule]
456			server.app.view_functions.pop.return_value = rule
457			# pylint: disable=protected-access
458			server.app.url_map._rules.pop.return_value = rule
459			# pylint: enable=protected-access
460
461			server.remove_napp_endpoints(napp)
462			server.app.view_functions.pop.assert_called_once_with(endpoint)
463			# pylint: disable=protected-access
464			server.app.url_map._rules.pop.assert_called_once_with(0)
465			# pylint: enable=protected-access
466			server.app.blueprints.pop.assert_called_once_with(napp.napp_id)
467
468			@classmethod
469			@patch('kytos.core.api_server.Blueprint')
470			def test_rule_with_slash(cls, mock_blueprint):
471			"""There should be no double slashes in a rule."""
472			class MyNApp(RESTNApp): # pylint: disable=too-few-public-methods
473			"""API decorator example usage."""
474
475			@rest('/rule')
476			def my_endpoint(self):
477			"""Do nothing."""
478
479			blueprint = Mock()
480			mock_blueprint.return_value = blueprint
481			cls._assert_rule_is_added(MyNApp, blueprint)
482
483			@classmethod
484			@patch('kytos.core.api_server.Blueprint')
485			def test_rule_from_classmethod(cls, mock_blueprint):
486			"""Use class methods as endpoints as well."""
487			class MyNApp(RESTNApp): # pylint: disable=too-few-public-methods
488			"""API decorator example usage."""
489
490			@rest('/rule')
491			@classmethod
492			def my_endpoint(cls):
493			"""Do nothing."""
494
495			blueprint = Mock()
496			mock_blueprint.return_value = blueprint
497			cls._assert_rule_is_added(MyNApp, blueprint)
498
499			@classmethod
500			@patch('kytos.core.api_server.Blueprint')
501			def test_rule_from_staticmethod(cls, mock_blueprint):
502			"""Use static methods as endpoints as well."""
503			class MyNApp(RESTNApp): # pylint: disable=too-few-public-methods
504			"""API decorator example usage."""
505
506			@rest('/rule')
507			@staticmethod
508			def my_endpoint():
509			"""Do nothing."""
510
511			blueprint = Mock()
512			mock_blueprint.return_value = blueprint
513			cls._assert_rule_is_added(MyNApp, blueprint)
514
515			@classmethod
516			def _assert_rule_is_added(cls, napp_class, blueprint):
517			"""Assert Flask's add_url_rule was called with the right parameters."""
518			napp = napp_class()
519			cls._mock_api_server(napp)
520			blueprint.add_url_rule.assert_called_once_with(
521			'/api/test/MyNApp/rule', None, napp.my_endpoint)
522
523			@staticmethod
524			def _mock_api_server(napp):
525			"""Instantiate APIServer, mock ``.app`` and start ``napp`` API."""
526			server = APIServer('test')
527			server.app = Mock() # Flask app
528			server.app.url_map.iter_rules.return_value = []
529
530			server.register_napp_endpoints(napp)
531			return server
532

kytos / kytos

Push — master ( b4d101...11876d )

TestAPIServer.test_web_ui__success() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like