kytos.cli.commands.napps.api.NAppsAPI.enable_napps() - Code Metrics - Inspection of "Bump certifi from 2019.11.28 to 2022.12.7 in /requ..." - kytos/kytos-utils - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#324)

unknown

created 2022-12-08 03:13 UTC

NAppsAPI.enable_napps() A

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

Complexity

Conditions

Size

Total Lines	12
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	2
dl	0
loc	12
rs	10
c	0
b	0
f	0
ccs	0
cts	6
cp	0
crap	6

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

kytos / kytos-utils

Pull Request — master (#324)

NAppsAPI.enable_napps() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like