NAppsAPI._print_napps() - Code Metrics - Inspection of "Merge pull request #71 from renanrodrigo/add_json_..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e11d7c...e4c956 )

by Beraldo

created 2017-04-11 19:28 UTC

NAppsAPI._print_napps() A

↳ Parent: NAppsAPI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	14
rs	9.2

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

import requests
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()
        for napp in args['<napp>']:
            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.')

    @classmethod
    def enable(cls, args):
        """Enable subcommand."""
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            cls.enable_napp(mgr)

    @staticmethod
    def enable_napp(mgr):
        """Install one NApp using NAppManager object."""
        try:
            if not mgr.is_enabled():
                log.info('  Enabling...')
                mgr.enable()
            log.info('  Enabled.')
        except (FileNotFoundError, PermissionError) as e:
            log.error('  %s', e)

    @classmethod
    def create(cls, args):
        """Bootstrap a basic NApp structure on the current folder."""
        NAppsManager.create_napp()

    @classmethod
    def upload(cls, args):
        """Upload the NApp to the NApps server.

        Create the NApp package and upload it to the NApp server.
        """
        try:
            NAppsManager().upload()
        except FileNotFoundError:
            log.error("Couldn't find kytos.json in current directory.")

    @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.uninstall()
            log.info('  Uninstalled.')

    @classmethod
    def install(cls, args):
        """Install local or remote NApps."""
        mgr = NAppsManager()
        for napp in args['<napp>']:
            mgr.set_napp(*napp)
            log.info('NApp %s:', mgr.napp_id)
            if not mgr.is_installed():
                cls.install_napp(mgr)
            else:
                log.info('  Installed.')

    @classmethod
    def install_napp(cls, mgr):
        """Install a NApp."""
        try:
            log.info('  Searching local NApp...')
            mgr.install_local()
            log.info('  Found and installed.')
            cls.enable_napp(mgr)
        except FileNotFoundError:
            log.info('  Not found. Downloading from NApps Server...')
            try:
                mgr.install_remote()
                log.info('  Downloaded and installed.')
                cls.enable_napp(mgr)
            except HTTPError as e:
                if e.code == 404:
                    log.error('  NApp not found.')
                else:
                    log.error('  NApps Server error: %s', e)

    @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):
        """List all installed NApps and inform whether they are installed."""
        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 coloumns
        napps.sort()
        napps_ordered = []
        for user, name, status in napps:
            napps_ordered.append((status, '{}/{}'.format(user, name),
                                  mgr.get_description(user, name)))

        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 = 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 e:
                if e.response.status_code == 405:
                    log.error('Delete Napp is not allowed yet.')
                else:
                    msg = json.loads(e.response.content)
                    log.error('  Server error: %s - ', msg['error'])


1			"""Translate cli commands to non-cli code."""
2			import json
3			import logging
4			import os
5			import re
6			from urllib.error import HTTPError
7
8			import requests
9			from kytos.utils.napps import NAppsManager
10
11			log = logging.getLogger(__name__)
12
13
14			class NAppsAPI:
15			"""An API for the command-line interface.
16
17			Use the config file only for required options. Static methods are called
18			by the parser and they instantiate an object of this class to fulfill the
19			request.
20			"""
21
22			@classmethod
23			def disable(cls, args):
24			"""Disable subcommand."""
25			mgr = NAppsManager()
26			for napp in args['<napp>']:
27			mgr.set_napp(*napp)
28			log.info('NApp %s:', mgr.napp_id)
29			cls.disable_napp(mgr)
30
31			@staticmethod
32			def disable_napp(mgr):
33			"""Disable a NApp."""
34			if mgr.is_enabled():
35			log.info(' Disabling...')
36			mgr.disable()
37			log.info(' Disabled.')
38
39			@classmethod
40			def enable(cls, args):
41			"""Enable subcommand."""
42			mgr = NAppsManager()
43			for napp in args['<napp>']:
44			mgr.set_napp(*napp)
45			log.info('NApp %s:', mgr.napp_id)
46			cls.enable_napp(mgr)
47
48			@staticmethod
49			def enable_napp(mgr):
50			"""Install one NApp using NAppManager object."""
51			try:
52			if not mgr.is_enabled():
53			log.info(' Enabling...')
54			mgr.enable()
55			log.info(' Enabled.')
56			except (FileNotFoundError, PermissionError) as e:
57			log.error(' %s', e)
58
59			@classmethod
60			def create(cls, args):
61			"""Bootstrap a basic NApp structure on the current folder."""
62			NAppsManager.create_napp()
63
64			@classmethod
65			def upload(cls, args):
66			"""Upload the NApp to the NApps server.
67
68			Create the NApp package and upload it to the NApp server.
69			"""
70			try:
71			NAppsManager().upload()
72			except FileNotFoundError:
73			log.error("Couldn't find kytos.json in current directory.")
74
75			@classmethod
76			def uninstall(cls, args):
77			"""Uninstall and delete NApps.
78
79			For local installations, do not delete code outside install_path and
80			enabled_path.
81			"""
82			mgr = NAppsManager()
83			for napp in args['<napp>']:
84			mgr.set_napp(*napp)
85			log.info('NApp %s:', mgr.napp_id)
86			if mgr.is_installed():
87			if mgr.is_enabled():
88			cls.disable_napp(mgr)
89			log.info(' Uninstalling...')
90			mgr.uninstall()
91			log.info(' Uninstalled.')
92
93			@classmethod
94			def install(cls, args):
95			"""Install local or remote NApps."""
96			mgr = NAppsManager()
97			for napp in args['<napp>']:
98			mgr.set_napp(*napp)
99			log.info('NApp %s:', mgr.napp_id)
100			if not mgr.is_installed():
101			cls.install_napp(mgr)
102			else:
103			log.info(' Installed.')
104
105			@classmethod
106			def install_napp(cls, mgr):
107			"""Install a NApp."""
108			try:
109			log.info(' Searching local NApp...')
110			mgr.install_local()
111			log.info(' Found and installed.')
112			cls.enable_napp(mgr)
113			except FileNotFoundError:
114			log.info(' Not found. Downloading from NApps Server...')
115			try:
116			mgr.install_remote()
117			log.info(' Downloaded and installed.')
118			cls.enable_napp(mgr)
119			except HTTPError as e:
120			if e.code == 404:
121			log.error(' NApp not found.')
122			else:
123			log.error(' NApps Server error: %s', e)
124
125			@classmethod
126			def search(cls, args):
127			"""Search for NApps in NApps server matching a pattern."""
128			safe_shell_pat = re.escape(args['<pattern>']).replace(r'\', '.')
129			pat_str = '.{}.'.format(safe_shell_pat)
130			pattern = re.compile(pat_str, re.IGNORECASE)
131			remote_json = NAppsManager.search(pattern)
132			remote = set()
133			for napp in remote_json:
134			# WARNING: This will be changed in future versions, when 'author'
135			# will be removed.
136			username = napp.get('username', napp.get('author'))
137			remote.add(((username, napp.get('name')), napp.get('description')))
138
139			cls._print_napps(remote)
140
141			@classmethod
142			def _print_napps(cls, napp_list):
143			"""Format the NApp list to be printed."""
144			mgr = NAppsManager()
145			enabled = mgr.get_enabled()
146			installed = mgr.get_installed()
147			napps = []
148			for napp, desc in sorted(napp_list):
149			status = 'i' if napp in installed else '-'
150			status += 'e' if napp in enabled else '-'
151			status = '[{}]'.format(status)
152			name = '{}/{}'.format(*napp)
153			napps.append((status, name, desc))
154			cls.print_napps(napps)
155
156			@classmethod
157			def list(cls, args):
158			"""List all installed NApps and inform whether they are installed."""
159			mgr = NAppsManager()
160
161			# Add status
162			napps = [napp + ('[ie]',) for napp in mgr.get_enabled()]
163			napps += [napp + ('[i-]',) for napp in mgr.get_disabled()]
164
165			# Sort, add description and reorder coloumns
166			napps.sort()
167			napps_ordered = []
168			for user, name, status in napps:
169			napps_ordered.append((status, '{}/{}'.format(user, name),
170			mgr.get_description(user, name)))
171
172			cls.print_napps(napps_ordered)
173
174			@staticmethod
175			def print_napps(napps):
176			"""Print status, name and description."""
177			if not napps:
178			print('No NApps found.')
179			return
180
181			stat_w = 6 # We already know the size of Status col
182			name_w = max(len(n[1]) for n in napps)
183			desc_w = max(len(n[2]) for n in napps)
184			term_w = os.popen('stty size', 'r').read().split()[1]
185			remaining = int(term_w) - stat_w - name_w - 6
186			desc_w = min(desc_w, remaining)
187			widths = (stat_w, name_w, desc_w)
188
189			header = '\n{:^%d} \| {:^%d} \| {:^%d}' % widths
190			row = '{:^%d} \| {:<%d} \| {:<%d}' % widths
191			print(header.format('Status', 'NApp ID', 'Description'))
192			print('=+='.join('=' * w for w in widths))
193			for user, name, desc in napps:
194			desc = (desc[:desc_w-3] + '...') if len(desc) > desc_w else desc
195			print(row.format(user, name, desc))
196
197			print('\nStatus: (i)nstalled, (e)nabled\n')
198
199			@staticmethod
200			def delete(args):
201			"""Delete NApps from server."""
202			mgr = NAppsManager()
203			for napp in args['<napp>']:
204			mgr.set_napp(*napp)
205			log.info('Deleting NApp %s from server...', mgr.napp_id)
206			try:
207			mgr.delete()
208			log.info(' Deleted.')
209			except requests.HTTPError as e:
210			if e.response.status_code == 405:
211			log.error('Delete Napp is not allowed yet.')
212			else:
213			msg = json.loads(e.response.content)
214			log.error(' Server error: %s - ', msg['error'])
215

kytos / kytos-utils

Push — master ( e11d7c...e4c956 )

NAppsAPI._print_napps() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like