kytos.utils.napps.NAppsManager.get_enabled_local() - Code Metrics - Inspection of "Bump certifi from 2019.11.28 to 2022.12.7 in /requ..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#324)

unknown

created 2022-12-08 03:13 UTC

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

↳ Parent: kytos.utils.napps

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	3
ccs	0
cts	2
cp	0
crap	2
rs	10
c	0
b	0
f	0

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

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

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

LOG = logging.getLogger(__name__)


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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        Args:
            pattern (str): Python regular expression.

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

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

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

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

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

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

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

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

        username = None
        napp_name = None
        print('--------------------------------------------------------------')
        print('Welcome to the bootstrap process of your NApp.')
        print('--------------------------------------------------------------')
        print('In order to answer both the username and the NApp name,')
        print('You must follow these naming rules:')
        print(' - name starts with a letter')
        print(' - name contains only letters, numbers or underscores')
        print(' - at least three characters')
        print('--------------------------------------------------------------')
        print('')
        try:
            while not cls.valid_name(username):
                username = input('Please, insert your NApps Server username: ')

            while not cls.valid_name(napp_name):
                napp_name = input('Please, insert your NApp name: ')

            description = input('Please, insert a brief description for your '
                                'NApp [optional]: ')
        except KeyboardInterrupt:
            print("User cancelled NApp creation.")
            sys.exit(0)
        if not description:
            # pylint: disable=fixme
            description = '# TODO: <<<< Insert your NApp description here >>>>'
            # pylint: enable=fixme

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

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

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

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

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

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

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

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

        print('\nCongratulations! Your NApp has been bootstrapped!\nNow you'
              f' 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			import json
3			import logging
4			import os
5			import pathlib
6			import re
7			import sys
8			import tarfile
9			import urllib
10			from http import HTTPStatus
11
12			# Disable pylint import checks that conflict with isort
13			# pylint: disable=ungrouped-imports,wrong-import-order
14			import pathspec
15			from jinja2 import Environment, FileSystemLoader
16			from ruamel.yaml import YAML
17
18			from kytos.utils.client import NAppsClient
19			from kytos.utils.config import KytosConfig
20			from kytos.utils.exceptions import KytosException
21			from kytos.utils.openapi import OpenAPI
22			from kytos.utils.settings import SKEL_PATH
23
24			LOG = logging.getLogger(__name__)
25
26
27			# 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
31			_NAPP_ENABLE = "api/kytos/core/napps/{}/{}/enable"
32			_NAPP_DISABLE = "api/kytos/core/napps/{}/{}/disable"
33			_NAPP_INSTALL = "api/kytos/core/napps/{}/{}/install"
34			_NAPP_UNINSTALL = "api/kytos/core/napps/{}/{}/uninstall"
35			_NAPPS_INSTALLED = "api/kytos/core/napps_installed"
36			_NAPPS_ENABLED = "api/kytos/core/napps_enabled"
37			_NAPP_METADATA = "api/kytos/core/napps/{}/{}/metadata/{}"
38
39			def __init__(self):
40			"""Instance a new NAppsManager.
41
42			This method do not need parameters.
43			"""
44			self._config = KytosConfig().config
45			self._kytos_api = self._config.get('kytos', 'api')
46
47			self.user = None
48			self.napp = None
49			self.version = None
50
51			# Automatically get from kytosd API when needed
52			self.__local_enabled = None
53			self.__local_installed = None
54
55			@property
56			def _enabled(self):
57			if self.__local_enabled is None:
58			self.__require_kytos_config()
59			return self.__local_enabled
60
61			@property
62			def _installed(self):
63			if self.__local_installed is None:
64			self.__require_kytos_config()
65			return self.__local_installed
66
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
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
97			@property
98			def napp_id(self):
99			"""Return a Identifier of NApp."""
100			return '/'.join((self.user, self.napp))
101
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			"""
108			jsons = napps_dir.glob('//kytos.json')
109			return sorted(j.parts[-3:-1] for j in jsons)
110
111			def get_enabled_local(self):
112			"""Sorted list of (username, napp_name) of enabled napps."""
113			return self._get_napps(self._enabled)
114
115			def get_installed_local(self):
116			"""Sorted list of (username, napp_name) of installed napps."""
117			return self._get_napps(self._installed)
118
119			def get_enabled(self):
120			"""Sorted list of (username, napp_name) of enabled napps."""
121			uri = self._kytos_api + self._NAPPS_ENABLED
122
123			try:
124			response = urllib.request.urlopen(uri)
125			if response.getcode() != 200:
126			msg = "Error calling Kytos to check enabled NApps."
127			raise KytosException(msg)
128
129			content = json.loads(response.read())
130			return sorted((c[0], c[1]) for c in content['napps'])
131			except urllib.error.URLError as exception:
132			LOG.error("Error checking enabled NApps. Is Kytos running?")
133			raise KytosException(exception)
134
135			def get_installed(self):
136			"""Sorted list of (username, napp_name) of installed napps."""
137			uri = self._kytos_api + self._NAPPS_INSTALLED
138
139			try:
140			response = urllib.request.urlopen(uri)
141			if response.getcode() != 200:
142			msg = "Error calling Kytos to check installed NApps."
143			raise KytosException(msg)
144
145			content = json.loads(response.read())
146			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
151			def is_installed(self):
152			"""Whether a NApp is installed."""
153			return (self.user, self.napp) in self.get_installed()
154
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
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
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
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
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
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
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
235			def enabled_dir(self):
236			"""Return the enabled dir from current napp."""
237			return self._enabled / self.user / self.napp
238
239			def installed_dir(self):
240			"""Return the installed dir from current napp."""
241			return self._installed / self.user / self.napp
242
243			def is_enabled(self):
244			"""Whether a NApp is enabled."""
245			return (self.user, self.napp) in self.get_enabled()
246
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
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
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
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
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			json.loads(urllib.request.urlopen(uri).read())
306
307			@classmethod
308			# pylint: disable=too-many-statements
309			def create_napp(cls, meta_package=False):
310			"""Bootstrap a basic NApp structure for you to develop your NApp.
311
312			This will create, on the current folder, a clean structure of a NAPP,
313			filling some contents on this structure.
314			"""
315			templates_path = SKEL_PATH / 'napp-structure/username/napp'
316
317			ui_templates_path = os.path.join(templates_path, 'ui')
318
319			username = None
320			napp_name = None
321			print('--------------------------------------------------------------')
322			print('Welcome to the bootstrap process of your NApp.')
323			print('--------------------------------------------------------------')
324			print('In order to answer both the username and the NApp name,')
325			print('You must follow these naming rules:')
326			print(' - name starts with a letter')
327			print(' - name contains only letters, numbers or underscores')
328			print(' - at least three characters')
329			print('--------------------------------------------------------------')
330			print('')
331			try:
332			while not cls.valid_name(username):
333			username = input('Please, insert your NApps Server username: ')
334
335			while not cls.valid_name(napp_name):
336			napp_name = input('Please, insert your NApp name: ')
337
338			description = input('Please, insert a brief description for your '
339			'NApp [optional]: ')
340			except KeyboardInterrupt:
341			print("User cancelled NApp creation.")
342			sys.exit(0)
343			if not description:
344			# pylint: disable=fixme
345			description = '# TODO: <<<< Insert your NApp description here >>>>'
346			# pylint: enable=fixme
347
348			context = {'username': username, 'napp': napp_name,
349			'description': description}
350
351			#: Creating the directory structure (username/napp_name)
352			os.makedirs(username, exist_ok=True)
353
354			#: Creating ``__init__.py`` files
355			with open(os.path.join(username, '__init__.py'), 'w') as init_file:
356			init_file.write(f'"""NApps for the user {username}.""""')
357
358			os.makedirs(os.path.join(username, napp_name))
359
360			#: Creating the other files based on the templates
361			templates = os.listdir(templates_path)
362			templates.remove('ui')
363			templates.remove('openapi.yml.template')
364
365			if meta_package:
366			templates.remove('main.py.template')
367			templates.remove('settings.py.template')
368
369			for tmp in templates:
370			fname = os.path.join(username, napp_name,
371			tmp.rsplit('.template')[0])
372			with open(fname, 'w') as file:
373			content = cls.render_template(templates_path, tmp, context)
374			file.write(content)
375
376			if not meta_package:
377			NAppsManager.create_ui_structure(username, napp_name,
378			ui_templates_path, context)
379
380			print('\nCongratulations! Your NApp has been bootstrapped!\nNow you'
381			f' can go to the directory "{username}/{napp_name}" and begin'
382			' to code your NApp.')
383			print('Have fun!')
384
385			@classmethod
386			def create_ui_structure(cls, username, napp_name, ui_templates_path,
387			context):
388			"""Create the ui directory structure."""
389			for section in ['k-info-panel', 'k-toolbar', 'k-action-menu']:
390			os.makedirs(os.path.join(username, napp_name, 'ui', section))
391
392			templates = os.listdir(ui_templates_path)
393
394			for tmp in templates:
395			fname = os.path.join(username, napp_name, 'ui',
396			tmp.rsplit('.template')[0])
397
398			with open(fname, 'w') as file:
399			content = cls.render_template(ui_templates_path, tmp,
400			context)
401			file.write(content)
402
403			@staticmethod
404			def _check_module(folder):
405			"""Create module folder with empty __init__.py if it doesn't exist.
406
407			Args:
408			folder (pathlib.pathlib.Path): Module path.
409
410			"""
411			if not folder.exists():
412			folder.mkdir(parents=True, exist_ok=True, mode=0o755)
413			(folder / '__init__.py').touch()
414
415			@staticmethod
416			def build_napp_package(napp_name):
417			"""Build the .napp file to be sent to the napps server.
418
419			Args:
420			napp_identifier (str): Identifier formatted as
421			<username>/<napp_name>
422
423			Return:
424			file_payload (binary): The binary representation of the napp
425			package that will be POSTed to the napp server.
426
427			"""
428			def get_matches(path):
429			"""Return all NApp files matching any .gitignore pattern."""
430			ignored_files = [".git"]
431			with open(".gitignore", 'r') as local_gitignore:
432			ignored_files.extend(local_gitignore.readlines())
433
434			user_gitignore_path = pathlib.Path("%s/.gitignore" %
435			pathlib.Path.home())
436			if user_gitignore_path.exists():
437			with open(user_gitignore_path, 'r') as user_gitignore:
438			ignored_files.extend(user_gitignore.readlines())
439
440			# Define Wildmatch pattern (default gitignore pattern)
441			pattern = pathspec.patterns.GitWildMatchPattern
442			spec = pathspec.PathSpec.from_lines(pattern, ignored_files)
443			# Get tree containing all matching files
444			match_tree = spec.match_tree(path)
445			# Create list with all absolute paths of match tree
446			return ["%s/%s" % (path, match) for match in match_tree]
447
448			files = []
449			path = os.getcwd()
450
451			for dir_file in os.walk(path):
452			dirname, _, arc = dir_file
453			files.extend([os.path.join(dirname, f) for f in arc])
454
455			# Allow the user to run `kytos napps upload` from outside the
456			# napp directory.
457			# Filter the files with the napp_name in their path
458			# Example: home/user/napps/kytos/, napp_name = kronos
459			# This filter will get all files from:
460			# home/user/napps/kytos/kronos/*
461			files = list(filter(lambda x: napp_name in x, files))
462
463			matches = get_matches(path)
464
465			for filename in files.copy():
466			if filename in matches:
467			files.remove(filename)
468
469			# Create the '.napp' package
470			napp_file = tarfile.open(napp_name + '.napp', 'x:xz')
471			for local_f in files:
472			# Add relative paths instead of absolute paths
473			napp_file.add(pathlib.PurePosixPath(local_f).relative_to(path))
474			napp_file.close()
475
476			# Get the binary payload of the package
477			file_payload = open(napp_name + '.napp', 'rb')
478
479			# remove the created package from the filesystem
480			os.remove(napp_name + '.napp')
481
482			return file_payload
483
484			@staticmethod
485			def create_metadata(args, *kwargs): # pylint: disable=unused-argument
486			"""Generate the metadata to send the napp package."""
487			json_filename = kwargs.get('json_filename', 'kytos.json')
488			readme_filename = kwargs.get('readme_filename', 'README.rst')
489			ignore_json = kwargs.get('ignore_json', False)
490			metadata = {}
491
492			if not ignore_json:
493			try:
494			with open(json_filename) as json_file:
495			metadata = json.load(json_file)
496			except FileNotFoundError:
497			print("ERROR: Could not access kytos.json file.")
498			sys.exit(1)
499
500			try:
501			with open(readme_filename) as readme_file:
502			metadata['readme'] = readme_file.read()
503			except FileNotFoundError:
504			metadata['readme'] = ''
505
506			try:
507			yaml = YAML(typ='safe')
508			openapi_dict = yaml.load(pathlib.Path('openapi.yml').open())
509			openapi = json.dumps(openapi_dict)
510			except FileNotFoundError:
511			openapi = ''
512			metadata['OpenAPI_Spec'] = openapi
513
514			return metadata
515
516			def upload(self, args, *kwargs):
517			"""Create package and upload it to NApps Server.
518
519			Raises:
520			FileNotFoundError: If kytos.json is not found.
521
522			"""
523			self.prepare()
524			metadata = self.create_metadata(args, *kwargs)
525			package = self.build_napp_package(metadata.get('name'))
526
527			NAppsClient().upload_napp(metadata, package)
528
529			def delete(self):
530			"""Delete a NApp.
531
532			Raises:
533			requests.HTTPError: When there's a server error.
534
535			"""
536			client = NAppsClient(self._config)
537			client.delete(self.user, self.napp)
538
539			@classmethod
540			def prepare(cls):
541			"""Prepare NApp to be uploaded by creating openAPI skeleton."""
542			if cls._ask_openapi():
543			napp_path = pathlib.Path()
544			tpl_path = SKEL_PATH / 'napp-structure/username/napp'
545			OpenAPI(napp_path, tpl_path).render_template()
546			print('Please, update your openapi.yml file.')
547			sys.exit()
548
549			@staticmethod
550			def _ask_openapi():
551			"""Return whether we should create a (new) skeleton."""
552			if pathlib.Path('openapi.yml').exists():
553			question = 'Override local openapi.yml with a new skeleton? (y/N) '
554			default = False
555			else:
556			question = 'Do you have REST endpoints and wish to create an API' \
557			' skeleton in openapi.yml? (Y/n) '
558			default = True
559
560			while True:
561			answer = input(question)
562			if answer == '':
563			return default
564			if answer.lower() in ['y', 'yes']:
565			return True
566			if answer.lower() in ['n', 'no']:
567			return False
568
569			def reload(self, napps=None):
570			"""Reload a NApp or all NApps.
571
572			Args:
573			napps (list): NApp list to be reloaded.
574			Raises:
575			requests.HTTPError: When there's a server error.
576
577			"""
578			client = NAppsClient(self._config)
579			client.reload_napps(napps)
580
581
582			# pylint: enable=too-many-instance-attributes,too-many-public-methods
583

kytos / kytos-utils

Pull Request — master (#324)

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like