glances.plugins.glances_system.Plugin.update() - Code Metrics - Inspection of "Resolv conflict from master" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( d0fde6...8e443d )

by Nicolas

created 2022-07-28 16:11 UTC

glances.plugins.glances_system.Plugin.update() F

↳ Parent: glances.plugins.glances_system

Complexity

Conditions

Size

Total Lines	67
Code Lines	45

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	18
eloc	45
nop	1
dl	0
loc	67
rs	1.2
c	0
b	0
f	0

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""System plugin."""

import os
import platform
import re
from io import open

from glances.logger import logger
from glances.compat import iteritems
from glances.plugins.glances_plugin import GlancesPlugin

# SNMP OID
snmp_oid = {
    'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
    'netapp': {
        'hostname': '1.3.6.1.2.1.1.5.0',
        'system_name': '1.3.6.1.2.1.1.1.0',
        'platform': '1.3.6.1.4.1.789.1.1.5.0',
    },
}

# SNMP to human read
# Dict (key: OS short name) of dict (reg exp OID to human)
# Windows:
# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
snmp_to_human = {
    'windows': {
        'Windows Version 10.0': 'Windows 10 or Server 2016',
        'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
        'Windows Version 6.2': 'Windows 8 or Server 2012',
        'Windows Version 6.1': 'Windows 7 or Server 2008R2',
        'Windows Version 6.0': 'Windows Vista or Server 2008',
        'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
        'Windows Version 5.1': 'Windows XP',
        'Windows Version 5.0': 'Windows 2000',
    }
}


def _linux_os_release():
    """Try to determine the name of a Linux distribution.

    This function checks for the /etc/os-release file.
    It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
    An empty string is returned if the above values cannot be determined.
    """
    pretty_name = ''
    ashtray = {}
    keys = ['NAME', 'VERSION_ID']
    try:
        with open(os.path.join('/etc', 'os-release')) as f:
            for line in f:
                for key in keys:
                    if line.startswith(key):
                        ashtray[key] = re.sub(r'^"|"$', '', line.strip().split('=')[1])
    except (OSError, IOError):
        return pretty_name

    if ashtray:
        if 'NAME' in ashtray:
            pretty_name = ashtray['NAME']
        if 'VERSION_ID' in ashtray:
            pretty_name += ' {}'.format(ashtray['VERSION_ID'])

    return pretty_name


class Plugin(GlancesPlugin):

    """Glances' host/system plugin.

    stats is a dict
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super(Plugin, self).__init__(args=args, config=config)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Set default rate to 60 seconds
        if self.get_refresh():
            self.set_refresh(60)

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update the host/system info using the input method.

        :return: the stats dict
        """
        # Init new stats
        stats = self.get_init_value()

        if self.input_method == 'local':
            # Update stats using the standard system lib
            stats['os_name'] = platform.system()
            stats['hostname'] = platform.node()
            stats['platform'] = platform.architecture()[0]
            if stats['os_name'] == "Linux":
                try:
                    linux_distro = platform.linux_distribution()
                except AttributeError:
                    stats['linux_distro'] = _linux_os_release()
                else:
                    if linux_distro[0] == '':
                        stats['linux_distro'] = _linux_os_release()
                    else:
                        stats['linux_distro'] = ' '.join(linux_distro[:2])
                stats['os_version'] = platform.release()
            elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
                stats['os_version'] = platform.release()
            elif stats['os_name'] == "Darwin":
                stats['os_version'] = platform.mac_ver()[0]
            elif stats['os_name'] == "Windows":
                os_version = platform.win32_ver()
                stats['os_version'] = ' '.join(os_version[::2])
                # if the python version is 32 bit perhaps the windows operating
                # system is 64bit
                if stats['platform'] == '32bit' and 'PROCESSOR_ARCHITEW6432' in os.environ:
                    stats['platform'] = '64bit'
            else:
                stats['os_version'] = ""
            # Add human readable name
            if stats['os_name'] == "Linux":
                stats['hr_name'] = stats['linux_distro']
            else:
                stats['hr_name'] = '{} {}'.format(stats['os_name'], stats['os_version'])
            stats['hr_name'] += ' {}'.format(stats['platform'])

        elif self.input_method == 'snmp':
            # Update stats using SNMP
            try:
                stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
            except KeyError:
                stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
            # Default behavior: display all the information
            stats['os_name'] = stats['system_name']
            # Windows OS tips
            if self.short_system_name == 'windows':
                for r, v in iteritems(snmp_to_human['windows']):
                    if re.search(r, stats['system_name']):
                        stats['os_name'] = v
                        break
            # Add human readable name
            stats['hr_name'] = stats['os_name']

        # Update the stats
        self.stats = stats

        return self.stats

    def msg_curse(self, args=None, max_width=None):
        """Return the string to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist and plugin not disabled
        if not self.stats or self.is_disabled():
            return ret

        # Build the string message
        if args.client:
            # Client mode
            if args.cs_status.lower() == "connected":
                msg = 'Connected to '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "snmp":
                msg = 'SNMP from '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "disconnected":
                msg = 'Disconnected from '
                ret.append(self.curse_add_line(msg, 'CRITICAL'))

        # Hostname is mandatory
        msg = self.stats['hostname']
        ret.append(self.curse_add_line(msg, "TITLE"))
        # System info
        if self.stats['os_name'] == "Linux" and self.stats['linux_distro']:
            msg = ' ({} {} / {} {})'.format(
                self.stats['linux_distro'], self.stats['platform'], self.stats['os_name'], self.stats['os_version']
            )
        else:
            try:
                msg = ' ({} {} {})'.format(self.stats['os_name'], self.stats['os_version'], self.stats['platform'])
            except Exception:
                msg = ' ({})'.format(self.stats['os_name'])
        ret.append(self.curse_add_line(msg, optional=True))

        # Return the message with decoration
        return ret


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8			#
9
10			"""System plugin."""
11
12			import os
13			import platform
14			import re
15			from io import open
16
17			from glances.logger import logger
18			from glances.compat import iteritems
19			from glances.plugins.glances_plugin import GlancesPlugin
20
21			# SNMP OID
22			snmp_oid = {
23			'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
24			'netapp': {
25			'hostname': '1.3.6.1.2.1.1.5.0',
26			'system_name': '1.3.6.1.2.1.1.1.0',
27			'platform': '1.3.6.1.4.1.789.1.1.5.0',
28			},
29			}
30
31			# SNMP to human read
32			# Dict (key: OS short name) of dict (reg exp OID to human)
33			# Windows:
34			# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
35			snmp_to_human = {
36			'windows': {
37			'Windows Version 10.0': 'Windows 10 or Server 2016',
38			'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
39			'Windows Version 6.2': 'Windows 8 or Server 2012',
40			'Windows Version 6.1': 'Windows 7 or Server 2008R2',
41			'Windows Version 6.0': 'Windows Vista or Server 2008',
42			'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
43			'Windows Version 5.1': 'Windows XP',
44			'Windows Version 5.0': 'Windows 2000',
45			}
46			}
47
48
49			def _linux_os_release():
50			"""Try to determine the name of a Linux distribution.
51
52			This function checks for the /etc/os-release file.
53			It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
54			An empty string is returned if the above values cannot be determined.
55			"""
56			pretty_name = ''
57			ashtray = {}
58			keys = ['NAME', 'VERSION_ID']
59			try:
60			with open(os.path.join('/etc', 'os-release')) as f:
61			for line in f:
62			for key in keys:
63			if line.startswith(key):
64			ashtray[key] = re.sub(r'^"\|"$', '', line.strip().split('=')[1])
65			except (OSError, IOError):
66			return pretty_name
67
68			if ashtray:
69			if 'NAME' in ashtray:
70			pretty_name = ashtray['NAME']
71			if 'VERSION_ID' in ashtray:
72			pretty_name += ' {}'.format(ashtray['VERSION_ID'])
73
74			return pretty_name
75
76
77			class Plugin(GlancesPlugin):
78
79			"""Glances' host/system plugin.
80
81			stats is a dict
82			"""
83
84			def __init__(self, args=None, config=None):
85			"""Init the plugin."""
86			super(Plugin, self).__init__(args=args, config=config)
87
88			# We want to display the stat in the curse interface
89			self.display_curse = True
90
91			# Set default rate to 60 seconds
92			if self.get_refresh():
93			self.set_refresh(60)
94
95			@GlancesPlugin._check_decorator
96			@GlancesPlugin._log_result_decorator
97			def update(self):
98			"""Update the host/system info using the input method.
99
100			:return: the stats dict
101			"""
102			# Init new stats
103			stats = self.get_init_value()
104
105			if self.input_method == 'local':
106			# Update stats using the standard system lib
107			stats['os_name'] = platform.system()
108			stats['hostname'] = platform.node()
109			stats['platform'] = platform.architecture()[0]
110			if stats['os_name'] == "Linux":
111			try:
112			linux_distro = platform.linux_distribution()
113			except AttributeError:
114			stats['linux_distro'] = _linux_os_release()
115			else:
116			if linux_distro[0] == '':
117			stats['linux_distro'] = _linux_os_release()
118			else:
119			stats['linux_distro'] = ' '.join(linux_distro[:2])
120			stats['os_version'] = platform.release()
121			elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
122			stats['os_version'] = platform.release()
123			elif stats['os_name'] == "Darwin":
124			stats['os_version'] = platform.mac_ver()[0]
125			elif stats['os_name'] == "Windows":
126			os_version = platform.win32_ver()
127			stats['os_version'] = ' '.join(os_version[::2])
128			# if the python version is 32 bit perhaps the windows operating
129			# system is 64bit
130			if stats['platform'] == '32bit' and 'PROCESSOR_ARCHITEW6432' in os.environ:
131			stats['platform'] = '64bit'
132			else:
133			stats['os_version'] = ""
134			# Add human readable name
135			if stats['os_name'] == "Linux":
136			stats['hr_name'] = stats['linux_distro']
137			else:
138			stats['hr_name'] = '{} {}'.format(stats['os_name'], stats['os_version'])
139			stats['hr_name'] += ' {}'.format(stats['platform'])
140
141			elif self.input_method == 'snmp':
142			# Update stats using SNMP
143			try:
144			stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
145			except KeyError:
146			stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
147			# Default behavior: display all the information
148			stats['os_name'] = stats['system_name']
149			# Windows OS tips
150			if self.short_system_name == 'windows':
151			for r, v in iteritems(snmp_to_human['windows']):
152			if re.search(r, stats['system_name']):
153			stats['os_name'] = v
154			break
155			# Add human readable name
156			stats['hr_name'] = stats['os_name']
157
158			# Update the stats
159			self.stats = stats
160
161			return self.stats
162
163			def msg_curse(self, args=None, max_width=None):
164			"""Return the string to display in the curse interface."""
165			# Init the return message
166			ret = []
167
168			# Only process if stats exist and plugin not disabled
169			if not self.stats or self.is_disabled():
170			return ret
171
172			# Build the string message
173			if args.client:
174			# Client mode
175			if args.cs_status.lower() == "connected":
176			msg = 'Connected to '
177			ret.append(self.curse_add_line(msg, 'OK'))
178			elif args.cs_status.lower() == "snmp":
179			msg = 'SNMP from '
180			ret.append(self.curse_add_line(msg, 'OK'))
181			elif args.cs_status.lower() == "disconnected":
182			msg = 'Disconnected from '
183			ret.append(self.curse_add_line(msg, 'CRITICAL'))
184
185			# Hostname is mandatory
186			msg = self.stats['hostname']
187			ret.append(self.curse_add_line(msg, "TITLE"))
188			# System info
189			if self.stats['os_name'] == "Linux" and self.stats['linux_distro']:
190			msg = ' ({} {} / {} {})'.format(
191			self.stats['linux_distro'], self.stats['platform'], self.stats['os_name'], self.stats['os_version']
192			)
193			else:
194			try:
195			msg = ' ({} {} {})'.format(self.stats['os_name'], self.stats['os_version'], self.stats['platform'])
196			except Exception:
197			msg = ' ({})'.format(self.stats['os_name'])
198			ret.append(self.curse_add_line(msg, optional=True))
199
200			# Return the message with decoration
201			return ret
202

nicolargo / glances

Push — master ( d0fde6...8e443d )

glances.plugins.glances_system.Plugin.update() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like