NAppsManager.render_template() - Code Metrics - Inspection of "Adding install NApps dependecies" - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#74)

by macartur

created 2017-05-02 19:59 UTC

NAppsManager.render_template() A

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

kytos / kytos-utils

Pull Request — master (#74)

NAppsManager.render_template() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like