kytos.utils.napps.NAppsManager.build_napp_package() - Code Metrics - Inspection of "Change .gitignore parser (Fix #221)" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#258)

by Gleyberson

created 2019-12-18 16:48 UTC

NAppsManager.build_napp_package() B

↳ Parent: kytos.utils.napps

Complexity

Conditions

Size

Total Lines	62
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	50.7597

Importance

Changes

Metric	Value
cc	7
eloc	27
nop	1
dl	0
loc	62
ccs	1
cts	27
cp	0.037
crap	50.7597
rs	7.8319
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 installed 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)

        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 installing the NApp.")

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

kytos / kytos-utils

Pull Request — master (#258)

NAppsManager.build_napp_package() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like