kytos.utils.napps.NAppsManager.remote_install() - Code Metrics - Inspection of "Enable Scrutinizer coverage stats" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 901c2b...6be349 )

by Humberto

created 2019-08-27 13:23 UTC

kytos.utils.napps.NAppsManager.remote_install() A

↳ Parent: kytos.utils.napps

Complexity

Conditions

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	9.3211

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	1
dl	0
loc	12
rs	9.95
c	0
b	0
f	0
ccs	1
cts	9
cp	0.1111
crap	9.3211

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

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

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

kytos / kytos-utils

Push — master ( 901c2b...6be349 )

kytos.utils.napps.NAppsManager.remote_install() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like