NAppsManager._get_local_folder() - Code Metrics - Inspection of "Merge pull request #121 from cemsbr/openapi" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( cb8f01...434676 )

by Carlos Eduardo

created 2017-09-19 22:10 UTC

NAppsManager._get_local_folder() C

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
cc	7
c	4
b	0
f	0
dl	0
loc	27
rs	5.5

"""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):
        """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):
        """Method used to 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):
        """Generic method used to 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.

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

        Return:
            pathlib.Path: NApp root folder.
        """
        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.

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

        Return:
            str: Downloaded temp filename.
        """
        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')
        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 here your NApp description >>>>'
            # 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 other files based on the templates
        templates = os.listdir(templates_path)
        templates.remove('__init__.py')
        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)

        msg = '\nCongratulations! Your NApp have 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

# 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			"""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			"""Method used to 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			napps = self._get_napp_key('napp_dependencies', user, napp)
136			return [tuple(napp.split('/')) for napp in napps]
137
138			def get_description(self, user=None, napp=None):
139			"""Return the description from kytos.json."""
140			return self._get_napp_key('description', user, napp)
141
142			def get_version(self, user=None, napp=None):
143			"""Return the version from kytos.json."""
144			return self._get_napp_key('version', user, napp) or 'latest'
145
146			def _get_napp_key(self, key, user=None, napp=None):
147			"""Generic method used to return a value from kytos.json.
148
149			Args:
150			user (string): A Username.
151			napp (string): A NApp name
152			key (string): Key used to get the value within kytos.json.
153			Returns:
154			meta (object): Value stored in kytos.json.
155			"""
156			if user is None:
157			user = self.user
158			if napp is None:
159			napp = self.napp
160			kytos_json = self._installed / user / napp / 'kytos.json'
161			try:
162			with kytos_json.open() as file_descriptor:
163			meta = json.load(file_descriptor)
164			return meta[key]
165			except (FileNotFoundError, json.JSONDecodeError, KeyError):
166			return ''
167
168			def disable(self):
169			"""Disable a NApp if it is enabled."""
170			enabled = self.enabled_dir()
171			try:
172			enabled.unlink()
173			if self._controller is not None:
174			self._controller.unload_napp(self.user, self.napp)
175			except FileNotFoundError:
176			pass # OK, it was already disabled
177
178			def enabled_dir(self):
179			"""Return the enabled dir from current napp."""
180			return self._enabled / self.user / self.napp
181
182			def installed_dir(self):
183			"""Return the installed dir from current napp."""
184			return self._installed / self.user / self.napp
185
186			def enable(self):
187			"""Enable a NApp if not already enabled.
188
189			Raises:
190			FileNotFoundError: If NApp is not installed.
191			PermissionError: No filesystem permission to enable NApp.
192			"""
193			enabled = self.enabled_dir()
194			installed = self.installed_dir()
195
196			if not installed.is_dir():
197			raise FileNotFoundError('Install NApp {} first.'.format(
198			self.napp_id))
199			elif not enabled.exists():
200			self._check_module(enabled.parent)
201			try:
202			# Create symlink
203			enabled.symlink_to(installed)
204			if self._controller is not None:
205			self._controller.load_napp(self.user, self.napp)
206			except FileExistsError:
207			pass # OK, NApp was already enabled
208			except PermissionError:
209			raise PermissionError('Permission error on enabling NApp. Try '
210			'with sudo.')
211
212			def is_enabled(self):
213			"""Whether a NApp is enabled."""
214			return (self.user, self.napp) in self.get_enabled()
215
216			def uninstall(self):
217			"""Delete code inside NApp directory, if existent."""
218			if self.is_installed():
219			installed = self.installed_dir()
220			if installed.is_symlink():
221			installed.unlink()
222			else:
223			shutil.rmtree(str(installed))
224
225			@staticmethod
226			def valid_name(username):
227			"""Check the validity of the given 'name'.
228
229			The following checks are done:
230			- name starts with a letter
231			- name contains only letters, numbers or underscores
232			"""
233			return username and re.match(r'[a-zA-Z][a-zA-Z0-9_]{2,}$', username)
234
235			@staticmethod
236			def render_template(templates_path, template_filename, context):
237			"""Render Jinja2 template for a NApp structure."""
238			template_env = Environment(
239			autoescape=False, trim_blocks=False,
240			loader=FileSystemLoader(str(templates_path)))
241			return template_env.get_template(str(template_filename)) \
242			.render(context)
243
244			@staticmethod
245			def search(pattern):
246			"""Search all server NApps matching pattern.
247
248			Args:
249			pattern (str): Python regular expression.
250			"""
251			def match(napp):
252			"""Whether a NApp metadata matches the pattern."""
253			# WARNING: This will change for future versions, when 'author' will
254			# be removed.
255			username = napp.get('username', napp.get('author'))
256
257			strings = ['{}/{}'.format(username, napp.get('name')),
258			napp.get('description')] + napp.get('tags')
259			return any(pattern.match(string) for string in strings)
260
261			napps = NAppsClient().get_napps()
262			return [napp for napp in napps if match(napp)]
263
264			def install_local(self):
265			"""Make a symlink in install folder to a local NApp.
266
267			Raises:
268			FileNotFoundError: If NApp is not found.
269			"""
270			folder = self._get_local_folder()
271			installed = self.installed_dir()
272			self._check_module(installed.parent)
273			installed.symlink_to(folder.resolve())
274
275			def _get_local_folder(self, root=None):
276			"""Return local NApp root folder.
277
278			Search for kytos.json in _./_ folder and _./user/napp_.
279
280			Args:
281			root (pathlib.Path): Where to begin searching.
282
283			Raises:
284			FileNotFoundError: If there is no such local NApp.
285
286			Return:
287			pathlib.Path: NApp root folder.
288			"""
289			if root is None:
290			root = Path()
291			for folders in ['.'], [self.user, self.napp]:
292			kytos_json = root / Path(*folders) / 'kytos.json'
293			if kytos_json.exists():
294			with kytos_json.open() as file_descriptor:
295			meta = json.load(file_descriptor)
296			# WARNING: This will change in future versions, when
297			# 'author' will be removed.
298			username = meta.get('username', meta.get('author'))
299			if username == self.user and meta.get('name') == self.napp:
300			return kytos_json.parent
301			raise FileNotFoundError('kytos.json not found.')
302
303			def install_remote(self):
304			"""Download, extract and install NApp."""
305			package, pkg_folder = None, None
306			try:
307			package = self._download()
308			pkg_folder = self._extract(package)
309			napp_folder = self._get_local_folder(pkg_folder)
310			dst = self._installed / self.user / self.napp
311			self._check_module(dst.parent)
312			shutil.move(str(napp_folder), str(dst))
313			finally:
314			# Delete temporary files
315			if package:
316			Path(package).unlink()
317			if pkg_folder and pkg_folder.exists():
318			shutil.rmtree(str(pkg_folder))
319
320			def _download(self):
321			"""Download NApp package from server.
322
323			Raises:
324			urllib.error.HTTPError: If download is not successful.
325
326			Return:
327			str: Downloaded temp filename.
328			"""
329			repo = self._config.get('napps', 'repo')
330			napp_id = '{}/{}-{}.napp'.format(self.user, self.napp, self.version)
331			uri = os.path.join(repo, napp_id)
332			return urllib.request.urlretrieve(uri)[0]
333
334			@staticmethod
335			def _extract(filename):
336			"""Extract package to a temporary folder.
337
338			Return:
339			pathlib.Path: Temp dir with package contents.
340			"""
341			random_string = '{:0d}'.format(randint(0, 10**6))
342			tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
343			os.mkdir(tmp)
344			with tarfile.open(filename, 'r:xz') as tar:
345			tar.extractall(tmp)
346			return Path(tmp)
347
348			@classmethod
349			def create_napp(cls):
350			"""Bootstrap a basic NApp strucutre for you to develop your NApp.
351
352			This will create, on the current folder, a clean structure of a NAPP,
353			filling some contents on this structure.
354			"""
355			base = os.environ.get('VIRTUAL_ENV', '/')
356
357			templates_path = os.path.join(base, 'etc', 'skel', 'kytos',
358			'napp-structure', 'username', 'napp')
359			username = None
360			napp_name = None
361			description = None
362			print('--------------------------------------------------------------')
363			print('Welcome to the bootstrap process of your NApp.')
364			print('--------------------------------------------------------------')
365			print('In order to answer both the username and the napp name,')
366			print('You must follow this naming rules:')
367			print(' - name starts with a letter')
368			print(' - name contains only letters, numbers or underscores')
369			print(' - at least three characters')
370			print('--------------------------------------------------------------')
371			print('')
372			msg = 'Please, insert your NApps Server username: '
373			while not cls.valid_name(username):
374			username = input(msg)
375
376			while not cls.valid_name(napp_name):
377			napp_name = input('Please, insert your NApp name: ')
378
379			msg = 'Please, insert a brief description for your NApp [optional]: '
380			description = input(msg)
381			if not description:
382			# pylint: disable=fixme
383			description = '# TODO: <<<< Insert here your NApp description >>>>'
384			# pylint: enable=fixme
385
386			context = {'username': username, 'napp': napp_name,
387			'description': description}
388
389			#: Creating the directory structure (username/napp_name)
390			os.makedirs(username, exist_ok=True)
391			#: Creating ``__init__.py`` files
392			with open(os.path.join(username, '__init__.py'), 'w'):
393			pass
394
395			os.makedirs(os.path.join(username, napp_name))
396			with open(os.path.join(username, napp_name, '__init__.py'), 'w'):
397			pass
398
399			#: Creating the other files based on the templates
400			templates = os.listdir(templates_path)
401			templates.remove('__init__.py')
402			for tmp in templates:
403			fname = os.path.join(username, napp_name,
404			tmp.rsplit('.template')[0])
405			with open(fname, 'w') as file:
406			content = cls.render_template(templates_path, tmp, context)
407			file.write(content)
408
409			msg = '\nCongratulations! Your NApp have been bootstrapped!\nNow you '
410			msg += 'can go to the directory {}/{} and begin to code your NApp.'
411			print(msg.format(username, napp_name))
412			print('Have fun!')
413
414			@staticmethod
415			def _check_module(folder):
416			"""Create module folder with empty __init__.py if it doesn't exist.
417
418			Args:
419			folder (pathlib.Path): Module path.
420			"""
421			if not folder.exists():
422			folder.mkdir(parents=True, exist_ok=True, mode=0o755)
423			(folder / '__init__.py').touch()
424
425			@staticmethod
426			def build_napp_package(napp_name):
427			"""Build the .napp file to be sent to the napps server.
428
429			Args:
430			napp_identifier (str): Identifier formatted as
431			<username>/<napp_name>
432
433			Return:
434			file_payload (binary): The binary representation of the napp
435			package that will be POSTed to the napp server.
436			"""
437			ignored_extensions = ['.swp', '.pyc', '.napp']
438			ignored_dirs = ['__pycache__']
439			files = os.listdir()
440			for filename in files:
441			if os.path.isfile(filename) and '.' in filename and \
442			filename.rsplit('.', 1)[1] in ignored_extensions:
443			files.remove(filename)
444			elif os.path.isdir(filename) and filename in ignored_dirs:
445			files.remove(filename)
446
447			# Create the '.napp' package
448			napp_file = tarfile.open(napp_name + '.napp', 'x:xz')
449			for local_f in files:
450			napp_file.add(local_f)
451			napp_file.close()
452
453			# Get the binary payload of the package
454			file_payload = open(napp_name + '.napp', 'rb')
455
456			# remove the created package from the filesystem
457			os.remove(napp_name + '.napp')
458
459			return file_payload
460
461			@staticmethod
462			def create_metadata(args, *kwargs): # pylint: disable=unused-argument
463			"""Generate the metadata to send the napp package."""
464			json_filename = kwargs.get('json_filename', 'kytos.json')
465			readme_filename = kwargs.get('readme_filename', 'README.rst')
466			ignore_json = kwargs.get('ignore_json', False)
467			metadata = {}
468
469			if not ignore_json:
470			try:
471			with open(json_filename) as json_file:
472			metadata = json.load(json_file)
473			except FileNotFoundError:
474			print("ERROR: Could not access kytos.json file.")
475			sys.exit(1)
476
477			try:
478			with open(readme_filename) as readme_file:
479			metadata['readme'] = readme_file.read()
480			except FileNotFoundError:
481			metadata['readme'] = ''
482
483			try:
484			yaml = YAML(typ='safe')
485			openapi_dict = yaml.load(Path('openapi.yml').open())
486			openapi = json.dumps(openapi_dict)
487			except FileNotFoundError:
488			openapi = ''
489			metadata['OpenAPI_Spec'] = openapi
490
491			return metadata
492
493			def upload(self, args, *kwargs):
494			"""Create package and upload it to NApps Server.
495
496			Raises:
497			FileNotFoundError: If kytos.json is not found.
498			"""
499			self.prepare()
500			metadata = self.create_metadata(args, *kwargs)
501			package = self.build_napp_package(metadata.get('name'))
502
503			NAppsClient().upload_napp(metadata, package)
504
505			def delete(self):
506			"""Delete a NApp.
507
508			Raises:
509			requests.HTTPError: When there's a server error.
510			"""
511			client = NAppsClient(self._config)
512			client.delete(self.user, self.napp)
513
514			@classmethod
515			def prepare(cls):
516			"""Prepare NApp to be uploaded by creating openAPI skeleton."""
517			if cls._ask_openapi():
518			napp_path = Path()
519			prefix = Path(sys.prefix)
520			tpl_path = prefix / 'etc/skel/kytos/napp-structure/username/napp'
521			OpenAPI(napp_path, tpl_path).render_template()
522			print('Please, update your openapi.yml file.')
523			sys.exit()
524
525			@staticmethod
526			def _ask_openapi():
527			"""Return whether we should create a (new) skeleton."""
528			if Path('openapi.yml').exists():
529			question = 'Override local openapi.yml with a new skeleton? (y/N) '
530			default = False
531			else:
532			question = 'Do you have REST endpoints and wish to create an API' \
533			' skeleton in openapi.yml? (Y/n) '
534			default = True
535
536			while True:
537			answer = input(question)
538			if answer == '':
539			return default
540			if answer.lower() in ['y', 'yes']:
541			return True
542			if answer.lower() in ['n', 'no']:
543			return False
544
545			# pylint: enable=too-many-instance-attributes,too-many-public-methods
546

kytos / kytos-utils

Push — master ( cb8f01...434676 )

NAppsManager._get_local_folder() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like