NAppsAPI.disable_napp() - Code Metrics - Inspection of "Merge pull request #109 from macartur/handle_multi..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 327bd3...378eb3 )

by Carlos Eduardo

created 2017-08-03 02:53 UTC

NAppsAPI.disable_napp() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
c	0
b	0
f	0
dl	0
loc	7
rs	9.4285

"""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()

        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.')

    @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()
            log.info('  Enabled.')
            cls.enable_napps(mgr.dependencies())
        except (FileNotFoundError, PermissionError) as e:
            log.error('  %s', e)

    @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):
        """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."""
        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)
            if not mgr.is_installed():
                cls.install_napp(mgr)
            else:
                log.info('  Installed.')
                cls.enable_napp(mgr)
            napp_dependencies = mgr.dependencies()
            if napp_dependencies:
                log.info('Installing Dependencies:')
                cls.install_napps(napp_dependencies)

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

kytos / kytos-utils

Push — master ( 327bd3...378eb3 )

NAppsAPI.disable_napp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like