tests.unit.test_napps.TestNapps.test_installed_dir() - Code Metrics - Inspection of "Improve unit tests" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#277)

by Gleyberson

created 2020-06-10 17:42 UTC

TestNapps.test_installed_dir() A

↳ Parent: tests.unit.test_napps

Complexity

Conditions

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	2
dl	0
loc	12
rs	9.95
c	0
b	0
f	0

"""kytos.utils.napps.NAppsManager tests."""
import json
import re
import tempfile
import unittest
from pathlib import Path, PurePosixPath
from unittest.mock import MagicMock, Mock, PropertyMock, call, patch

from kytos.utils.exceptions import KytosException
from kytos.utils.napps import NAppsManager
from kytos.utils.settings import SKEL_PATH


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

    def setUp(self):
        """Execute steps before each tests."""
        self.napps_manager = NAppsManager()

    @staticmethod
    def get_napps_response_mock(napps=None):
        """Get mock to napps response."""
        if napps is None:
            napps = [["kytos", "mef_eline"], ["kytos", "of_lldp"]]
        response = json.dumps({"napps": napps})

        mock_response = MagicMock()
        mock_response.getcode.return_value = 200
        mock_response.read.return_value = response
        mock_response.__enter__.return_value = mock_response
        return mock_response

    @patch('urllib.request.urlopen')
    def test_enabled_property(self, mock_urlopen):
        """Test enabled property."""
        data = MagicMock()
        data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
        mock_urlopen.return_value = data

        self.assertEqual(str(self.napps_manager._enabled), 'ABC')

    def test_enabled_property__error(self):
        """Test enabled property to error case."""

        with self.assertRaises(SystemExit):
            # pylint: disable=pointless-statement
            self.napps_manager._enabled
            # pylint: enable=pointless-statement

        self.assertIsNone(self.napps_manager._NAppsManager__local_enabled)

    @patch('urllib.request.urlopen')
    def test_installed_property(self, mock_urlopen):
        """Test installed property."""
        data = MagicMock()
        data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
        mock_urlopen.return_value = data

        self.assertEqual(str(self.napps_manager._installed), 'DEF')

    def test_installed_property__error(self):
        """Test installed property to error case."""

        with self.assertRaises(SystemExit):
            # pylint: disable=pointless-statement
            self.napps_manager._installed
            # pylint: enable=pointless-statement

        self.assertIsNone(self.napps_manager._NAppsManager__local_installed)

    def test_napp_id_property(self):
        """Test napp_id property."""
        self.napps_manager.user = 'user'
        self.napps_manager.napp = 'napp'

        self.assertEqual(self.napps_manager.napp_id, 'user/napp')

    def test_set_napp(self):
        """Test set_napp method."""
        self.napps_manager.set_napp('user', 'napp', 'version')

        self.assertEqual(self.napps_manager.user, 'user')
        self.assertEqual(self.napps_manager.napp, 'napp')
        self.assertEqual(self.napps_manager.version, 'version')

    def test_get_napps(self):
        """Test method get_napps used to find
        enabled and installed napps.
        """
        mock_path = Mock()

        def glob_side_effect(args):
            """Path.glob to mock finding paths with kytos.json file."""
            self.assertEqual(args, "*/*/kytos.json")

            mock_path1 = Mock()
            mock_path1.parts = ['kytos', 'of_core', 'kytos.json']

            mock_path2 = Mock()
            mock_path2.parts = ['kytos', 'of_lldp', 'kytos.json']

            return [mock_path1, mock_path2]

        mock_path.glob = glob_side_effect

        # pylint: disable=protected-access
        get_return = self.napps_manager._get_napps(mock_path)
        self.assertEqual(get_return[0][0], 'kytos')
        self.assertEqual(get_return[0][1], 'of_core')
        self.assertEqual(get_return[1][0], 'kytos')
        self.assertEqual(get_return[1][1], 'of_lldp')

    def test_get_enabled_local(self):

        """Test get_enabled_local used to find
        enabled napps in local machine"""
        # Mock kytos.json path
        mock_path = Mock()

        def glob_side_effect(args):
            """Path.glob to mock finding paths with kytos.json file."""
            self.assertEqual(args, "*/*/kytos.json")

            mock_path1 = Mock()
            mock_path1.parts = ['kytos', 'of_core', 'kytos.json']
            return [mock_path1]
        mock_path.glob = glob_side_effect

        mock_prop_enabled = PropertyMock()
        with patch.object(NAppsManager, '_enabled', mock_prop_enabled):
            mock_prop_enabled.return_value = mock_path

            get_return = self.napps_manager.get_enabled_local()

            self.assertEqual(get_return[0][0], 'kytos')
            self.assertEqual(get_return[0][1], 'of_core')
            self.assertEqual(mock_prop_enabled.call_count, 1)

    def test_get_installed_local(self):

        """Test get_installed_local used to find
        installed napps in local machine"""
        # Mock kytos.json path
        mock_path = Mock()

        def glob_side_effect(args):
            """Path.glob to mock finding paths with kytos.json file."""
            self.assertEqual(args, "*/*/kytos.json")

            mock_path1 = Mock()
            mock_path1.parts = ['kytos', 'of_core', 'kytos.json']
            return [mock_path1]
        mock_path.glob = glob_side_effect

        mock_prop_installed = PropertyMock()
        with patch.object(NAppsManager, '_installed', mock_prop_installed):
            mock_prop_installed.return_value = mock_path

            get_return = self.napps_manager.get_installed_local()

            self.assertEqual(get_return[0][0], 'kytos')
            self.assertEqual(get_return[0][1], 'of_core')
            self.assertEqual(mock_prop_installed.call_count, 1)

    @patch('urllib.request.urlopen')
    def test_get_installed(self, mock_urlopen):
        """Test method get_installed to find all installed napps."""
        mock_urlopen.return_value = self.get_napps_response_mock()

        installed_napps = self.napps_manager.get_installed()

        self.assertEqual(len(installed_napps), 2)
        self.assertEqual(installed_napps[0], ("kytos", "mef_eline"))
        self.assertEqual(installed_napps[1], ("kytos", "of_lldp"))

    @patch('urllib.request.urlopen')
    def test_get_installed__error(self, mock_urlopen):
        """Test method get_installed with API error."""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 500
        mock_urlopen.return_value = mock_response

        with self.assertRaises(KytosException) as context:
            self.napps_manager.get_installed()

        self.assertEqual('Error calling Kytos to check installed NApps.',
                         str(context.exception))

    @patch('urllib.request.urlopen')
    def test_get_enabled(self, mock_urlopen):
        """Test method get_enabled to find all enabled napps."""
        mock_urlopen.return_value = self.get_napps_response_mock()

        installed_napps = self.napps_manager.get_enabled()

        self.assertEqual(len(installed_napps), 2)
        self.assertEqual(installed_napps[0], ("kytos", "mef_eline"))
        self.assertEqual(installed_napps[1], ("kytos", "of_lldp"))

    @patch('urllib.request.urlopen')
    def test_get_enabled__error(self, mock_urlopen):
        """Test method get_enabled with API error."""
        mock_response = MagicMock()
        mock_response.getcode.return_value = 500
        mock_urlopen.return_value = mock_response

        with self.assertRaises(KytosException) as context:
            self.napps_manager.get_enabled()

        self.assertEqual('Error calling Kytos to check enabled NApps.',
                         str(context.exception))

    @patch('urllib.request.urlopen')
    def test_is_enabled(self, mock_urlopen):
        """Test is_enabled method."""
        mock_urlopen.return_value = self.get_napps_response_mock()

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.assertTrue(self.napps_manager.is_enabled())

    @patch('urllib.request.urlopen')
    def test_is_installed(self, mock_urlopen):
        """Test is_installed method."""
        mock_urlopen.return_value = self.get_napps_response_mock()

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.assertTrue(self.napps_manager.is_installed())

    @patch('urllib.request.urlopen')
    def test_get_disabled(self, mock_urlopen):
        """Test get_disabled method."""
        enabled = [["kytos", "mef_eline"]]
        mock_urlopen.side_effect = [self.get_napps_response_mock(),
                                    self.get_napps_response_mock(enabled)]

        disabled = self.napps_manager.get_disabled()

        self.assertEqual(disabled, [('kytos', 'of_lldp')])

    @patch('urllib.request.urlopen')
    def test_dependencies(self, mock_urlopen):
        """Test dependencies method."""
        napps = {"napp_dependencies": ["kytos/mef_eline", "kytos/of_lldp"]}
        data = MagicMock()
        data.read.return_value = json.dumps(napps)
        mock_urlopen.return_value = data

        dependencies = self.napps_manager.dependencies()

        expected_dependencies = [('kytos', 'mef_eline'), ('kytos', 'of_lldp')]
        self.assertEqual(dependencies, expected_dependencies)

    @patch('urllib.request.urlopen')
    def test_get_description(self, mock_urlopen):
        """Test get_description method."""
        data = MagicMock()
        data.read.return_value = '{"description": "ABC"}'
        mock_urlopen.return_value = data

        description = self.napps_manager.get_description()

        self.assertEqual(description, 'ABC')

    @patch('urllib.request.urlopen')
    def test_get_version(self, mock_urlopen):
        """Test get_version method."""
        data = MagicMock()
        data.read.return_value = '{"version": "123"}'
        mock_urlopen.return_value = data

        version = self.napps_manager.get_version()

        self.assertEqual(version, '123')

    @patch('urllib.request.urlopen')
    def test_get_napp_key(self, mock_urlopen):
        """Test _get_napp_key method."""
        data = MagicMock()
        data.read.return_value = '{"key": "ABC"}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        meta_key = self.napps_manager._get_napp_key('key')

        self.assertEqual(meta_key, 'ABC')

    @patch('urllib.request.urlopen')
    def test_disable(self, mock_urlopen):
        """Test disable method."""
        data = MagicMock()
        data.read.return_value = '{}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.napps_manager.disable()

        uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_DISABLE
        uri = uri.format('kytos', 'mef_eline')

        mock_urlopen.assert_called_with(uri)

    @patch('urllib.request.urlopen')
    def test_enable(self, mock_urlopen):
        """Test enable method."""
        data = MagicMock()
        data.read.return_value = '{}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.napps_manager.enable()

        uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_ENABLE
        uri = uri.format('kytos', 'mef_eline')

        mock_urlopen.assert_called_with(uri)

    @patch('urllib.request.urlopen')
    def test_enabled_dir(self, mock_urlopen):
        """Test enabled_dir method."""
        data = MagicMock()
        data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        enabled_dir = self.napps_manager.enabled_dir()
        self.assertEqual(str(enabled_dir), 'ABC/kytos/mef_eline')

    @patch('urllib.request.urlopen')
    def test_installed_dir(self, mock_urlopen):
        """Test installed_dir method."""
        data = MagicMock()
        data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        installed_dir = self.napps_manager.installed_dir()
        self.assertEqual(str(installed_dir), 'DEF/kytos/mef_eline')

    @patch('urllib.request.urlopen')
    def test_remote_uninstall(self, mock_urlopen):
        """Test remote_uninstall method."""
        data = MagicMock()
        data.read.return_value = '{}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.napps_manager.remote_uninstall()

        uninstall_uri = self.napps_manager._NAPP_UNINSTALL
        uri = self.napps_manager._kytos_api + uninstall_uri
        uri = uri.format('kytos', 'mef_eline')

        mock_urlopen.assert_called_with(uri)

    @patch('urllib.request.urlopen')
    def test_remote_install(self, mock_urlopen):
        """Test remote_install method."""
        data = MagicMock()
        data.read.return_value = '{}'
        mock_urlopen.return_value = data

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.napps_manager.remote_install()

        install_uri = self.napps_manager._NAPP_INSTALL
        uri = self.napps_manager._kytos_api + install_uri
        uri = uri.format('kytos', 'mef_eline')

        mock_urlopen.assert_called_with(uri)

    def test_valid_name(self):
        """Test valid_name method."""
        valid_name = self.napps_manager.valid_name('username')
        invalid_name = self.napps_manager.valid_name('_username')

        self.assertTrue(valid_name)
        self.assertFalse(invalid_name)

    @patch('jinja2.Environment.get_template')
    def test_render_template(self, mock_get_template):
        """Test render_template method."""
        template = MagicMock()
        mock_get_template.return_value = template

        self.napps_manager.render_template('', 'filename', 'context')

        mock_get_template.assert_called_with('filename')
        template.render.assert_called_with('context')

    @patch('kytos.utils.napps.NAppsClient')
    def test_search(self, mock_napps_client):
        """Test search method."""
        napp_1 = {'username': 'kytos', 'name': 'mef_eline', 'description': '',
                  'tags': ['A', 'B']}
        napp_2 = {'username': '0_kytos', 'name': 'any', 'description': '',
                  'tags': ['A', 'B']}
        napps_client = MagicMock()
        napps_client.get_napps.return_value = [napp_1, napp_2]
        mock_napps_client.return_value = napps_client

        # pattern to match strings that start with letters
        pattern = re.compile('^[a-z]+')
        napps = self.napps_manager.search(pattern)

        self.assertEqual(napps, [napp_1])

    @patch('os.makedirs')
    @patch('builtins.open')
    @patch('builtins.input')
    @patch('kytos.utils.napps.NAppsManager.render_template')
    def test_create_napp(self, *args):
        """Test create_napp method."""
        (mock_render_template, mock_input, _, mock_mkdirs) = args
        mock_input.side_effect = ['username', 'napp', None]

        self.napps_manager.create_napp()

        tmpl_path = SKEL_PATH / 'napp-structure/username/napp'
        description = '# TODO: <<<< Insert your NApp description here >>>>'
        context = {'username': 'username', 'napp': 'napp',
                   'description': description}

        calls = []
        for tmp in ['__init__.py', 'main.py', '.gitignore', 'kytos.json',
                    'README.rst', 'settings.py']:
            calls.append(call(tmpl_path, '{}.template'.format(tmp), context))
        calls.append(call('{}/ui'.format(tmpl_path), 'README.rst.template',
                          context))

        mock_mkdirs.assert_has_calls([call('username', exist_ok=True),
                                      call('username/napp'),
                                      call('username/napp/ui/k-info-panel'),
                                      call('username/napp/ui/k-toolbar'),
                                      call('username/napp/ui/k-action-menu')])
        mock_render_template.assert_has_calls(calls, any_order=True)

    def test_check_module(self):
        """Test _check_module method."""
        folder = MagicMock()
        folder.exists.return_value = False

        self.napps_manager._check_module(folder)

        folder.mkdir.assert_called()
        (folder / '__init__.py').touch.assert_called()

    @patch('pathspec.pathspec.PathSpec.match_tree')
    @patch('tarfile.TarFile.add')
    @patch('os.remove')
    @patch('os.walk')
    @patch('os.getcwd')
    @patch('builtins.open')
    def test_build_napp_package(self, *args):
        """Test build_napp_package method."""
        (_, mock_getcwd, mock_walk, _, mock_add, mock_match_tree) = args
        with tempfile.TemporaryDirectory() as tmp_dir:
            mock_getcwd.return_value = tmp_dir

            files = ['username/napp/A', 'username/napp/B', 'username/napp/C']
            mock_walk.return_value = [(tmp_dir, ['username/napp/.git'], files)]

            mock_match_tree.return_value = ['username/napp/C']

            self.napps_manager.build_napp_package('username/napp')

            calls = [call(PurePosixPath('username/napp/A')),
                     call(PurePosixPath('username/napp/B'))]
            mock_add.assert_has_calls(calls)

    @patch('ruamel.yaml.YAML.load', return_value='openapi')
    @patch('pathlib.Path.open')
    @patch('builtins.open')
    def test_create_metadata(self, *args):
        """Test create_metadata method."""
        (mock_open, _, _) = args
        enter_file_1 = MagicMock()
        enter_file_1.read.return_value = '{}'

        enter_file_2 = MagicMock()
        enter_file_2.read.return_value = 'readme'

        mock_open.return_value.__enter__.side_effect = [enter_file_1,
                                                        enter_file_2]

        metadata = self.napps_manager.create_metadata()

        self.assertEqual(metadata, {'readme': 'readme',
                                    'OpenAPI_Spec': '"openapi"'})

    @patch('kytos.utils.napps.NAppsClient')
    @patch('kytos.utils.napps.NAppsManager.build_napp_package')
    @patch('kytos.utils.napps.NAppsManager.create_metadata')
    @patch('kytos.utils.napps.NAppsManager.prepare')
    def test_upload(self, *args):
        """Test upload method."""
        (mock_prepare, mock_create, mock_build, mock_napps_client) = args
        mock_create.return_value = {'name': 'ABC'}
        mock_build.return_value = 'package'
        napps_client = MagicMock()
        mock_napps_client.return_value = napps_client

        self.napps_manager.upload()

        mock_prepare.assert_called()
        mock_create.assert_called()
        mock_build.assert_called_with('ABC')
        napps_client.upload_napp.assert_called_with({'name': 'ABC'}, 'package')

    @patch('kytos.utils.napps.NAppsClient')
    def test_delete(self, mock_napps_client):
        """Test delete method."""
        napps_client = MagicMock()
        mock_napps_client.return_value = napps_client

        self.napps_manager.user = 'kytos'
        self.napps_manager.napp = 'mef_eline'

        self.napps_manager.delete()

        napps_client.delete.assert_called_with('kytos', 'mef_eline')

    @patch('sys.exit')
    @patch('kytos.utils.napps.OpenAPI')
    @patch('kytos.utils.napps.NAppsManager._ask_openapi', return_value=True)
    def test_prepare(self, *args):
        """Test prepare method."""
        (_, mock_openapi, _) = args
        self.napps_manager.prepare()

        napp_path = Path()
        tpl_path = SKEL_PATH / 'napp-structure/username/napp'
        mock_openapi.assert_called_with(napp_path, tpl_path)
        mock_openapi.return_value.render_template.assert_called()

    @patch('pathlib.Path.exists')
    @patch('builtins.input')
    def test_ask_openapi(self, *args):
        """Test _ask_openapi method."""
        (mock_input, mock_exists) = args
        mock_input.side_effect = ['', '', 'yes', 'no']
        mock_exists.side_effect = [True, False, False, False]

        for expected in [False, True, True, False]:
            response = self.napps_manager._ask_openapi()
            self.assertEqual(response, expected)

    @patch('kytos.utils.napps.NAppsClient')
    def test_reload(self, mock_napps_client):
        """Test reload method."""
        napps_client = MagicMock()
        mock_napps_client.return_value = napps_client

        napps = []
        self.napps_manager.reload(napps)

        napps_client.reload_napps.assert_called_with(napps)


1		"""kytos.utils.napps.NAppsManager tests."""
2		import json
3		import re
4		import tempfile
5		import unittest
6		from pathlib import Path, PurePosixPath
7		from unittest.mock import MagicMock, Mock, PropertyMock, call, patch
8
9		from kytos.utils.exceptions import KytosException
10		from kytos.utils.napps import NAppsManager
11		from kytos.utils.settings import SKEL_PATH
12
13
14		# pylint: disable=protected-access, too-many-public-methods
15		class TestNapps(unittest.TestCase):
16		"""Test the class NAppsManager."""
17
18		def setUp(self):
19		"""Execute steps before each tests."""
20		self.napps_manager = NAppsManager()
21
22		@staticmethod
23		def get_napps_response_mock(napps=None):
24		"""Get mock to napps response."""
25		if napps is None:
26		napps = [["kytos", "mef_eline"], ["kytos", "of_lldp"]]
27		response = json.dumps({"napps": napps})
28
29		mock_response = MagicMock()
30		mock_response.getcode.return_value = 200
31		mock_response.read.return_value = response
32		mock_response.__enter__.return_value = mock_response
33		return mock_response
34
35		@patch('urllib.request.urlopen')
36		def test_enabled_property(self, mock_urlopen):
37		"""Test enabled property."""
38		data = MagicMock()
39		data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
40		mock_urlopen.return_value = data
41
42		self.assertEqual(str(self.napps_manager._enabled), 'ABC')
43
44		def test_enabled_property__error(self):
45		"""Test enabled property to error case."""
46
47		with self.assertRaises(SystemExit):
48		# pylint: disable=pointless-statement
49		self.napps_manager._enabled
50		# pylint: enable=pointless-statement
51
52		self.assertIsNone(self.napps_manager._NAppsManager__local_enabled)
53
54		@patch('urllib.request.urlopen')
55		def test_installed_property(self, mock_urlopen):
56		"""Test installed property."""
57		data = MagicMock()
58		data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
59		mock_urlopen.return_value = data
60
61		self.assertEqual(str(self.napps_manager._installed), 'DEF')
62
63		def test_installed_property__error(self):
64		"""Test installed property to error case."""
65
66		with self.assertRaises(SystemExit):
67		# pylint: disable=pointless-statement
68		self.napps_manager._installed
69		# pylint: enable=pointless-statement
70
71		self.assertIsNone(self.napps_manager._NAppsManager__local_installed)
72
73		def test_napp_id_property(self):
74		"""Test napp_id property."""
75		self.napps_manager.user = 'user'
76		self.napps_manager.napp = 'napp'
77
78		self.assertEqual(self.napps_manager.napp_id, 'user/napp')
79
80		def test_set_napp(self):
81		"""Test set_napp method."""
82		self.napps_manager.set_napp('user', 'napp', 'version')
83
84		self.assertEqual(self.napps_manager.user, 'user')
85		self.assertEqual(self.napps_manager.napp, 'napp')
86		self.assertEqual(self.napps_manager.version, 'version')
87
88		def test_get_napps(self):
89		"""Test method get_napps used to find
90		enabled and installed napps.
91		"""
92		mock_path = Mock()
93
94		def glob_side_effect(args):
95		"""Path.glob to mock finding paths with kytos.json file."""
96		self.assertEqual(args, "//kytos.json")
97
98		mock_path1 = Mock()
99		mock_path1.parts = ['kytos', 'of_core', 'kytos.json']
100
101		mock_path2 = Mock()
102		mock_path2.parts = ['kytos', 'of_lldp', 'kytos.json']
103
104		return [mock_path1, mock_path2]
105
106		mock_path.glob = glob_side_effect
107
108		# pylint: disable=protected-access
109		get_return = self.napps_manager._get_napps(mock_path)
110		self.assertEqual(get_return[0][0], 'kytos')
111		self.assertEqual(get_return[0][1], 'of_core')
112		self.assertEqual(get_return[1][0], 'kytos')
113		self.assertEqual(get_return[1][1], 'of_lldp')
114
115	View Code Duplication	def test_get_enabled_local(self):
		0 ignored issues – show Duplication introduced 2020-06-09 13:15 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
116		"""Test get_enabled_local used to find
117		enabled napps in local machine"""
118		# Mock kytos.json path
119		mock_path = Mock()
120
121		def glob_side_effect(args):
122		"""Path.glob to mock finding paths with kytos.json file."""
123		self.assertEqual(args, "//kytos.json")
124
125		mock_path1 = Mock()
126		mock_path1.parts = ['kytos', 'of_core', 'kytos.json']
127		return [mock_path1]
128		mock_path.glob = glob_side_effect
129
130		mock_prop_enabled = PropertyMock()
131		with patch.object(NAppsManager, '_enabled', mock_prop_enabled):
132		mock_prop_enabled.return_value = mock_path
133
134		get_return = self.napps_manager.get_enabled_local()
135
136		self.assertEqual(get_return[0][0], 'kytos')
137		self.assertEqual(get_return[0][1], 'of_core')
138		self.assertEqual(mock_prop_enabled.call_count, 1)
139
140	View Code Duplication	def test_get_installed_local(self):
		0 ignored issues – show Duplication introduced 2020-06-09 13:15 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
141		"""Test get_installed_local used to find
142		installed napps in local machine"""
143		# Mock kytos.json path
144		mock_path = Mock()
145
146		def glob_side_effect(args):
147		"""Path.glob to mock finding paths with kytos.json file."""
148		self.assertEqual(args, "//kytos.json")
149
150		mock_path1 = Mock()
151		mock_path1.parts = ['kytos', 'of_core', 'kytos.json']
152		return [mock_path1]
153		mock_path.glob = glob_side_effect
154
155		mock_prop_installed = PropertyMock()
156		with patch.object(NAppsManager, '_installed', mock_prop_installed):
157		mock_prop_installed.return_value = mock_path
158
159		get_return = self.napps_manager.get_installed_local()
160
161		self.assertEqual(get_return[0][0], 'kytos')
162		self.assertEqual(get_return[0][1], 'of_core')
163		self.assertEqual(mock_prop_installed.call_count, 1)
164
165		@patch('urllib.request.urlopen')
166		def test_get_installed(self, mock_urlopen):
167		"""Test method get_installed to find all installed napps."""
168		mock_urlopen.return_value = self.get_napps_response_mock()
169
170		installed_napps = self.napps_manager.get_installed()
171
172		self.assertEqual(len(installed_napps), 2)
173		self.assertEqual(installed_napps[0], ("kytos", "mef_eline"))
174		self.assertEqual(installed_napps[1], ("kytos", "of_lldp"))
175
176		@patch('urllib.request.urlopen')
177		def test_get_installed__error(self, mock_urlopen):
178		"""Test method get_installed with API error."""
179		mock_response = MagicMock()
180		mock_response.getcode.return_value = 500
181		mock_urlopen.return_value = mock_response
182
183		with self.assertRaises(KytosException) as context:
184		self.napps_manager.get_installed()
185
186		self.assertEqual('Error calling Kytos to check installed NApps.',
187		str(context.exception))
188
189		@patch('urllib.request.urlopen')
190		def test_get_enabled(self, mock_urlopen):
191		"""Test method get_enabled to find all enabled napps."""
192		mock_urlopen.return_value = self.get_napps_response_mock()
193
194		installed_napps = self.napps_manager.get_enabled()
195
196		self.assertEqual(len(installed_napps), 2)
197		self.assertEqual(installed_napps[0], ("kytos", "mef_eline"))
198		self.assertEqual(installed_napps[1], ("kytos", "of_lldp"))
199
200		@patch('urllib.request.urlopen')
201		def test_get_enabled__error(self, mock_urlopen):
202		"""Test method get_enabled with API error."""
203		mock_response = MagicMock()
204		mock_response.getcode.return_value = 500
205		mock_urlopen.return_value = mock_response
206
207		with self.assertRaises(KytosException) as context:
208		self.napps_manager.get_enabled()
209
210		self.assertEqual('Error calling Kytos to check enabled NApps.',
211		str(context.exception))
212
213		@patch('urllib.request.urlopen')
214		def test_is_enabled(self, mock_urlopen):
215		"""Test is_enabled method."""
216		mock_urlopen.return_value = self.get_napps_response_mock()
217
218		self.napps_manager.user = 'kytos'
219		self.napps_manager.napp = 'mef_eline'
220
221		self.assertTrue(self.napps_manager.is_enabled())
222
223		@patch('urllib.request.urlopen')
224		def test_is_installed(self, mock_urlopen):
225		"""Test is_installed method."""
226		mock_urlopen.return_value = self.get_napps_response_mock()
227
228		self.napps_manager.user = 'kytos'
229		self.napps_manager.napp = 'mef_eline'
230
231		self.assertTrue(self.napps_manager.is_installed())
232
233		@patch('urllib.request.urlopen')
234		def test_get_disabled(self, mock_urlopen):
235		"""Test get_disabled method."""
236		enabled = [["kytos", "mef_eline"]]
237		mock_urlopen.side_effect = [self.get_napps_response_mock(),
238		self.get_napps_response_mock(enabled)]
239
240		disabled = self.napps_manager.get_disabled()
241
242		self.assertEqual(disabled, [('kytos', 'of_lldp')])
243
244		@patch('urllib.request.urlopen')
245		def test_dependencies(self, mock_urlopen):
246		"""Test dependencies method."""
247		napps = {"napp_dependencies": ["kytos/mef_eline", "kytos/of_lldp"]}
248		data = MagicMock()
249		data.read.return_value = json.dumps(napps)
250		mock_urlopen.return_value = data
251
252		dependencies = self.napps_manager.dependencies()
253
254		expected_dependencies = [('kytos', 'mef_eline'), ('kytos', 'of_lldp')]
255		self.assertEqual(dependencies, expected_dependencies)
256
257		@patch('urllib.request.urlopen')
258		def test_get_description(self, mock_urlopen):
259		"""Test get_description method."""
260		data = MagicMock()
261		data.read.return_value = '{"description": "ABC"}'
262		mock_urlopen.return_value = data
263
264		description = self.napps_manager.get_description()
265
266		self.assertEqual(description, 'ABC')
267
268		@patch('urllib.request.urlopen')
269		def test_get_version(self, mock_urlopen):
270		"""Test get_version method."""
271		data = MagicMock()
272		data.read.return_value = '{"version": "123"}'
273		mock_urlopen.return_value = data
274
275		version = self.napps_manager.get_version()
276
277		self.assertEqual(version, '123')
278
279		@patch('urllib.request.urlopen')
280		def test_get_napp_key(self, mock_urlopen):
281		"""Test _get_napp_key method."""
282		data = MagicMock()
283		data.read.return_value = '{"key": "ABC"}'
284		mock_urlopen.return_value = data
285
286		self.napps_manager.user = 'kytos'
287		self.napps_manager.napp = 'mef_eline'
288
289		meta_key = self.napps_manager._get_napp_key('key')
290
291		self.assertEqual(meta_key, 'ABC')
292
293		@patch('urllib.request.urlopen')
294		def test_disable(self, mock_urlopen):
295		"""Test disable method."""
296		data = MagicMock()
297		data.read.return_value = '{}'
298		mock_urlopen.return_value = data
299
300		self.napps_manager.user = 'kytos'
301		self.napps_manager.napp = 'mef_eline'
302
303		self.napps_manager.disable()
304
305		uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_DISABLE
306		uri = uri.format('kytos', 'mef_eline')
307
308		mock_urlopen.assert_called_with(uri)
309
310		@patch('urllib.request.urlopen')
311		def test_enable(self, mock_urlopen):
312		"""Test enable method."""
313		data = MagicMock()
314		data.read.return_value = '{}'
315		mock_urlopen.return_value = data
316
317		self.napps_manager.user = 'kytos'
318		self.napps_manager.napp = 'mef_eline'
319
320		self.napps_manager.enable()
321
322		uri = self.napps_manager._kytos_api + self.napps_manager._NAPP_ENABLE
323		uri = uri.format('kytos', 'mef_eline')
324
325		mock_urlopen.assert_called_with(uri)
326
327		@patch('urllib.request.urlopen')
328		def test_enabled_dir(self, mock_urlopen):
329		"""Test enabled_dir method."""
330		data = MagicMock()
331		data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
332		mock_urlopen.return_value = data
333
334		self.napps_manager.user = 'kytos'
335		self.napps_manager.napp = 'mef_eline'
336
337		enabled_dir = self.napps_manager.enabled_dir()
338		self.assertEqual(str(enabled_dir), 'ABC/kytos/mef_eline')
339
340		@patch('urllib.request.urlopen')
341		def test_installed_dir(self, mock_urlopen):
342		"""Test installed_dir method."""
343		data = MagicMock()
344		data.read.return_value = '{"napps": "ABC", "installed_napps": "DEF"}'
345		mock_urlopen.return_value = data
346
347		self.napps_manager.user = 'kytos'
348		self.napps_manager.napp = 'mef_eline'
349
350		installed_dir = self.napps_manager.installed_dir()
351		self.assertEqual(str(installed_dir), 'DEF/kytos/mef_eline')
352
353		@patch('urllib.request.urlopen')
354		def test_remote_uninstall(self, mock_urlopen):
355		"""Test remote_uninstall method."""
356		data = MagicMock()
357		data.read.return_value = '{}'
358		mock_urlopen.return_value = data
359
360		self.napps_manager.user = 'kytos'
361		self.napps_manager.napp = 'mef_eline'
362
363		self.napps_manager.remote_uninstall()
364
365		uninstall_uri = self.napps_manager._NAPP_UNINSTALL
366		uri = self.napps_manager._kytos_api + uninstall_uri
367		uri = uri.format('kytos', 'mef_eline')
368
369		mock_urlopen.assert_called_with(uri)
370
371		@patch('urllib.request.urlopen')
372		def test_remote_install(self, mock_urlopen):
373		"""Test remote_install method."""
374		data = MagicMock()
375		data.read.return_value = '{}'
376		mock_urlopen.return_value = data
377
378		self.napps_manager.user = 'kytos'
379		self.napps_manager.napp = 'mef_eline'
380
381		self.napps_manager.remote_install()
382
383		install_uri = self.napps_manager._NAPP_INSTALL
384		uri = self.napps_manager._kytos_api + install_uri
385		uri = uri.format('kytos', 'mef_eline')
386
387		mock_urlopen.assert_called_with(uri)
388
389		def test_valid_name(self):
390		"""Test valid_name method."""
391		valid_name = self.napps_manager.valid_name('username')
392		invalid_name = self.napps_manager.valid_name('_username')
393
394		self.assertTrue(valid_name)
395		self.assertFalse(invalid_name)
396
397		@patch('jinja2.Environment.get_template')
398		def test_render_template(self, mock_get_template):
399		"""Test render_template method."""
400		template = MagicMock()
401		mock_get_template.return_value = template
402
403		self.napps_manager.render_template('', 'filename', 'context')
404
405		mock_get_template.assert_called_with('filename')
406		template.render.assert_called_with('context')
407
408		@patch('kytos.utils.napps.NAppsClient')
409		def test_search(self, mock_napps_client):
410		"""Test search method."""
411		napp_1 = {'username': 'kytos', 'name': 'mef_eline', 'description': '',
412		'tags': ['A', 'B']}
413		napp_2 = {'username': '0_kytos', 'name': 'any', 'description': '',
414		'tags': ['A', 'B']}
415		napps_client = MagicMock()
416		napps_client.get_napps.return_value = [napp_1, napp_2]
417		mock_napps_client.return_value = napps_client
418
419		# pattern to match strings that start with letters
420		pattern = re.compile('^[a-z]+')
421		napps = self.napps_manager.search(pattern)
422
423		self.assertEqual(napps, [napp_1])
424
425		@patch('os.makedirs')
426		@patch('builtins.open')
427		@patch('builtins.input')
428		@patch('kytos.utils.napps.NAppsManager.render_template')
429		def test_create_napp(self, *args):
430		"""Test create_napp method."""
431		(mock_render_template, mock_input, _, mock_mkdirs) = args
432		mock_input.side_effect = ['username', 'napp', None]
433
434		self.napps_manager.create_napp()
435
436		tmpl_path = SKEL_PATH / 'napp-structure/username/napp'
437		description = '# TODO: <<<< Insert your NApp description here >>>>'
438		context = {'username': 'username', 'napp': 'napp',
439		'description': description}
440
441		calls = []
442		for tmp in ['__init__.py', 'main.py', '.gitignore', 'kytos.json',
443		'README.rst', 'settings.py']:
444		calls.append(call(tmpl_path, '{}.template'.format(tmp), context))
445		calls.append(call('{}/ui'.format(tmpl_path), 'README.rst.template',
446		context))
447
448		mock_mkdirs.assert_has_calls([call('username', exist_ok=True),
449		call('username/napp'),
450		call('username/napp/ui/k-info-panel'),
451		call('username/napp/ui/k-toolbar'),
452		call('username/napp/ui/k-action-menu')])
453		mock_render_template.assert_has_calls(calls, any_order=True)
454
455		def test_check_module(self):
456		"""Test _check_module method."""
457		folder = MagicMock()
458		folder.exists.return_value = False
459
460		self.napps_manager._check_module(folder)
461
462		folder.mkdir.assert_called()
463		(folder / '__init__.py').touch.assert_called()
464
465		@patch('pathspec.pathspec.PathSpec.match_tree')
466		@patch('tarfile.TarFile.add')
467		@patch('os.remove')
468		@patch('os.walk')
469		@patch('os.getcwd')
470		@patch('builtins.open')
471		def test_build_napp_package(self, *args):
472		"""Test build_napp_package method."""
473		(_, mock_getcwd, mock_walk, _, mock_add, mock_match_tree) = args
474		with tempfile.TemporaryDirectory() as tmp_dir:
475		mock_getcwd.return_value = tmp_dir
476
477		files = ['username/napp/A', 'username/napp/B', 'username/napp/C']
478		mock_walk.return_value = [(tmp_dir, ['username/napp/.git'], files)]
479
480		mock_match_tree.return_value = ['username/napp/C']
481
482		self.napps_manager.build_napp_package('username/napp')
483
484		calls = [call(PurePosixPath('username/napp/A')),
485		call(PurePosixPath('username/napp/B'))]
486		mock_add.assert_has_calls(calls)
487
488		@patch('ruamel.yaml.YAML.load', return_value='openapi')
489		@patch('pathlib.Path.open')
490		@patch('builtins.open')
491		def test_create_metadata(self, *args):
492		"""Test create_metadata method."""
493		(mock_open, _, _) = args
494		enter_file_1 = MagicMock()
495		enter_file_1.read.return_value = '{}'
496
497		enter_file_2 = MagicMock()
498		enter_file_2.read.return_value = 'readme'
499
500		mock_open.return_value.__enter__.side_effect = [enter_file_1,
501		enter_file_2]
502
503		metadata = self.napps_manager.create_metadata()
504
505		self.assertEqual(metadata, {'readme': 'readme',
506		'OpenAPI_Spec': '"openapi"'})
507
508		@patch('kytos.utils.napps.NAppsClient')
509		@patch('kytos.utils.napps.NAppsManager.build_napp_package')
510		@patch('kytos.utils.napps.NAppsManager.create_metadata')
511		@patch('kytos.utils.napps.NAppsManager.prepare')
512		def test_upload(self, *args):
513		"""Test upload method."""
514		(mock_prepare, mock_create, mock_build, mock_napps_client) = args
515		mock_create.return_value = {'name': 'ABC'}
516		mock_build.return_value = 'package'
517		napps_client = MagicMock()
518		mock_napps_client.return_value = napps_client
519
520		self.napps_manager.upload()
521
522		mock_prepare.assert_called()
523		mock_create.assert_called()
524		mock_build.assert_called_with('ABC')
525		napps_client.upload_napp.assert_called_with({'name': 'ABC'}, 'package')
526
527		@patch('kytos.utils.napps.NAppsClient')
528		def test_delete(self, mock_napps_client):
529		"""Test delete method."""
530		napps_client = MagicMock()
531		mock_napps_client.return_value = napps_client
532
533		self.napps_manager.user = 'kytos'
534		self.napps_manager.napp = 'mef_eline'
535
536		self.napps_manager.delete()
537
538		napps_client.delete.assert_called_with('kytos', 'mef_eline')
539
540		@patch('sys.exit')
541		@patch('kytos.utils.napps.OpenAPI')
542		@patch('kytos.utils.napps.NAppsManager._ask_openapi', return_value=True)
543		def test_prepare(self, *args):
544		"""Test prepare method."""
545		(_, mock_openapi, _) = args
546		self.napps_manager.prepare()
547
548		napp_path = Path()
549		tpl_path = SKEL_PATH / 'napp-structure/username/napp'
550		mock_openapi.assert_called_with(napp_path, tpl_path)
551		mock_openapi.return_value.render_template.assert_called()
552
553		@patch('pathlib.Path.exists')
554		@patch('builtins.input')
555		def test_ask_openapi(self, *args):
556		"""Test _ask_openapi method."""
557		(mock_input, mock_exists) = args
558		mock_input.side_effect = ['', '', 'yes', 'no']
559		mock_exists.side_effect = [True, False, False, False]
560
561		for expected in [False, True, True, False]:
562		response = self.napps_manager._ask_openapi()
563		self.assertEqual(response, expected)
564
565		@patch('kytos.utils.napps.NAppsClient')
566		def test_reload(self, mock_napps_client):
567		"""Test reload method."""
568		napps_client = MagicMock()
569		mock_napps_client.return_value = napps_client
570
571		napps = []
572		self.napps_manager.reload(napps)
573
574		napps_client.reload_napps.assert_called_with(napps)
575

kytos / kytos-utils

Pull Request — master (#277)

TestNapps.test_installed_dir() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like