NAppsManager.reload() - Code Metrics - Inspection of "Merge pull request #176 from llgoncalves/master" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 79cc3a...1e4540 )

by Beraldo

created 2018-06-15 13:28 UTC

NAppsManager.reload() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	11
rs	9.4285

"""Manage Network Application files."""
import json
import logging
import os
import re
import shutil
import sys
import tarfile
import urllib
from pathlib import Path
from random import randint

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.openapi import OpenAPI

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

    def __init__(self, controller=None):
        """If controller is not informed, the necessary paths must be.

        If ``controller`` is available, NApps will be (un)loaded at runtime and
        you don't need to inform the paths. Otherwise, you should inform the
        required paths for the methods called.

        Args:
            controller (kytos.Controller): Controller to (un)load NApps.
            install_path (str): Folder where NApps should be installed. If
                None, use the controller's configuration.
            enabled_path (str): Folder where enabled NApps are stored. If None,
                use the controller's configuration.
        """
        self._controller = controller
        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.__enabled = None
        self.__installed = None

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

    @property
    def _installed(self):
        if self.__installed is None:
            self.__require_kytos_config()
        return self.__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.__enabled is None:
            uri = self._kytos_api + 'api/kytos/core/config/'
            try:
                options = json.loads(urllib.request.urlopen(uri).read())
            except urllib.error.URLError:
                print('Kytos is not running.')
                sys.exit()
            self.__enabled = Path(options.get('napps'))
            self.__installed = Path(options.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``."""
        jsons = napps_dir.glob('*/*/kytos.json')
        return sorted(j.parts[-3:-1] for j in jsons)

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

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

    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
        kytos_json = self._installed / user / napp / 'kytos.json'
        try:
            with kytos_json.open() as file_descriptor:
                meta = json.load(file_descriptor)
                return meta[key]
        except (FileNotFoundError, json.JSONDecodeError, KeyError):
            return ''

    def disable(self):
        """Disable a NApp if it is enabled."""
        enabled = self.enabled_dir()
        try:
            enabled.unlink()
            if self._controller is not None:
                self._controller.unload_napp(self.user, self.napp)
        except FileNotFoundError:
            pass  # OK, it was already disabled

    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 enable(self):
        """Enable a NApp if not already enabled.

        Raises:
            FileNotFoundError: If NApp is not installed.
            PermissionError: No filesystem permission to enable NApp.

        """
        enabled = self.enabled_dir()
        installed = self.installed_dir()

        if not installed.is_dir():
            raise FileNotFoundError('Install NApp {} first.'.format(
                self.napp_id))
        elif not enabled.exists():
            self._check_module(enabled.parent)
            try:
                # Create symlink
                enabled.symlink_to(installed)
                if self._controller is not None:
                    self._controller.load_napp(self.user, self.napp)
            except FileExistsError:
                pass  # OK, NApp was already enabled
            except PermissionError:
                raise PermissionError('Permission error on enabling NApp. Try '
                                      'with sudo.')

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

    def uninstall(self):
        """Delete code inside NApp directory, if existent."""
        if self.is_installed():
            installed = self.installed_dir()
            if installed.is_symlink():
                installed.unlink()
            else:
                shutil.rmtree(str(installed))

    @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 install_local(self):
        """Make a symlink in install folder to a local NApp.

        Raises:
            FileNotFoundError: If NApp is not found.

        """
        folder = self._get_local_folder()
        installed = self.installed_dir()
        self._check_module(installed.parent)
        installed.symlink_to(folder.resolve())

    def _get_local_folder(self, root=None):
        """Return local NApp root folder.

        Search for kytos.json in _./_ folder and _./user/napp_.

        Args:
            root (pathlib.Path): Where to begin searching.

        Return:
            pathlib.Path: NApp root folder.

        Raises:
            FileNotFoundError: If there is no such local NApp.

        """
        if root is None:
            root = Path()
        for folders in ['.'], [self.user, self.napp]:
            kytos_json = root / Path(*folders) / 'kytos.json'
            if kytos_json.exists():
                with kytos_json.open() as file_descriptor:
                    meta = json.load(file_descriptor)
                    # WARNING: This will change in future versions, when
                    # 'author' will be removed.
                    username = meta.get('username', meta.get('author'))
                    if username == self.user and meta.get('name') == self.napp:
                        return kytos_json.parent
        raise FileNotFoundError('kytos.json not found.')

    def install_remote(self):
        """Download, extract and install NApp."""
        package, pkg_folder = None, None
        try:
            package = self._download()
            pkg_folder = self._extract(package)
            napp_folder = self._get_local_folder(pkg_folder)
            dst = self._installed / self.user / self.napp
            self._check_module(dst.parent)
            shutil.move(str(napp_folder), str(dst))
        finally:
            # Delete temporary files
            if package:
                Path(package).unlink()
            if pkg_folder and pkg_folder.exists():
                shutil.rmtree(str(pkg_folder))

    def _download(self):
        """Download NApp package from server.

        Return:
            str: Downloaded temp filename.

        Raises:
            urllib.error.HTTPError: If download is not successful.

        """
        repo = self._config.get('napps', 'repo')
        napp_id = '{}/{}-{}.napp'.format(self.user, self.napp, self.version)
        uri = os.path.join(repo, napp_id)
        return urllib.request.urlretrieve(uri)[0]

    @staticmethod
    def _extract(filename):
        """Extract package to a temporary folder.

        Return:
            pathlib.Path: Temp dir with package contents.

        """
        random_string = '{:0d}'.format(randint(0, 10**6))
        tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
        os.mkdir(tmp)
        with tarfile.open(filename, 'r:xz') as tar:
            tar.extractall(tmp)
        return Path(tmp)

    @classmethod
    def create_napp(cls):
        """Bootstrap a basic NApp strucutre 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.
        """
        base = os.environ.get('VIRTUAL_ENV', '/')

        templates_path = os.path.join(base, 'etc', 'skel', 'kytos',
                                      'napp-structure', 'username', 'napp')

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

        username = None
        napp_name = None
        description = 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('')
        msg = 'Please, insert your NApps Server username: '
        while not cls.valid_name(username):
            username = input(msg)

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

        msg = 'Please, insert a brief description for your NApp [optional]: '
        description = input(msg)
        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'):
            pass

        os.makedirs(os.path.join(username, napp_name))
        with open(os.path.join(username, napp_name, '__init__.py'), 'w'):
            pass

        #: Creating 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))

        #: Creating the other files based on the templates
        templates = os.listdir(templates_path)
        templates.remove('ui')
        templates.remove('__init__.py')
        templates.remove('openapi.yml.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)

        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)

        msg = '\nCongratulations! Your NApp has been bootstrapped!\nNow you '
        msg += 'can go to the directory {}/{} and begin to code your NApp.'
        print(msg.format(username, napp_name))
        print('Have fun!')

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

        Args:
            folder (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.

        """
        ignored_extensions = ['.swp', '.pyc', '.napp']
        ignored_dirs = ['__pycache__']
        files = os.listdir()
        for filename in files:
            if os.path.isfile(filename) and '.' in filename and \
                    filename.rsplit('.', 1)[1] in ignored_extensions:
                files.remove(filename)
            elif os.path.isdir(filename) and filename in ignored_dirs:
                files.remove(filename)

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

kytos / kytos-utils

Push — master ( 79cc3a...1e4540 )

NAppsManager.reload() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like