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

Test Failed

Push — master ( 3f9872...314d6d )

by Beraldo

created 2017-06-08 13:34 UTC

NAppsManager.dependencies() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	12
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._installed = Path(self._config.get('napps', 'installed_path'))
        self._enabled = Path(self._config.get('napps', 'enabled_path'))

        self.user = None
        self.napp = None

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

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

    @property
    def napp_id(self):
        """Identifier of NApp."""
        return '/'.join((self.user, self.napp))

    @staticmethod
    def _get_napps(napps_dir):
        """List of (username, napp_name) found in ``napps_dir``."""
        jsons = napps_dir.glob('*/*/kytos.json')
        return sorted(j.parts[-3:-1] for j in jsons)

    def get_enabled(self):
        """Sorted list of (username, napp_name) of enabled napps."""
        return self._get_napps(self._enabled)

    def get_installed(self):
        """Sorted list of (username, napp_name) of installed napps."""
        return self._get_napps(self._installed)

    def is_installed(self):
        """Whether a NApp is installed."""
        return (self.user, self.napp) in self.get_installed()

    def get_disabled(self):
        """Sorted list of (username, napp_name) of disabled napps.

        The difference of installed and enabled.
        """
        installed = set(self.get_installed())
        enabled = set(self.get_enabled())
        return sorted(installed - enabled)

    def dependencies(self, user=None, napp=None):
        """Method used to get napp_dependencies from install NApp.

        Args:
            user(string)  A Username.
            napp(string): A NApp name.
        Returns:
            napps(list): List with tuples with Username and NApp name.
                         e.g. [('kytos'/'of_core'), ('kytos/of_l2ls')]
        """
        napps = self._get_napp_key('napp_dependencies', user, napp)
        return [tuple(napp.split('/')) for napp in napps]

    def get_description(self, user=None, napp=None):
        """Return the description from kytos.json."""
        return self._get_napp_key('description', user, napp)

    def _get_napp_key(self, key, user=None, napp=None):
        """Generic method used to return a value from kytos.json.

        Args:
            user (string): A Username.
            napp (string): A NApp name
            key (string): Key used to get the value within kytos.json.
        Returns:
            meta (object): Value stored in kytos.json.
        """
        if user is None:
            user = self.user
        if napp is None:
            napp = self.napp
        kj = self._installed / user / napp / 'kytos.json'
        try:
            with kj.open() as f:
                meta = json.load(f)
                return meta[key]
        except (FileNotFoundError, json.JSONDecodeError, KeyError):
            return ''

    def disable(self):
        """Disable a NApp if it is enabled."""
        enabled = self._enabled / self.user / self.napp
        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 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 / self.user / self.napp
        installed = self._installed / self.user / self.napp

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

kytos / kytos-utils

Push — master ( 3f9872...314d6d )

NAppsManager.dependencies() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like