kytos.cli.commands.napps.api.NAppsAPI.print_napps() - Code Metrics - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

NAppsAPI.print_napps() A
last analyzed 2021-05-31 20:20 UTC

↳ Parent: kytos.cli.commands.napps.api

Complexity

Conditions

Size

Total Lines	24
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	17
CRAP Score	4.0186

Importance

Changes

Metric	Value
cc	4
eloc	20
nop	1
dl	0
loc	24
ccs	17
cts	19
cp	0.8947
crap	4.0186
rs	9.4
c	0
b	0
f	0

"""Translate cli commands to non-cli code."""
import json
import logging
import os
import re
from urllib.error import HTTPError, URLError

import requests

from kytos.utils.exceptions import KytosException
from kytos.utils.napps import NAppsManager

LOG = logging.getLogger(__name__)


class NAppsAPI:
    """An API for the command-line interface.

    Use the config file only for required options. Static methods are called
    by the parser and they instantiate an object of this class to fulfill the
    request.
    """

    @classmethod
    def disable(cls, args):
        """Disable subcommand."""
        mgr = NAppsManager()

        if args['all']:
            napps = mgr.get_enabled()
        else:
            napps = args['<napp>']

        for napp in napps:
            mgr.set_napp(*napp)
            LOG.info('NApp %s:', mgr.napp_id)
            cls.disable_napp(mgr)

    @staticmethod
    def disable_napp(mgr):
        """Disable a NApp."""
        if mgr.is_enabled():
            LOG.info('  Disabling...')
            mgr.disable()
            LOG.info('  Disabled.')
        else:
            LOG.error("  NApp isn't enabled.")

    @classmethod
    def enable(cls, args):
        """Enable subcommand."""
        mgr = NAppsManager()

        if args['all']:
            napps = mgr.get_disabled()
        else:
            napps = args['<napp>']

        cls.enable_napps(napps)

    @classmethod
    def enable_napp(cls, mgr):
        """Install one NApp using NAppManager object."""
        try:
            if not mgr.is_enabled():
                LOG.info('    Enabling...')
                mgr.enable()

            # Check if NApp is enabled
            if mgr.is_enabled():
                LOG.info('    Enabled.')
            else:
                LOG.error('    Error enabling NApp.')
        except (FileNotFoundError, PermissionError) as exception:
            LOG.error('  %s', exception)

    @classmethod
    def enable_napps(cls, napps):
        """Enable a list of NApps.

        Args:
            napps (list): List of NApps.
        """
        mgr = NAppsManager()
        for napp in napps:
            mgr.set_napp(*napp)
            LOG.info('NApp %s:', mgr.napp_id)
            cls.enable_napp(mgr)

    @classmethod
    def create(cls, args):  # pylint: disable=unused-argument
        """Bootstrap a basic NApp structure on the current folder."""
        NAppsManager.create_napp(meta_package=args.get('--meta', False))

    @classmethod
    def upload(cls, args):  # pylint: disable=unused-argument
        """Upload the NApp to the NApps server.

        Create the NApp package and upload it to the NApp server.
        """
        try:
            NAppsManager().upload()
        except FileNotFoundError as err:
            LOG.error("Couldn't find %s in current directory.", err.filename)

    @classmethod
    def uninstall(cls, args):
        """Uninstall and delete NApps.

        For local installations, do not delete code outside install_path and
        enabled_path.
        """
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            LOG.info('NApp %s:', mgr.napp_id)
            if mgr.is_installed():
                if mgr.is_enabled():
                    cls.disable_napp(mgr)
                LOG.info('  Uninstalling...')
                mgr.remote_uninstall()
                LOG.info('  Uninstalled.')
            else:
                LOG.error("  NApp isn't installed.")

    @classmethod
    def install(cls, args):
        """Install local or remote NApps."""
        cls.install_napps(args['<napp>'])

    @classmethod
    def install_napps(cls, napps):
        """Install local or remote NApps.

        This method is recursive, it will install each napps and your
        dependencies.
        """
        mgr = NAppsManager()
        for napp in napps:
            mgr.set_napp(*napp)
            LOG.info('  NApp %s:', mgr.napp_id)

            try:
                if not mgr.is_installed():
                    # Try to install all NApps, even if
                    # some of them fail.
                    cls.install_napp(mgr)

                    # Enable the NApp
                    if not mgr.is_enabled():
                        cls.enable_napp(mgr)
                        napp_dependencies = mgr.dependencies()
                        if napp_dependencies:
                            LOG.info('Installing Dependencies:')
                            cls.install_napps(napp_dependencies)
                    else:
                        LOG.info('    Enabled.')
                else:
                    LOG.warning('  Napp already installed.')
            except KytosException:
                LOG.error('Error installing NApp.')
                continue

    @classmethod
    def install_napp(cls, mgr):
        """Install a NApp.

        Raises:
            KytosException: If a NApp hasn't been found.

        """
        LOG.info('    Downloading from NApps Server...')
        try:
            mgr.remote_install()
            LOG.info('    Downloaded and installed.')
            return
        except HTTPError as exception:
            if exception.code == 404:
                LOG.error('    NApp not found.')
                LOG.info("    If you are trying to install a local NApp, "
                         "use the command 'python setup.py develop' "
                         "inside the 'user/napp' directory.")
            elif exception.code == 400:
                LOG.error('    NApps Server error: %s', exception)
        except URLError as exception:
            LOG.error('    NApps Server error: %s', str(exception.reason))
        raise KytosException("NApp not found.")

    @classmethod
    def search(cls, args):
        """Search for NApps in NApps server matching a pattern."""
        safe_shell_pat = re.escape(args['<pattern>']).replace(r'\*', '.*')
        pat_str = '.*{}.*'.format(safe_shell_pat)
        pattern = re.compile(pat_str, re.IGNORECASE)
        remote_json = NAppsManager.search(pattern)
        remote = set()
        for napp in remote_json:
            # WARNING: This will be changed in future versions, when 'author'
            # will be removed.
            username = napp.get('username', napp.get('author'))
            remote.add(((username, napp.get('name')), napp.get('description')))

        cls._print_napps(remote)

    @classmethod
    def _print_napps(cls, napp_list):
        """Format the NApp list to be printed."""
        mgr = NAppsManager()
        enabled = mgr.get_enabled()
        installed = mgr.get_installed()
        napps = []
        for napp, desc in sorted(napp_list):
            status = 'i' if napp in installed else '-'
            status += 'e' if napp in enabled else '-'
            status = '[{}]'.format(status)
            name = '{}/{}'.format(*napp)
            napps.append((status, name, desc))
        cls.print_napps(napps)

    @classmethod
    def list(cls, args):  # pylint: disable=unused-argument
        """List all installed NApps and inform whether they are enabled."""
        mgr = NAppsManager()

        # Add status
        napps = [napp + ('[ie]',) for napp in mgr.get_enabled()]
        napps += [napp + ('[i-]',) for napp in mgr.get_disabled()]

        # Sort, add description and reorder columns
        napps.sort()
        napps_ordered = []
        for user, name, status in napps:
            description = mgr.get_description(user, name)
            version = mgr.get_version(user, name)
            napp_id = f'{user}/{name}'
            if version:
                napp_id += f':{version}'

            napps_ordered.append((status, napp_id, description))

        cls.print_napps(napps_ordered)

    @staticmethod
    def print_napps(napps):
        """Print status, name and description."""
        if not napps:
            print('No NApps found.')
            return

        stat_w = 6  # We already know the size of Status col
        name_w = max(len(n[1]) for n in napps)
        desc_w = max(len(n[2]) for n in napps)
        term_w = os.popen('stty size', 'r').read().split()[1]
        remaining = max(0, int(term_w) - stat_w - name_w - 6)
        desc_w = min(desc_w, remaining)
        widths = (stat_w, name_w, desc_w)

        header = '\n{:^%d} | {:^%d} | {:^%d}' % widths
        row = '{:^%d} | {:<%d} | {:<%d}' % widths
        print(header.format('Status', 'NApp ID', 'Description'))
        print('=+='.join('=' * w for w in widths))
        for user, name, desc in napps:
            desc = (desc[:desc_w - 3] + '...') if len(desc) > desc_w else desc
            print(row.format(user, name, desc))

        print('\nStatus: (i)nstalled, (e)nabled\n')

    @staticmethod
    def delete(args):
        """Delete NApps from server."""
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            LOG.info('Deleting NApp %s from server...', mgr.napp_id)
            try:
                mgr.delete()
                LOG.info('  Deleted.')
            except requests.HTTPError as exception:
                if exception.response.status_code == 405:
                    LOG.error('Delete Napp is not allowed yet.')
                else:
                    msg = json.loads(exception.response.content)
                    LOG.error('  Server error: %s - ', msg['error'])

    @classmethod
    def prepare(cls, args):  # pylint: disable=unused-argument
        """Create OpenAPI v3.0 spec skeleton."""
        mgr = NAppsManager()
        mgr.prepare()

    @classmethod
    def reload(cls, args):
        """Reload NApps code."""
        LOG.info('Reloading NApps...')
        mgr = NAppsManager()

        try:
            if args['all']:
                mgr.reload(None)
            else:
                napps = args['<napp>']
                mgr.reload(napps)

            LOG.info('\tReloaded.')
        except requests.HTTPError as exception:
            if exception.response.status_code != 200:
                msg = json.loads(exception.response.content)
                LOG.error('\tServer error: %s - ', msg['error'])


1		"""Translate cli commands to non-cli code."""
2	1	import json
3	1	import logging
4	1	import os
5	1	import re
6	1	from urllib.error import HTTPError, URLError
7
8	1	import requests
9
10	1	from kytos.utils.exceptions import KytosException
11	1	from kytos.utils.napps import NAppsManager
12
13	1	LOG = logging.getLogger(__name__)
14
15
16	1	class NAppsAPI:
17		"""An API for the command-line interface.
18
19		Use the config file only for required options. Static methods are called
20		by the parser and they instantiate an object of this class to fulfill the
21		request.
22		"""
23
24	1	@classmethod
25		def disable(cls, args):
26		"""Disable subcommand."""
27	1	mgr = NAppsManager()
28
29	1	if args['all']:
30	1	napps = mgr.get_enabled()
31		else:
32	1	napps = args['<napp>']
33
34	1	for napp in napps:
35	1	mgr.set_napp(*napp)
36	1	LOG.info('NApp %s:', mgr.napp_id)
37	1	cls.disable_napp(mgr)
38
39	1	@staticmethod
40		def disable_napp(mgr):
41		"""Disable a NApp."""
42	1	if mgr.is_enabled():
43	1	LOG.info(' Disabling...')
44	1	mgr.disable()
45	1	LOG.info(' Disabled.')
46		else:
47	1	LOG.error(" NApp isn't enabled.")
48
49	1	@classmethod
50		def enable(cls, args):
51		"""Enable subcommand."""
52	1	mgr = NAppsManager()
53
54	1	if args['all']:
55	1	napps = mgr.get_disabled()
56		else:
57	1	napps = args['<napp>']
58
59	1	cls.enable_napps(napps)
60
61	1	@classmethod
62		def enable_napp(cls, mgr):
63		"""Install one NApp using NAppManager object."""
64	1	try:
65	1	if not mgr.is_enabled():
66	1	LOG.info(' Enabling...')
67	1	mgr.enable()
68
69		# Check if NApp is enabled
70	1	if mgr.is_enabled():
71	1	LOG.info(' Enabled.')
72		else:
73	1	LOG.error(' Error enabling NApp.')
74		except (FileNotFoundError, PermissionError) as exception:
75		LOG.error(' %s', exception)
76
77	1	@classmethod
78		def enable_napps(cls, napps):
79		"""Enable a list of NApps.
80
81		Args:
82		napps (list): List of NApps.
83		"""
84	1	mgr = NAppsManager()
85	1	for napp in napps:
86	1	mgr.set_napp(*napp)
87	1	LOG.info('NApp %s:', mgr.napp_id)
88	1	cls.enable_napp(mgr)
89
90	1	@classmethod
91		def create(cls, args): # pylint: disable=unused-argument
92		"""Bootstrap a basic NApp structure on the current folder."""
93	1	NAppsManager.create_napp(meta_package=args.get('--meta', False))
94
95	1	@classmethod
96		def upload(cls, args): # pylint: disable=unused-argument
97		"""Upload the NApp to the NApps server.
98
99		Create the NApp package and upload it to the NApp server.
100		"""
101	1	try:
102	1	NAppsManager().upload()
103		except FileNotFoundError as err:
104		LOG.error("Couldn't find %s in current directory.", err.filename)
105
106	1	@classmethod
107		def uninstall(cls, args):
108		"""Uninstall and delete NApps.
109
110		For local installations, do not delete code outside install_path and
111		enabled_path.
112		"""
113	1	mgr = NAppsManager()
114	1	for napp in args['<napp>']:
115	1	mgr.set_napp(*napp)
116	1	LOG.info('NApp %s:', mgr.napp_id)
117	1	if mgr.is_installed():
118	1	if mgr.is_enabled():
119	1	cls.disable_napp(mgr)
120	1	LOG.info(' Uninstalling...')
121	1	mgr.remote_uninstall()
122	1	LOG.info(' Uninstalled.')
123		else:
124		LOG.error(" NApp isn't installed.")
125
126	1	@classmethod
127		def install(cls, args):
128		"""Install local or remote NApps."""
129	1	cls.install_napps(args['<napp>'])
130
131	1	@classmethod
132		def install_napps(cls, napps):
133		"""Install local or remote NApps.
134
135		This method is recursive, it will install each napps and your
136		dependencies.
137		"""
138	1	mgr = NAppsManager()
139	1	for napp in napps:
140	1	mgr.set_napp(*napp)
141	1	LOG.info(' NApp %s:', mgr.napp_id)
142
143	1	try:
144	1	if not mgr.is_installed():
145		# Try to install all NApps, even if
146		# some of them fail.
147	1	cls.install_napp(mgr)
148
149		# Enable the NApp
150	1	if not mgr.is_enabled():
151		cls.enable_napp(mgr)
152		napp_dependencies = mgr.dependencies()
153		if napp_dependencies:
154		LOG.info('Installing Dependencies:')
155		cls.install_napps(napp_dependencies)
156		else:
157	1	LOG.info(' Enabled.')
158		else:
159		LOG.warning(' Napp already installed.')
160		except KytosException:
161		LOG.error('Error installing NApp.')
162		continue
163
164	1	@classmethod
165		def install_napp(cls, mgr):
166		"""Install a NApp.
167
168		Raises:
169		KytosException: If a NApp hasn't been found.
170
171		"""
172	1	LOG.info(' Downloading from NApps Server...')
173	1	try:
174	1	mgr.remote_install()
175	1	LOG.info(' Downloaded and installed.')
176	1	return
177	1	except HTTPError as exception:
178	1	if exception.code == 404:
179	1	LOG.error(' NApp not found.')
180	1	LOG.info(" If you are trying to install a local NApp, "
181		"use the command 'python setup.py develop' "
182		"inside the 'user/napp' directory.")
183	1	elif exception.code == 400:
184	1	LOG.error(' NApps Server error: %s', exception)
185		except URLError as exception:
186		LOG.error(' NApps Server error: %s', str(exception.reason))
187	1	raise KytosException("NApp not found.")
188
189	1	@classmethod
190		def search(cls, args):
191		"""Search for NApps in NApps server matching a pattern."""
192	1	safe_shell_pat = re.escape(args['<pattern>']).replace(r'\', '.')
193	1	pat_str = '.{}.'.format(safe_shell_pat)
194	1	pattern = re.compile(pat_str, re.IGNORECASE)
195	1	remote_json = NAppsManager.search(pattern)
196	1	remote = set()
197	1	for napp in remote_json:
198		# WARNING: This will be changed in future versions, when 'author'
199		# will be removed.
200	1	username = napp.get('username', napp.get('author'))
201	1	remote.add(((username, napp.get('name')), napp.get('description')))
202
203	1	cls._print_napps(remote)
204
205	1	@classmethod
206		def _print_napps(cls, napp_list):
207		"""Format the NApp list to be printed."""
208	1	mgr = NAppsManager()
209	1	enabled = mgr.get_enabled()
210	1	installed = mgr.get_installed()
211	1	napps = []
212	1	for napp, desc in sorted(napp_list):
213	1	status = 'i' if napp in installed else '-'
214	1	status += 'e' if napp in enabled else '-'
215	1	status = '[{}]'.format(status)
216	1	name = '{}/{}'.format(*napp)
217	1	napps.append((status, name, desc))
218	1	cls.print_napps(napps)
219
220	1	@classmethod
221		def list(cls, args): # pylint: disable=unused-argument
222		"""List all installed NApps and inform whether they are enabled."""
223	1	mgr = NAppsManager()
224
225		# Add status
226	1	napps = [napp + ('[ie]',) for napp in mgr.get_enabled()]
227	1	napps += [napp + ('[i-]',) for napp in mgr.get_disabled()]
228
229		# Sort, add description and reorder columns
230	1	napps.sort()
231	1	napps_ordered = []
232	1	for user, name, status in napps:
233	1	description = mgr.get_description(user, name)
234	1	version = mgr.get_version(user, name)
235	1	napp_id = f'{user}/{name}'
236	1	if version:
237	1	napp_id += f':{version}'
238
239	1	napps_ordered.append((status, napp_id, description))
240
241	1	cls.print_napps(napps_ordered)
242
243	1	@staticmethod
244		def print_napps(napps):
245		"""Print status, name and description."""
246	1	if not napps:
247		print('No NApps found.')
248		return
249
250	1	stat_w = 6 # We already know the size of Status col
251	1	name_w = max(len(n[1]) for n in napps)
252	1	desc_w = max(len(n[2]) for n in napps)
253	1	term_w = os.popen('stty size', 'r').read().split()[1]
254	1	remaining = max(0, int(term_w) - stat_w - name_w - 6)
255	1	desc_w = min(desc_w, remaining)
256	1	widths = (stat_w, name_w, desc_w)
257
258	1	header = '\n{:^%d} \| {:^%d} \| {:^%d}' % widths
259	1	row = '{:^%d} \| {:<%d} \| {:<%d}' % widths
260	1	print(header.format('Status', 'NApp ID', 'Description'))
261	1	print('=+='.join('=' * w for w in widths))
262	1	for user, name, desc in napps:
263	1	desc = (desc[:desc_w - 3] + '...') if len(desc) > desc_w else desc
264	1	print(row.format(user, name, desc))
265
266	1	print('\nStatus: (i)nstalled, (e)nabled\n')
267
268	1	@staticmethod
269		def delete(args):
270		"""Delete NApps from server."""
271	1	mgr = NAppsManager()
272	1	for napp in args['<napp>']:
273	1	mgr.set_napp(*napp)
274	1	LOG.info('Deleting NApp %s from server...', mgr.napp_id)
275	1	try:
276	1	mgr.delete()
277	1	LOG.info(' Deleted.')
278	1	except requests.HTTPError as exception:
279	1	if exception.response.status_code == 405:
280	1	LOG.error('Delete Napp is not allowed yet.')
281		else:
282	1	msg = json.loads(exception.response.content)
283	1	LOG.error(' Server error: %s - ', msg['error'])
284
285	1	@classmethod
286		def prepare(cls, args): # pylint: disable=unused-argument
287		"""Create OpenAPI v3.0 spec skeleton."""
288	1	mgr = NAppsManager()
289	1	mgr.prepare()
290
291	1	@classmethod
292		def reload(cls, args):
293		"""Reload NApps code."""
294	1	LOG.info('Reloading NApps...')
295	1	mgr = NAppsManager()
296
297	1	try:
298	1	if args['all']:
299	1	mgr.reload(None)
300		else:
301	1	napps = args['<napp>']
302	1	mgr.reload(napps)
303
304	1	LOG.info('\tReloaded.')
305	1	except requests.HTTPError as exception:
306	1	if exception.response.status_code != 200:
307	1	msg = json.loads(exception.response.content)
308		LOG.error('\tServer error: %s - ', msg['error'])
309

kytos / kytos-utils

NAppsAPI.print_napps() A last analyzed 2021-05-31 20:20 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

NAppsAPI.print_napps() A
last analyzed 2021-05-31 20:20 UTC