kytos.utils.napps.NAppsManager.build_napp_package() - Code Metrics - Inspection of "Merge pull request #279 from josemauro/issue-278" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f73ec9...5a2d58 )

by Humberto

created 2020-09-08 18:18 UTC

NAppsManager.build_napp_package() C

↳ Parent: kytos.utils.napps

Complexity

Conditions

Size

Total Lines	68
Code Lines	32

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	28
CRAP Score	9.0239

Importance

Changes

Metric	Value
cc	9
eloc	32
nop	1
dl	0
loc	68
ccs	28
cts	30
cp	0.9333
crap	9.0239
rs	6.6666
c	0
b	0
f	0

How to fix Long Method

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

        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 this naming rules:')
        print(' - name starts with a letter')
        print(' - name contains only letters, numbers or underscores')
        print(' - at least three characters')
        print('--------------------------------------------------------------')
        print('')
        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]: ')
        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 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	1	def create_napp(cls, meta_package=False):
309		"""Bootstrap a basic NApp structure for you to develop your NApp.
310
311		This will create, on the current folder, a clean structure of a NAPP,
312		filling some contents on this structure.
313		"""
314	1	templates_path = SKEL_PATH / 'napp-structure/username/napp'
315
316	1	ui_templates_path = os.path.join(templates_path, 'ui')
317
318	1	username = None
319	1	napp_name = None
320
321	1	print('--------------------------------------------------------------')
322	1	print('Welcome to the bootstrap process of your NApp.')
323	1	print('--------------------------------------------------------------')
324	1	print('In order to answer both the username and the napp name,')
325	1	print('You must follow this naming rules:')
326	1	print(' - name starts with a letter')
327	1	print(' - name contains only letters, numbers or underscores')
328	1	print(' - at least three characters')
329	1	print('--------------------------------------------------------------')
330	1	print('')
331	1	while not cls.valid_name(username):
332	1	username = input('Please, insert your NApps Server username: ')
333
334	1	while not cls.valid_name(napp_name):
335	1	napp_name = input('Please, insert your NApp name: ')
336
337	1	description = input('Please, insert a brief description for your'
338		'NApp [optional]: ')
339	1	if not description:
340		# pylint: disable=fixme
341	1	description = '# TODO: <<<< Insert your NApp description here >>>>'
342		# pylint: enable=fixme
343
344	1	context = {'username': username, 'napp': napp_name,
345		'description': description}
346
347		#: Creating the directory structure (username/napp_name)
348	1	os.makedirs(username, exist_ok=True)
349
350		#: Creating ``__init__.py`` files
351	1	with open(os.path.join(username, '__init__.py'), 'w') as init_file:
352	1	init_file.write(f'"""Napps for the user {username}.""""')
353
354	1	os.makedirs(os.path.join(username, napp_name))
355
356		#: Creating the other files based on the templates
357	1	templates = os.listdir(templates_path)
358	1	templates.remove('ui')
359	1	templates.remove('openapi.yml.template')
360
361	1	if meta_package:
362		templates.remove('main.py.template')
363		templates.remove('settings.py.template')
364
365	1	for tmp in templates:
366	1	fname = os.path.join(username, napp_name,
367		tmp.rsplit('.template')[0])
368	1	with open(fname, 'w') as file:
369	1	content = cls.render_template(templates_path, tmp, context)
370	1	file.write(content)
371
372	1	if not meta_package:
373	1	NAppsManager.create_ui_structure(username, napp_name,
374		ui_templates_path, context)
375
376	1	print()
377	1	print(f'Congratulations! Your NApp has been bootstrapped!\nNow you '
378		'can go to the directory {username}/{napp_name} and begin to '
379		'code your NApp.')
380	1	print('Have fun!')
381
382	1	@classmethod
383		def create_ui_structure(cls, username, napp_name, ui_templates_path,
384		context):
385		"""Create the ui directory structure."""
386	1	for section in ['k-info-panel', 'k-toolbar', 'k-action-menu']:
387	1	os.makedirs(os.path.join(username, napp_name, 'ui', section))
388
389	1	templates = os.listdir(ui_templates_path)
390
391	1	for tmp in templates:
392	1	fname = os.path.join(username, napp_name, 'ui',
393		tmp.rsplit('.template')[0])
394
395	1	with open(fname, 'w') as file:
396	1	content = cls.render_template(ui_templates_path, tmp,
397		context)
398	1	file.write(content)
399
400	1	@staticmethod
401		def _check_module(folder):
402		"""Create module folder with empty __init__.py if it doesn't exist.
403
404		Args:
405		folder (pathlib.pathlib.Path): Module path.
406
407		"""
408	1	if not folder.exists():
409	1	folder.mkdir(parents=True, exist_ok=True, mode=0o755)
410	1	(folder / '__init__.py').touch()
411
412	1	@staticmethod
413		def build_napp_package(napp_name):
414		"""Build the .napp file to be sent to the napps server.
415
416		Args:
417		napp_identifier (str): Identifier formatted as
418		<username>/<napp_name>
419
420		Return:
421		file_payload (binary): The binary representation of the napp
422		package that will be POSTed to the napp server.
423
424		"""
425	1	def get_matches(path):
426		"""Return all NApp files matching any .gitignore pattern."""
427	1	ignored_files = [".git"]
428	1	with open(".gitignore", 'r') as local_gitignore:
429	1	ignored_files.extend(local_gitignore.readlines())
430
431	1	user_gitignore_path = pathlib.Path("%s/.gitignore" %
432		pathlib.Path.home())
433	1	if user_gitignore_path.exists():
434		with open(user_gitignore_path, 'r') as user_gitignore:
435		ignored_files.extend(user_gitignore.readlines())
436
437		# Define Wildmatch pattern (default gitignore pattern)
438	1	pattern = pathspec.patterns.GitWildMatchPattern
439	1	spec = pathspec.PathSpec.from_lines(pattern, ignored_files)
440		# Get tree containing all matching files
441	1	match_tree = spec.match_tree(path)
442		# Create list with all absolute paths of match tree
443	1	return ["%s/%s" % (path, match) for match in match_tree]
444
445	1	files = []
446	1	path = os.getcwd()
447
448	1	for dir_file in os.walk(path):
449	1	dirname, _, arc = dir_file
450	1	files.extend([os.path.join(dirname, f) for f in arc])
451
452		# Allow the user to run `kytos napps upload` from outside the
453		# napp directory.
454		# Filter the files with the napp_name in their path
455		# Example: home/user/napps/kytos/, napp_name = kronos
456		# This filter will get all files from:
457		# home/user/napps/kytos/kronos/*
458	1	files = list(filter(lambda x: napp_name in x, files))
459
460	1	matches = get_matches(path)
461
462	1	for filename in files.copy():
463	1	if filename in matches:
464	1	files.remove(filename)
465
466		# Create the '.napp' package
467	1	napp_file = tarfile.open(napp_name + '.napp', 'x:xz')
468	1	for local_f in files:
469		# Add relative paths instead of absolute paths
470	1	napp_file.add(pathlib.PurePosixPath(local_f).relative_to(path))
471	1	napp_file.close()
472
473		# Get the binary payload of the package
474	1	file_payload = open(napp_name + '.napp', 'rb')
475
476		# remove the created package from the filesystem
477	1	os.remove(napp_name + '.napp')
478
479	1	return file_payload
480
481	1	@staticmethod
482		def create_metadata(args, *kwargs): # pylint: disable=unused-argument
483		"""Generate the metadata to send the napp package."""
484	1	json_filename = kwargs.get('json_filename', 'kytos.json')
485	1	readme_filename = kwargs.get('readme_filename', 'README.rst')
486	1	ignore_json = kwargs.get('ignore_json', False)
487	1	metadata = {}
488
489	1	if not ignore_json:
490	1	try:
491	1	with open(json_filename) as json_file:
492	1	metadata = json.load(json_file)
493		except FileNotFoundError:
494		print("ERROR: Could not access kytos.json file.")
495		sys.exit(1)
496
497	1	try:
498	1	with open(readme_filename) as readme_file:
499	1	metadata['readme'] = readme_file.read()
500		except FileNotFoundError:
501		metadata['readme'] = ''
502
503	1	try:
504	1	yaml = YAML(typ='safe')
505	1	openapi_dict = yaml.load(pathlib.Path('openapi.yml').open())
506	1	openapi = json.dumps(openapi_dict)
507		except FileNotFoundError:
508		openapi = ''
509	1	metadata['OpenAPI_Spec'] = openapi
510
511	1	return metadata
512
513	1	def upload(self, args, *kwargs):
514		"""Create package and upload it to NApps Server.
515
516		Raises:
517		FileNotFoundError: If kytos.json is not found.
518
519		"""
520	1	self.prepare()
521	1	metadata = self.create_metadata(args, *kwargs)
522	1	package = self.build_napp_package(metadata.get('name'))
523
524	1	NAppsClient().upload_napp(metadata, package)
525
526	1	def delete(self):
527		"""Delete a NApp.
528
529		Raises:
530		requests.HTTPError: When there's a server error.
531
532		"""
533	1	client = NAppsClient(self._config)
534	1	client.delete(self.user, self.napp)
535
536	1	@classmethod
537		def prepare(cls):
538		"""Prepare NApp to be uploaded by creating openAPI skeleton."""
539	1	if cls._ask_openapi():
540	1	napp_path = pathlib.Path()
541	1	tpl_path = SKEL_PATH / 'napp-structure/username/napp'
542	1	OpenAPI(napp_path, tpl_path).render_template()
543	1	print('Please, update your openapi.yml file.')
544	1	sys.exit()
545
546	1	@staticmethod
547		def _ask_openapi():
548		"""Return whether we should create a (new) skeleton."""
549	1	if pathlib.Path('openapi.yml').exists():
550	1	question = 'Override local openapi.yml with a new skeleton? (y/N) '
551	1	default = False
552		else:
553	1	question = 'Do you have REST endpoints and wish to create an API' \
554		' skeleton in openapi.yml? (Y/n) '
555	1	default = True
556
557	1	while True:
558	1	answer = input(question)
559	1	if answer == '':
560	1	return default
561	1	if answer.lower() in ['y', 'yes']:
562	1	return True
563	1	if answer.lower() in ['n', 'no']:
564	1	return False
565
566	1	def reload(self, napps=None):
567		"""Reload a NApp or all NApps.
568
569		Args:
570		napps (list): NApp list to be reloaded.
571		Raises:
572		requests.HTTPError: When there's a server error.
573
574		"""
575	1	client = NAppsClient(self._config)
576	1	client.reload_napps(napps)
577
578
579		# pylint: enable=too-many-instance-attributes,too-many-public-methods
580

kytos / kytos-utils

Push — master ( f73ec9...5a2d58 )

NAppsManager.build_napp_package() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like