kytos.utils.napps.NAppsManager.create_napp() - Code Metrics - Inspection of "Friendly response to Keyboard Interrupt in the NAp..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#310)

unknown

created 2020-12-23 17:25 UTC

kytos.utils.napps.NAppsManager.create_napp() C

↳ Parent: kytos.utils.napps

Complexity

Conditions

Size

Total Lines	79
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	44
CRAP Score	10.1061

Importance

Changes

Metric	Value
cc	10
eloc	52
nop	2
dl	0
loc	79
ccs	44
cts	49
cp	0.898
crap	10.1061
rs	5.7709
c	0
b	0
f	0

How to fix Long Method Complexity

"""Manage Network Application files."""
import json
import logging
import os
import pathlib
import re
import sys
import tarfile
import urllib
from http import HTTPStatus

# Disable pylint import checks that conflict with isort
# pylint: disable=ungrouped-imports,wrong-import-order
import pathspec
from jinja2 import Environment, FileSystemLoader
from ruamel.yaml import YAML

from kytos.utils.client import NAppsClient
from kytos.utils.config import KytosConfig
from kytos.utils.exceptions import KytosException
from kytos.utils.openapi import OpenAPI
from kytos.utils.settings import SKEL_PATH

LOG = logging.getLogger(__name__)


# pylint: disable=too-many-instance-attributes,too-many-public-methods
class NAppsManager:
    """Deal with NApps at filesystem level and ask Kytos to (un)load NApps."""

    _NAPP_ENABLE = "api/kytos/core/napps/{}/{}/enable"
    _NAPP_DISABLE = "api/kytos/core/napps/{}/{}/disable"
    _NAPP_INSTALL = "api/kytos/core/napps/{}/{}/install"
    _NAPP_UNINSTALL = "api/kytos/core/napps/{}/{}/uninstall"
    _NAPPS_INSTALLED = "api/kytos/core/napps_installed"
    _NAPPS_ENABLED = "api/kytos/core/napps_enabled"
    _NAPP_METADATA = "api/kytos/core/napps/{}/{}/metadata/{}"

    def __init__(self):
        """Instance a new NAppsManager.

        This method do not need parameters.
        """
        self._config = KytosConfig().config
        self._kytos_api = self._config.get('kytos', 'api')

        self.user = None
        self.napp = None
        self.version = None

        # Automatically get from kytosd API when needed
        self.__local_enabled = None
        self.__local_installed = None

    @property
    def _enabled(self):
        if self.__local_enabled is None:
            self.__require_kytos_config()
        return self.__local_enabled

    @property
    def _installed(self):
        if self.__local_installed is None:
            self.__require_kytos_config()
        return self.__local_installed

    def __require_kytos_config(self):
        """Set path locations from kytosd API.

        It should not be called directly, but from properties that require a
        running kytosd instance.
        """
        if self.__local_enabled is None:
            uri = self._kytos_api + 'api/kytos/core/config/'
            try:
                ops = json.loads(urllib.request.urlopen(uri).read())
            except urllib.error.URLError as err:
                msg = f'Error connecting to Kytos daemon: {uri} {err.reason}'
                print(msg)
                sys.exit(1)
            self.__local_enabled = pathlib.Path(ops.get('napps'))
            self.__local_installed = pathlib.Path(ops.get('installed_napps'))

    def set_napp(self, user, napp, version=None):
        """Set info about NApp.

        Args:
            user (str): NApps Server username.
            napp (str): NApp name.
            version (str): NApp version.

        """
        self.user = user
        self.napp = napp
        self.version = version or 'latest'

    @property
    def napp_id(self):
        """Return a Identifier of NApp."""
        return '/'.join((self.user, self.napp))

    @staticmethod
    def _get_napps(napps_dir):
        """List of (username, napp_name) found in ``napps_dir``.

        Ex: [('kytos', 'of_core'), ('kytos', 'of_lldp')]
        """
        jsons = napps_dir.glob('*/*/kytos.json')
        return sorted(j.parts[-3:-1] for j in jsons)

    def get_enabled_local(self):
        """Sorted list of (username, napp_name) of enabled napps."""
        return self._get_napps(self._enabled)

    def get_installed_local(self):
        """Sorted list of (username, napp_name) of installed napps."""
        return self._get_napps(self._installed)

    def get_enabled(self):
        """Sorted list of (username, napp_name) of enabled napps."""
        uri = self._kytos_api + self._NAPPS_ENABLED

        try:
            response = urllib.request.urlopen(uri)
            if response.getcode() != 200:
                msg = "Error calling Kytos to check enabled NApps."
                raise KytosException(msg)

            content = json.loads(response.read())
            return sorted((c[0], c[1]) for c in content['napps'])
        except urllib.error.URLError as exception:
            LOG.error("Error checking enabled NApps. Is Kytos running?")
            raise KytosException(exception)

    def get_installed(self):
        """Sorted list of (username, napp_name) of installed napps."""
        uri = self._kytos_api + self._NAPPS_INSTALLED

        try:
            response = urllib.request.urlopen(uri)
            if response.getcode() != 200:
                msg = "Error calling Kytos to check installed NApps."
                raise KytosException(msg)

            content = json.loads(response.read())
            return sorted((c[0], c[1]) for c in content['napps'])
        except urllib.error.URLError as exception:
            LOG.error("Error checking installed NApps. Is Kytos running?")
            raise KytosException(exception)

    def is_installed(self):
        """Whether a NApp is installed."""
        return (self.user, self.napp) in self.get_installed()

    def get_disabled(self):
        """Sorted list of (username, napp_name) of disabled napps.

        The difference of installed and enabled.
        """
        installed = set(self.get_installed())
        enabled = set(self.get_enabled())
        return sorted(installed - enabled)

    def dependencies(self, user=None, napp=None):
        """Get napp_dependencies from install NApp.

        Args:
            user(string)  A Username.
            napp(string): A NApp name.
        Returns:
            napps(list): List with tuples with Username and NApp name.
                         e.g. [('kytos'/'of_core'), ('kytos/of_l2ls')]

        """
        napps = self._get_napp_key('napp_dependencies', user, napp)
        return [tuple(napp.split('/')) for napp in napps]

    def get_description(self, user=None, napp=None):
        """Return the description from kytos.json."""
        return self._get_napp_key('description', user, napp)

    def get_version(self, user=None, napp=None):
        """Return the version from kytos.json."""
        return self._get_napp_key('version', user, napp) or 'latest'

    def _get_napp_key(self, key, user=None, napp=None):
        """Return a value from kytos.json.

        Args:
            user (string): A Username.
            napp (string): A NApp name
            key (string): Key used to get the value within kytos.json.

        Returns:
            meta (object): Value stored in kytos.json.

        """
        if user is None:
            user = self.user
        if napp is None:
            napp = self.napp

        uri = self._kytos_api + self._NAPP_METADATA
        uri = uri.format(user, napp, key)

        meta = json.loads(urllib.request.urlopen(uri).read())
        return meta[key]

    def disable(self):
        """Disable a NApp if it is enabled."""
        uri = self._kytos_api + self._NAPP_DISABLE
        uri = uri.format(self.user, self.napp)

        try:
            json.loads(urllib.request.urlopen(uri).read())
        except urllib.error.HTTPError as exception:
            if exception.code == HTTPStatus.BAD_REQUEST.value:
                LOG.error("NApp is not installed. Check the NApp list.")
            else:
                LOG.error("Error disabling the NApp")

    def enable(self):
        """Enable a NApp if not already enabled."""
        uri = self._kytos_api + self._NAPP_ENABLE
        uri = uri.format(self.user, self.napp)

        try:
            json.loads(urllib.request.urlopen(uri).read())
        except urllib.error.HTTPError as exception:
            if exception.code == HTTPStatus.BAD_REQUEST.value:
                LOG.error("NApp is not installed. Check the NApp list.")
            else:
                LOG.error("Error enabling the NApp")

    def enabled_dir(self):
        """Return the enabled dir from current napp."""
        return self._enabled / self.user / self.napp

    def installed_dir(self):
        """Return the installed dir from current napp."""
        return self._installed / self.user / self.napp

    def is_enabled(self):
        """Whether a NApp is enabled."""
        return (self.user, self.napp) in self.get_enabled()

    def remote_uninstall(self):
        """Delete code inside NApp directory, if existent."""
        uri = self._kytos_api + self._NAPP_UNINSTALL
        uri = uri.format(self.user, self.napp)

        try:
            json.loads(urllib.request.urlopen(uri).read())
        except urllib.error.HTTPError as exception:
            if exception.code == HTTPStatus.BAD_REQUEST.value:
                LOG.error("Check if the NApp is installed.")
            else:
                LOG.error("Error uninstalling the NApp")

    @staticmethod
    def valid_name(username):
        """Check the validity of the given 'name'.

        The following checks are done:
        - name starts with a letter
        - name contains only letters, numbers or underscores
        """
        return username and re.match(r'[a-zA-Z][a-zA-Z0-9_]{2,}$', username)

    @staticmethod
    def render_template(templates_path, template_filename, context):
        """Render Jinja2 template for a NApp structure."""
        template_env = Environment(
            autoescape=False, trim_blocks=False,
            loader=FileSystemLoader(str(templates_path)))
        return template_env.get_template(str(template_filename)) \
            .render(context)

    @staticmethod
    def search(pattern):
        """Search all server NApps matching pattern.

        Args:
            pattern (str): Python regular expression.

        """
        def match(napp):
            """Whether a NApp metadata matches the pattern."""
            # WARNING: This will change for future versions, when 'author' will
            # be removed.
            username = napp.get('username', napp.get('author'))

            strings = ['{}/{}'.format(username, napp.get('name')),
                       napp.get('description')] + napp.get('tags')
            return any(pattern.match(string) for string in strings)

        napps = NAppsClient().get_napps()
        return [napp for napp in napps if match(napp)]

    def remote_install(self):
        """Ask kytos server to install NApp."""
        uri = self._kytos_api + self._NAPP_INSTALL
        uri = uri.format(self.user, self.napp)

        json.loads(urllib.request.urlopen(uri).read())

    @classmethod
    # pylint: disable=too-many-statements
    def create_napp(cls, meta_package=False):
        """Bootstrap a basic NApp structure for you to develop your NApp.

        This will create, on the current folder, a clean structure of a NAPP,
        filling some contents on this structure.
        """
        templates_path = SKEL_PATH / 'napp-structure/username/napp'

        ui_templates_path = os.path.join(templates_path, 'ui')

        username = None
        napp_name = None
        keyint_error = "\nUser cancelled napps creation."

        print('--------------------------------------------------------------')
        print('Welcome to the bootstrap process of your NApp.')
        print('--------------------------------------------------------------')
        print('In order to answer both the username and the NApp name,')
        print('You must follow these naming rules:')
        print(' - name starts with a letter')
        print(' - name contains only letters, numbers or underscores')
        print(' - at least three characters')
        print('--------------------------------------------------------------')
        print('')
        try:
            while not cls.valid_name(username):
                username = input('Please, insert your NApps Server username: ')
            while not cls.valid_name(napp_name):
                napp_name = input('Please, insert your NApp name: ')
                description = input('Please, insert a brief description for your '
                                    'NApp [optional]: ')
        except KeyboardInterrupt:
            print(keyint_error)
            sys.exit(0)
        if not description:

            # pylint: disable=fixme
            description = '# TODO: <<<< Insert your NApp description here >>>>'
            # pylint: enable=fixme

        context = {'username': username, 'napp': napp_name,
                   'description': description}

        #: Creating the directory structure (username/napp_name)
        os.makedirs(username, exist_ok=True)

        #: Creating ``__init__.py`` files
        with open(os.path.join(username, '__init__.py'), 'w') as init_file:
            init_file.write(f'"""NApps for the user {username}.""""')

        os.makedirs(os.path.join(username, napp_name))

        #: Creating the other files based on the templates
        templates = os.listdir(templates_path)
        templates.remove('ui')
        templates.remove('openapi.yml.template')

        if meta_package:
            templates.remove('main.py.template')
            templates.remove('settings.py.template')

        for tmp in templates:
            fname = os.path.join(username, napp_name,
                                 tmp.rsplit('.template')[0])
            with open(fname, 'w') as file:
                content = cls.render_template(templates_path, tmp, context)
                file.write(content)

        if not meta_package:
            NAppsManager.create_ui_structure(username, napp_name,
                                             ui_templates_path, context)

        print()

        print(f'Congratulations! Your NApp has been bootstrapped!\nNow  '
              f'you can go to the directory "{username}/{napp_name}" and '
              ' begin to code your NApp.')
        print('Have fun!')

    @classmethod
    def create_ui_structure(cls, username, napp_name, ui_templates_path,
                            context):
        """Create the ui directory structure."""
        for section in ['k-info-panel', 'k-toolbar', 'k-action-menu']:
            os.makedirs(os.path.join(username, napp_name, 'ui', section))

        templates = os.listdir(ui_templates_path)

        for tmp in templates:
            fname = os.path.join(username, napp_name, 'ui',
                                 tmp.rsplit('.template')[0])

            with open(fname, 'w') as file:
                content = cls.render_template(ui_templates_path, tmp,
                                              context)
                file.write(content)

    @staticmethod
    def _check_module(folder):
        """Create module folder with empty __init__.py if it doesn't exist.

        Args:
            folder (pathlib.pathlib.Path): Module path.

        """
        if not folder.exists():
            folder.mkdir(parents=True, exist_ok=True, mode=0o755)
            (folder / '__init__.py').touch()

    @staticmethod
    def build_napp_package(napp_name):
        """Build the .napp file to be sent to the napps server.

        Args:
            napp_identifier (str): Identifier formatted as
                <username>/<napp_name>

        Return:
            file_payload (binary): The binary representation of the napp
                package that will be POSTed to the napp server.

        """
        def get_matches(path):
            """Return all NApp files matching any .gitignore pattern."""
            ignored_files = [".git"]
            with open(".gitignore", 'r') as local_gitignore:
                ignored_files.extend(local_gitignore.readlines())

            user_gitignore_path = pathlib.Path("%s/.gitignore" %
                                               pathlib.Path.home())
            if user_gitignore_path.exists():
                with open(user_gitignore_path, 'r') as user_gitignore:
                    ignored_files.extend(user_gitignore.readlines())

            # Define Wildmatch pattern (default gitignore pattern)
            pattern = pathspec.patterns.GitWildMatchPattern
            spec = pathspec.PathSpec.from_lines(pattern, ignored_files)
            # Get tree containing all matching files
            match_tree = spec.match_tree(path)
            # Create list with all absolute paths of match tree
            return ["%s/%s" % (path, match) for match in match_tree]

        files = []
        path = os.getcwd()

        for dir_file in os.walk(path):
            dirname, _, arc = dir_file
            files.extend([os.path.join(dirname, f) for f in arc])

        # Allow the user to run `kytos napps upload` from outside the
        # napp directory.
        # Filter the files with the napp_name in their path
        # Example: home/user/napps/kytos/, napp_name = kronos
        # This filter will get all files from:
        # home/user/napps/kytos/kronos/*
        files = list(filter(lambda x: napp_name in x, files))

        matches = get_matches(path)

        for filename in files.copy():
            if filename in matches:
                files.remove(filename)

        # Create the '.napp' package
        napp_file = tarfile.open(napp_name + '.napp', 'x:xz')
        for local_f in files:
            # Add relative paths instead of absolute paths
            napp_file.add(pathlib.PurePosixPath(local_f).relative_to(path))
        napp_file.close()

        # Get the binary payload of the package
        file_payload = open(napp_name + '.napp', 'rb')

        # remove the created package from the filesystem
        os.remove(napp_name + '.napp')

        return file_payload

    @staticmethod
    def create_metadata(*args, **kwargs):  # pylint: disable=unused-argument
        """Generate the metadata to send the napp package."""
        json_filename = kwargs.get('json_filename', 'kytos.json')
        readme_filename = kwargs.get('readme_filename', 'README.rst')
        ignore_json = kwargs.get('ignore_json', False)
        metadata = {}

        if not ignore_json:
            try:
                with open(json_filename) as json_file:
                    metadata = json.load(json_file)
            except FileNotFoundError:
                print("ERROR: Could not access kytos.json file.")
                sys.exit(1)

        try:
            with open(readme_filename) as readme_file:
                metadata['readme'] = readme_file.read()
        except FileNotFoundError:
            metadata['readme'] = ''

        try:
            yaml = YAML(typ='safe')
            openapi_dict = yaml.load(pathlib.Path('openapi.yml').open())
            openapi = json.dumps(openapi_dict)
        except FileNotFoundError:
            openapi = ''
        metadata['OpenAPI_Spec'] = openapi

        return metadata

    def upload(self, *args, **kwargs):
        """Create package and upload it to NApps Server.

        Raises:
            FileNotFoundError: If kytos.json is not found.

        """
        self.prepare()
        metadata = self.create_metadata(*args, **kwargs)
        package = self.build_napp_package(metadata.get('name'))

        NAppsClient().upload_napp(metadata, package)

    def delete(self):
        """Delete a NApp.

        Raises:
            requests.HTTPError: When there's a server error.

        """
        client = NAppsClient(self._config)
        client.delete(self.user, self.napp)

    @classmethod
    def prepare(cls):
        """Prepare NApp to be uploaded by creating openAPI skeleton."""
        if cls._ask_openapi():
            napp_path = pathlib.Path()
            tpl_path = SKEL_PATH / 'napp-structure/username/napp'
            OpenAPI(napp_path, tpl_path).render_template()
            print('Please, update your openapi.yml file.')
            sys.exit()

    @staticmethod
    def _ask_openapi():
        """Return whether we should create a (new) skeleton."""
        if pathlib.Path('openapi.yml').exists():
            question = 'Override local openapi.yml with a new skeleton? (y/N) '
            default = False
        else:
            question = 'Do you have REST endpoints and wish to create an API' \
                ' skeleton in openapi.yml? (Y/n) '
            default = True

        while True:
            answer = input(question)
            if answer == '':
                return default
            if answer.lower() in ['y', 'yes']:
                return True
            if answer.lower() in ['n', 'no']:
                return False

    def reload(self, napps=None):
        """Reload a NApp or all NApps.

        Args:
            napps (list): NApp list to be reloaded.
        Raises:
            requests.HTTPError: When there's a server error.

        """
        client = NAppsClient(self._config)
        client.reload_napps(napps)


# pylint: enable=too-many-instance-attributes,too-many-public-methods


1		"""Manage Network Application files."""
2	1	import json
3	1	import logging
4	1	import os
5	1	import pathlib
6	1	import re
7	1	import sys
8	1	import tarfile
9	1	import urllib
10	1	from http import HTTPStatus
11
12		# Disable pylint import checks that conflict with isort
13		# pylint: disable=ungrouped-imports,wrong-import-order
14	1	import pathspec
15	1	from jinja2 import Environment, FileSystemLoader
16	1	from ruamel.yaml import YAML
17
18	1	from kytos.utils.client import NAppsClient
19	1	from kytos.utils.config import KytosConfig
20	1	from kytos.utils.exceptions import KytosException
21	1	from kytos.utils.openapi import OpenAPI
22	1	from kytos.utils.settings import SKEL_PATH
23
24	1	LOG = logging.getLogger(__name__)
25
26
27		# pylint: disable=too-many-instance-attributes,too-many-public-methods
28	1	class NAppsManager:
29		"""Deal with NApps at filesystem level and ask Kytos to (un)load NApps."""
30
31	1	_NAPP_ENABLE = "api/kytos/core/napps/{}/{}/enable"
32	1	_NAPP_DISABLE = "api/kytos/core/napps/{}/{}/disable"
33	1	_NAPP_INSTALL = "api/kytos/core/napps/{}/{}/install"
34	1	_NAPP_UNINSTALL = "api/kytos/core/napps/{}/{}/uninstall"
35	1	_NAPPS_INSTALLED = "api/kytos/core/napps_installed"
36	1	_NAPPS_ENABLED = "api/kytos/core/napps_enabled"
37	1	_NAPP_METADATA = "api/kytos/core/napps/{}/{}/metadata/{}"
38
39	1	def __init__(self):
40		"""Instance a new NAppsManager.
41
42		This method do not need parameters.
43		"""
44	1	self._config = KytosConfig().config
45	1	self._kytos_api = self._config.get('kytos', 'api')
46
47	1	self.user = None
48	1	self.napp = None
49	1	self.version = None
50
51		# Automatically get from kytosd API when needed
52	1	self.__local_enabled = None
53	1	self.__local_installed = None
54
55	1	@property
56		def _enabled(self):
57	1	if self.__local_enabled is None:
58	1	self.__require_kytos_config()
59	1	return self.__local_enabled
60
61	1	@property
62		def _installed(self):
63	1	if self.__local_installed is None:
64	1	self.__require_kytos_config()
65	1	return self.__local_installed
66
67	1	def __require_kytos_config(self):
68		"""Set path locations from kytosd API.
69
70		It should not be called directly, but from properties that require a
71		running kytosd instance.
72		"""
73	1	if self.__local_enabled is None:
74	1	uri = self._kytos_api + 'api/kytos/core/config/'
75	1	try:
76	1	ops = json.loads(urllib.request.urlopen(uri).read())
77	1	except urllib.error.URLError as err:
78	1	msg = f'Error connecting to Kytos daemon: {uri} {err.reason}'
79	1	print(msg)
80	1	sys.exit(1)
81	1	self.__local_enabled = pathlib.Path(ops.get('napps'))
82	1	self.__local_installed = pathlib.Path(ops.get('installed_napps'))
83
84	1	def set_napp(self, user, napp, version=None):
85		"""Set info about NApp.
86
87		Args:
88		user (str): NApps Server username.
89		napp (str): NApp name.
90		version (str): NApp version.
91
92		"""
93	1	self.user = user
94	1	self.napp = napp
95	1	self.version = version or 'latest'
96
97	1	@property
98		def napp_id(self):
99		"""Return a Identifier of NApp."""
100	1	return '/'.join((self.user, self.napp))
101
102	1	@staticmethod
103		def _get_napps(napps_dir):
104		"""List of (username, napp_name) found in ``napps_dir``.
105
106		Ex: [('kytos', 'of_core'), ('kytos', 'of_lldp')]
107		"""
108	1	jsons = napps_dir.glob('//kytos.json')
109	1	return sorted(j.parts[-3:-1] for j in jsons)
110
111	1	def get_enabled_local(self):
112		"""Sorted list of (username, napp_name) of enabled napps."""
113	1	return self._get_napps(self._enabled)
114
115	1	def get_installed_local(self):
116		"""Sorted list of (username, napp_name) of installed napps."""
117	1	return self._get_napps(self._installed)
118
119	1	def get_enabled(self):
120		"""Sorted list of (username, napp_name) of enabled napps."""
121	1	uri = self._kytos_api + self._NAPPS_ENABLED
122
123	1	try:
124	1	response = urllib.request.urlopen(uri)
125	1	if response.getcode() != 200:
126	1	msg = "Error calling Kytos to check enabled NApps."
127	1	raise KytosException(msg)
128
129	1	content = json.loads(response.read())
130	1	return sorted((c[0], c[1]) for c in content['napps'])
131	1	except urllib.error.URLError as exception:
132	1	LOG.error("Error checking enabled NApps. Is Kytos running?")
133	1	raise KytosException(exception)
134
135	1	def get_installed(self):
136		"""Sorted list of (username, napp_name) of installed napps."""
137	1	uri = self._kytos_api + self._NAPPS_INSTALLED
138
139	1	try:
140	1	response = urllib.request.urlopen(uri)
141	1	if response.getcode() != 200:
142	1	msg = "Error calling Kytos to check installed NApps."
143	1	raise KytosException(msg)
144
145	1	content = json.loads(response.read())
146	1	return sorted((c[0], c[1]) for c in content['napps'])
147	1	except urllib.error.URLError as exception:
148	1	LOG.error("Error checking installed NApps. Is Kytos running?")
149	1	raise KytosException(exception)
150
151	1	def is_installed(self):
152		"""Whether a NApp is installed."""
153	1	return (self.user, self.napp) in self.get_installed()
154
155	1	def get_disabled(self):
156		"""Sorted list of (username, napp_name) of disabled napps.
157
158		The difference of installed and enabled.
159		"""
160	1	installed = set(self.get_installed())
161	1	enabled = set(self.get_enabled())
162	1	return sorted(installed - enabled)
163
164	1	def dependencies(self, user=None, napp=None):
165		"""Get napp_dependencies from install NApp.
166
167		Args:
168		user(string) A Username.
169		napp(string): A NApp name.
170		Returns:
171		napps(list): List with tuples with Username and NApp name.
172		e.g. [('kytos'/'of_core'), ('kytos/of_l2ls')]
173
174		"""
175	1	napps = self._get_napp_key('napp_dependencies', user, napp)
176	1	return [tuple(napp.split('/')) for napp in napps]
177
178	1	def get_description(self, user=None, napp=None):
179		"""Return the description from kytos.json."""
180	1	return self._get_napp_key('description', user, napp)
181
182	1	def get_version(self, user=None, napp=None):
183		"""Return the version from kytos.json."""
184	1	return self._get_napp_key('version', user, napp) or 'latest'
185
186	1	def _get_napp_key(self, key, user=None, napp=None):
187		"""Return a value from kytos.json.
188
189		Args:
190		user (string): A Username.
191		napp (string): A NApp name
192		key (string): Key used to get the value within kytos.json.
193
194		Returns:
195		meta (object): Value stored in kytos.json.
196
197		"""
198	1	if user is None:
199	1	user = self.user
200	1	if napp is None:
201	1	napp = self.napp
202
203	1	uri = self._kytos_api + self._NAPP_METADATA
204	1	uri = uri.format(user, napp, key)
205
206	1	meta = json.loads(urllib.request.urlopen(uri).read())
207	1	return meta[key]
208
209	1	def disable(self):
210		"""Disable a NApp if it is enabled."""
211	1	uri = self._kytos_api + self._NAPP_DISABLE
212	1	uri = uri.format(self.user, self.napp)
213
214	1	try:
215	1	json.loads(urllib.request.urlopen(uri).read())
216	1	except urllib.error.HTTPError as exception:
217	1	if exception.code == HTTPStatus.BAD_REQUEST.value:
218	1	LOG.error("NApp is not installed. Check the NApp list.")
219		else:
220	1	LOG.error("Error disabling the NApp")
221
222	1	def enable(self):
223		"""Enable a NApp if not already enabled."""
224	1	uri = self._kytos_api + self._NAPP_ENABLE
225	1	uri = uri.format(self.user, self.napp)
226
227	1	try:
228	1	json.loads(urllib.request.urlopen(uri).read())
229	1	except urllib.error.HTTPError as exception:
230	1	if exception.code == HTTPStatus.BAD_REQUEST.value:
231	1	LOG.error("NApp is not installed. Check the NApp list.")
232		else:
233	1	LOG.error("Error enabling the NApp")
234
235	1	def enabled_dir(self):
236		"""Return the enabled dir from current napp."""
237	1	return self._enabled / self.user / self.napp
238
239	1	def installed_dir(self):
240		"""Return the installed dir from current napp."""
241	1	return self._installed / self.user / self.napp
242
243	1	def is_enabled(self):
244		"""Whether a NApp is enabled."""
245	1	return (self.user, self.napp) in self.get_enabled()
246
247	1	def remote_uninstall(self):
248		"""Delete code inside NApp directory, if existent."""
249	1	uri = self._kytos_api + self._NAPP_UNINSTALL
250	1	uri = uri.format(self.user, self.napp)
251
252	1	try:
253	1	json.loads(urllib.request.urlopen(uri).read())
254	1	except urllib.error.HTTPError as exception:
255	1	if exception.code == HTTPStatus.BAD_REQUEST.value:
256	1	LOG.error("Check if the NApp is installed.")
257		else:
258	1	LOG.error("Error uninstalling the NApp")
259
260	1	@staticmethod
261		def valid_name(username):
262		"""Check the validity of the given 'name'.
263
264		The following checks are done:
265		- name starts with a letter
266		- name contains only letters, numbers or underscores
267		"""
268	1	return username and re.match(r'[a-zA-Z][a-zA-Z0-9_]{2,}$', username)
269
270	1	@staticmethod
271		def render_template(templates_path, template_filename, context):
272		"""Render Jinja2 template for a NApp structure."""
273	1	template_env = Environment(
274		autoescape=False, trim_blocks=False,
275		loader=FileSystemLoader(str(templates_path)))
276	1	return template_env.get_template(str(template_filename)) \
277		.render(context)
278
279	1	@staticmethod
280		def search(pattern):
281		"""Search all server NApps matching pattern.
282
283		Args:
284		pattern (str): Python regular expression.
285
286		"""
287	1	def match(napp):
288		"""Whether a NApp metadata matches the pattern."""
289		# WARNING: This will change for future versions, when 'author' will
290		# be removed.
291	1	username = napp.get('username', napp.get('author'))
292
293	1	strings = ['{}/{}'.format(username, napp.get('name')),
294		napp.get('description')] + napp.get('tags')
295	1	return any(pattern.match(string) for string in strings)
296
297	1	napps = NAppsClient().get_napps()
298	1	return [napp for napp in napps if match(napp)]
299
300	1	def remote_install(self):
301		"""Ask kytos server to install NApp."""
302	1	uri = self._kytos_api + self._NAPP_INSTALL
303	1	uri = uri.format(self.user, self.napp)
304
305	1	json.loads(urllib.request.urlopen(uri).read())
306
307	1	@classmethod
308		# pylint: disable=too-many-statements
309	1	def create_napp(cls, meta_package=False):
310		"""Bootstrap a basic NApp structure for you to develop your NApp.
311
312		This will create, on the current folder, a clean structure of a NAPP,
313		filling some contents on this structure.
314		"""
315	1	templates_path = SKEL_PATH / 'napp-structure/username/napp'
316
317	1	ui_templates_path = os.path.join(templates_path, 'ui')
318
319	1	username = None
320	1	napp_name = None
321	1	keyint_error = "\nUser cancelled napps creation."
322
323	1	print('--------------------------------------------------------------')
324	1	print('Welcome to the bootstrap process of your NApp.')
325	1	print('--------------------------------------------------------------')
326	1	print('In order to answer both the username and the NApp name,')
327	1	print('You must follow these naming rules:')
328	1	print(' - name starts with a letter')
329	1	print(' - name contains only letters, numbers or underscores')
330	1	print(' - at least three characters')
331	1	print('--------------------------------------------------------------')
332	1	print('')
333	1	try:
334	1	while not cls.valid_name(username):
335	1	username = input('Please, insert your NApps Server username: ')
336	1	while not cls.valid_name(napp_name):
337	1	napp_name = input('Please, insert your NApp name: ')
338	1	description = input('Please, insert a brief description for your '
339		'NApp [optional]: ')
340		except KeyboardInterrupt:
341		print(keyint_error)
342		sys.exit(0)
343	1	if not description:
		0 ignored issues – show introduced 2020-12-23 17:13 UTC by Report Bug Copy Issue Report The variable `description` does not seem to be defined in case the `while` loop on line `336` is not entered. Are you sure this can never be the case? Loading history...
344		# pylint: disable=fixme
345	1	description = '# TODO: <<<< Insert your NApp description here >>>>'
346		# pylint: enable=fixme
347
348	1	context = {'username': username, 'napp': napp_name,
349		'description': description}
350
351		#: Creating the directory structure (username/napp_name)
352	1	os.makedirs(username, exist_ok=True)
353
354		#: Creating ``__init__.py`` files
355	1	with open(os.path.join(username, '__init__.py'), 'w') as init_file:
356	1	init_file.write(f'"""NApps for the user {username}.""""')
357
358	1	os.makedirs(os.path.join(username, napp_name))
359
360		#: Creating the other files based on the templates
361	1	templates = os.listdir(templates_path)
362	1	templates.remove('ui')
363	1	templates.remove('openapi.yml.template')
364
365	1	if meta_package:
366		templates.remove('main.py.template')
367		templates.remove('settings.py.template')
368
369	1	for tmp in templates:
370	1	fname = os.path.join(username, napp_name,
371		tmp.rsplit('.template')[0])
372	1	with open(fname, 'w') as file:
373	1	content = cls.render_template(templates_path, tmp, context)
374	1	file.write(content)
375
376	1	if not meta_package:
377	1	NAppsManager.create_ui_structure(username, napp_name,
378		ui_templates_path, context)
379
380	1	print()
381
382	1	print(f'Congratulations! Your NApp has been bootstrapped!\nNow '
383		f'you can go to the directory "{username}/{napp_name}" and '
384		' begin to code your NApp.')
385	1	print('Have fun!')
386
387	1	@classmethod
388		def create_ui_structure(cls, username, napp_name, ui_templates_path,
389		context):
390		"""Create the ui directory structure."""
391	1	for section in ['k-info-panel', 'k-toolbar', 'k-action-menu']:
392	1	os.makedirs(os.path.join(username, napp_name, 'ui', section))
393
394	1	templates = os.listdir(ui_templates_path)
395
396	1	for tmp in templates:
397	1	fname = os.path.join(username, napp_name, 'ui',
398		tmp.rsplit('.template')[0])
399
400	1	with open(fname, 'w') as file:
401	1	content = cls.render_template(ui_templates_path, tmp,
402		context)
403	1	file.write(content)
404
405	1	@staticmethod
406		def _check_module(folder):
407		"""Create module folder with empty __init__.py if it doesn't exist.
408
409		Args:
410		folder (pathlib.pathlib.Path): Module path.
411
412		"""
413	1	if not folder.exists():
414	1	folder.mkdir(parents=True, exist_ok=True, mode=0o755)
415	1	(folder / '__init__.py').touch()
416
417	1	@staticmethod
418		def build_napp_package(napp_name):
419		"""Build the .napp file to be sent to the napps server.
420
421		Args:
422		napp_identifier (str): Identifier formatted as
423		<username>/<napp_name>
424
425		Return:
426		file_payload (binary): The binary representation of the napp
427		package that will be POSTed to the napp server.
428
429		"""
430	1	def get_matches(path):
431		"""Return all NApp files matching any .gitignore pattern."""
432	1	ignored_files = [".git"]
433	1	with open(".gitignore", 'r') as local_gitignore:
434	1	ignored_files.extend(local_gitignore.readlines())
435
436	1	user_gitignore_path = pathlib.Path("%s/.gitignore" %
437		pathlib.Path.home())
438	1	if user_gitignore_path.exists():
439		with open(user_gitignore_path, 'r') as user_gitignore:
440		ignored_files.extend(user_gitignore.readlines())
441
442		# Define Wildmatch pattern (default gitignore pattern)
443	1	pattern = pathspec.patterns.GitWildMatchPattern
444	1	spec = pathspec.PathSpec.from_lines(pattern, ignored_files)
445		# Get tree containing all matching files
446	1	match_tree = spec.match_tree(path)
447		# Create list with all absolute paths of match tree
448	1	return ["%s/%s" % (path, match) for match in match_tree]
449
450	1	files = []
451	1	path = os.getcwd()
452
453	1	for dir_file in os.walk(path):
454	1	dirname, _, arc = dir_file
455	1	files.extend([os.path.join(dirname, f) for f in arc])
456
457		# Allow the user to run `kytos napps upload` from outside the
458		# napp directory.
459		# Filter the files with the napp_name in their path
460		# Example: home/user/napps/kytos/, napp_name = kronos
461		# This filter will get all files from:
462		# home/user/napps/kytos/kronos/*
463	1	files = list(filter(lambda x: napp_name in x, files))
464
465	1	matches = get_matches(path)
466
467	1	for filename in files.copy():
468	1	if filename in matches:
469	1	files.remove(filename)
470
471		# Create the '.napp' package
472	1	napp_file = tarfile.open(napp_name + '.napp', 'x:xz')
473	1	for local_f in files:
474		# Add relative paths instead of absolute paths
475	1	napp_file.add(pathlib.PurePosixPath(local_f).relative_to(path))
476	1	napp_file.close()
477
478		# Get the binary payload of the package
479	1	file_payload = open(napp_name + '.napp', 'rb')
480
481		# remove the created package from the filesystem
482	1	os.remove(napp_name + '.napp')
483
484	1	return file_payload
485
486	1	@staticmethod
487		def create_metadata(args, *kwargs): # pylint: disable=unused-argument
488		"""Generate the metadata to send the napp package."""
489	1	json_filename = kwargs.get('json_filename', 'kytos.json')
490	1	readme_filename = kwargs.get('readme_filename', 'README.rst')
491	1	ignore_json = kwargs.get('ignore_json', False)
492	1	metadata = {}
493
494	1	if not ignore_json:
495	1	try:
496	1	with open(json_filename) as json_file:
497	1	metadata = json.load(json_file)
498		except FileNotFoundError:
499		print("ERROR: Could not access kytos.json file.")
500		sys.exit(1)
501
502	1	try:
503	1	with open(readme_filename) as readme_file:
504	1	metadata['readme'] = readme_file.read()
505		except FileNotFoundError:
506		metadata['readme'] = ''
507
508	1	try:
509	1	yaml = YAML(typ='safe')
510	1	openapi_dict = yaml.load(pathlib.Path('openapi.yml').open())
511	1	openapi = json.dumps(openapi_dict)
512		except FileNotFoundError:
513		openapi = ''
514	1	metadata['OpenAPI_Spec'] = openapi
515
516	1	return metadata
517
518	1	def upload(self, args, *kwargs):
519		"""Create package and upload it to NApps Server.
520
521		Raises:
522		FileNotFoundError: If kytos.json is not found.
523
524		"""
525	1	self.prepare()
526	1	metadata = self.create_metadata(args, *kwargs)
527	1	package = self.build_napp_package(metadata.get('name'))
528
529	1	NAppsClient().upload_napp(metadata, package)
530
531	1	def delete(self):
532		"""Delete a NApp.
533
534		Raises:
535		requests.HTTPError: When there's a server error.
536
537		"""
538	1	client = NAppsClient(self._config)
539	1	client.delete(self.user, self.napp)
540
541	1	@classmethod
542		def prepare(cls):
543		"""Prepare NApp to be uploaded by creating openAPI skeleton."""
544	1	if cls._ask_openapi():
545	1	napp_path = pathlib.Path()
546	1	tpl_path = SKEL_PATH / 'napp-structure/username/napp'
547	1	OpenAPI(napp_path, tpl_path).render_template()
548	1	print('Please, update your openapi.yml file.')
549	1	sys.exit()
550
551	1	@staticmethod
552		def _ask_openapi():
553		"""Return whether we should create a (new) skeleton."""
554	1	if pathlib.Path('openapi.yml').exists():
555	1	question = 'Override local openapi.yml with a new skeleton? (y/N) '
556	1	default = False
557		else:
558	1	question = 'Do you have REST endpoints and wish to create an API' \
559		' skeleton in openapi.yml? (Y/n) '
560	1	default = True
561
562	1	while True:
563	1	answer = input(question)
564	1	if answer == '':
565	1	return default
566	1	if answer.lower() in ['y', 'yes']:
567	1	return True
568	1	if answer.lower() in ['n', 'no']:
569	1	return False
570
571	1	def reload(self, napps=None):
572		"""Reload a NApp or all NApps.
573
574		Args:
575		napps (list): NApp list to be reloaded.
576		Raises:
577		requests.HTTPError: When there's a server error.
578
579		"""
580	1	client = NAppsClient(self._config)
581	1	client.reload_napps(napps)
582
583
584		# pylint: enable=too-many-instance-attributes,too-many-public-methods
585

kytos / kytos-utils

Pull Request — master (#310)

kytos.utils.napps.NAppsManager.create_napp() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like