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

Test Failed

Pull Request — master (#88)

by macartur

created 2017-06-06 18:39 UTC

NAppsManager.uninstall() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	3
c	3
b	1
f	0
dl	0
loc	8
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

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

kytos / kytos-utils

Pull Request — master (#88)

NAppsManager.uninstall() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like