NAppsManager.napp_id() - Code Metrics - Inspection of "Merge pull request #88 from macartur/issue-#85" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8e3af8...0c35a4 )

by Beraldo

created 2017-06-12 17:23 UTC

NAppsManager.napp_id() 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	4
rs	10

"""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 kytos.utils.client import NAppsClient
from kytos.utils.config import KytosConfig

log = logging.getLogger(__name__)


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._load_kytos_configuration()

        self.user = None
        self.napp = None

    def _load_kytos_configuration(self):
        """Request current configurations loaded by Kytos instance."""
        uri = self._kytos_api + 'kytos/config/'
        try:
            options = json.loads(urllib.request.urlopen(uri).read())
        except urllib.error.URLError:
            print('Kytos is not running.')
            sys.exit()

        self._installed = Path(options.get('installed_napps'))
        self._enabled = Path(options.get('napps'))

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

        Args:
            user (str): NApps Server username.
            napp (str): NApp name.
        """
        self.user = user
        self.napp = napp

    @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_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
        kj = self._installed / user / napp / 'kytos.json'
        try:
            with kj.open() as f:
                meta = json.load(f)
                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(templates_path))
        return TEMPLATE_ENV.get_template(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 f:
                    meta = json.load(f)
                    # 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')
        uri = os.path.join(repo, self.user, '{}-latest.napp'.format(self.napp))
        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:
            description = \
                '# TODO: <<<< Insert here your NApp description >>>>'  # noqa

        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):
        """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'] = ''

        return metadata

    def upload(self, *args, **kwargs):
        """Create package and upload it to NApps Server.

        Raises:
            FileNotFoundError: If kytos.json is not found.
        """
        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)


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

kytos / kytos-utils

Push — master ( 8e3af8...0c35a4 )

NAppsManager.napp_id() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like