NAppsManager.get_version() - Code Metrics - Inspection of "Merge pull request #109 from macartur/handle_multi..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 327bd3...378eb3 )

by Carlos Eduardo

created 2017-08-03 02:53 UTC

NAppsManager.get_version() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	1	Features	0

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

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

from jinja2 import Environment, FileSystemLoader

from 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
        self.version = 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, 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."""
        version = self._get_napp_key('version', user, napp)
        if not version:
            msg = '{}/{} does not have the key "version" in kytos.json'
            raise KeyError(msg.format(user, napp))
        return version

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

kytos / kytos-utils

Push — master ( 327bd3...378eb3 )

NAppsManager.get_version() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like