NAppsManager.get_version() - Code Metrics - Inspection of "Change 'kytos napps list' to show napp version" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#109)

by macartur

created 2017-07-21 13:29 UTC

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

kytos / kytos-utils

Pull Request — master (#109)

NAppsManager.get_version() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like