tests.unit.test_napps

Complexity

Total Complexity

39

Size/Duplication

Total Lines	434
Duplicated Lines	11.06 %

Importance

Changes

0

31 Methods

Rating	Name	Duplication	Size	Complexity
A	TestNapps.setUp()	0	3	1
A	TestNapps.test_is_installed()	0	9	1
A	TestNapps.test_get_enabled__error()	0	12	2
A	TestNapps.test_get_version()	0	10	1
A	TestNapps.test_get_enabled()	0	10	1
A	TestNapps.test_installed_property__error()	0	9	2
A	TestNapps.test_get_disabled()	0	10	1
A	TestNapps.test_get_installed()	0	10	1
A	TestNapps.test_remote_uninstall()	0	17	1
A	TestNapps.test_get_installed__error()	0	12	2
A	TestNapps.test_remote_install()	0	17	1
A	TestNapps.test_is_enabled()	0	9	1
A	TestNapps.test_get_napp_key()	0	13	1
A	TestNapps.test_installed_dir()	0	12	1
A	TestNapps.test_valid_name()	0	7	1
A	TestNapps.test_disable()	0	16	1
A	TestNapps.test_set_napp()	0	7	1
A	TestNapps.test_dependencies()	0	12	1
A	TestNapps.test_create_napp()	0	29	2
A	TestNapps.test_get_enabled_local()	24	24	2
A	TestNapps.test_installed_property()	0	8	1
A	TestNapps.test_enable()	0	16	1
A	TestNapps.test_enabled_dir()	0	12	1
A	TestNapps.test_get_installed_local()	24	24	2
A	TestNapps.get_napps_response_mock()	0	12	2
A	TestNapps.test_napp_id_property()	0	6	1
A	TestNapps.test_get_napps()	0	26	1
A	TestNapps.test_get_description()	0	10	1
A	TestNapps.test_enabled_property()	0	8	1
A	TestNapps.test_render_template()	0	10	1
A	TestNapps.test_enabled_property__error()	0	9	2

"""kytos.utils.napps.NAppsManager tests."""
import json
import unittest
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):

This code seems to be duplicated in your project.

        """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):

This code seems to be duplicated in your project.

        """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('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)