NAppsManager.installed_dir() - Code Metrics - Inspection of "Change to use napps_dir from kytos instance" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#88)

by macartur

created 2017-06-06 18:52 UTC

NAppsManager.installed_dir() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

kytos / kytos-utils

Pull Request — master (#88)

NAppsManager.installed_dir() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like