2356

pending completion

Build # 2356

Build Type

push

travis-ci

Committed by

nicolargo

Commit Message

<a href="https://github.com/nicolargo/glances/commit/<a class=hub.com/nicolargo/glances/commit/<a class="double-link" href="https://git"><a class=hub.com/nicolargo/glances/commit/<a class="double-link" href="https://git"><a class=hub.com/nicolargo/glances/commit/<a class="double-link" href="https://git"><a class=hub.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685">233ebc960<a href="https://github.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/nicolargo/glances/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/nicolargo/glances/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685&quot;&gt;233ebc960&lt;/a&gt;&lt;a href=&quot;https://github.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685&amp;quot;&amp;gt;&amp;amp;quot;&amp;amp;gt;Correct Travis IC

(cherry picked from commit &amp;amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/nicolargo/glances/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; h</a>ref="https://github.com/nicolargo/glances/commit/<a class="double-link" href="https://github.com/nicolargo/glances/commit/5b3b7984e7df449d8b18d24033d1e0b1636e6e0f">5b3b7984e</a>">5b3b7984e</a>">5b3b7984e</a><a href="https://github.com/nicolargo/glances/commit/233ebc9608a7b193f2d93e8b4068ad53aa4e7685">)

Run Details

2109 of 5581 relevant lines covered (37.79%)

1.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.21

/glances/plugins/glances_ip.py

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2017 Nicolargo <nicolas@nicolargo.com>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""IP plugin."""

import threading
from json import loads

from glances.compat import iterkeys, urlopen, queue
from glances.globals import BSD
from glances.logger import logger
from glances.timer import Timer
from glances.plugins.glances_plugin import GlancesPlugin

# XXX *BSDs: Segmentation fault (core dumped)
# -- https://bitbucket.org/al45tair/netifaces/issues/15
# Also used in the ports_list script
if not BSD:
    try:
        import netifaces
        netifaces_tag = True
    except ImportError:
        netifaces_tag = False
else:
    netifaces_tag = False

# List of online services to retreive public IP address
# List of tuple (url, json, key)
# - url: URL of the Web site
# - json: service return a JSON (True) or string (False)
# - key: key of the IP addresse in the JSON structure
urls = [('http://ip.42.pl/raw', False, None),
        ('http://httpbin.org/ip', True, 'origin'),
        ('http://jsonip.com', True, 'ip'),
        ('https://api.ipify.org/?format=json', True, 'ip')]


class Plugin(GlancesPlugin):

    """Glances IP Plugin.

    stats is a dict
    """

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

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

        # Get the public IP address once
        self.public_address = PublicIpAddress().get()

        # Init the stats
        self.reset()

    def reset(self):
        """Reset/init the stats."""
        self.stats = {}

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update IP stats using the input method.

        Stats is dict
        """
        # Reset stats
        self.reset()

        if self.input_method == 'local' and netifaces_tag:
            # Update stats using the netifaces lib
            try:
                default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
            except (KeyError, AttributeError) as e:
                logger.debug("Cannot grab the default gateway ({})".format(e))
            else:
                try:
                    self.stats['address'] = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']
                    self.stats['mask'] = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']
                    self.stats['mask_cidr'] = self.ip_to_cidr(self.stats['mask'])
                    self.stats['gateway'] = netifaces.gateways()['default'][netifaces.AF_INET][0]
                    # !!! SHOULD be done once, not on each refresh
                    self.stats['public_address'] = self.public_address
                except (KeyError, AttributeError) as e:
                    logger.debug("Cannot grab IP information: {}".format(e))
        elif self.input_method == 'snmp':
            # Not implemented yet
            pass

        return self.stats

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super(Plugin, self).update_views()

        # Add specifics informations
        # Optional
        for key in iterkeys(self.stats):
            self.views[key]['optional'] = True

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

        # Only process if stats exist and display plugin enable...
        if not self.stats or self.is_disable():
            return ret

        # Build the string message
        msg = ' - '
        ret.append(self.curse_add_line(msg))
        msg = 'IP '
        ret.append(self.curse_add_line(msg, 'TITLE'))
        msg = '{}'.format(self.stats['address'])
        ret.append(self.curse_add_line(msg))
        if 'mask_cidr' in self.stats:
            # VPN with no internet access (issue #842)
            msg = '/{}'.format(self.stats['mask_cidr'])
            ret.append(self.curse_add_line(msg))
        try:
            msg_pub = '{}'.format(self.stats['public_address'])
        except UnicodeEncodeError:
            pass
        else:
            if self.stats['public_address'] is not None:
                msg = ' Pub '
                ret.append(self.curse_add_line(msg, 'TITLE'))
                ret.append(self.curse_add_line(msg_pub))

        return ret

    @staticmethod
    def ip_to_cidr(ip):
        """Convert IP address to CIDR.

        Example: '255.255.255.0' will return 24
        """
        return sum([int(x) << 8 for x in ip.split('.')]) // 8128


class PublicIpAddress(object):
    """Get public IP address from online services"""

    def __init__(self, timeout=2):
        self.timeout = timeout

    def get(self):
        """Get the first public IP address returned by one of the online services"""
        q = queue.Queue()

        for u, j, k in urls:
            t = threading.Thread(target=self._get_ip_public, args=(q, u, j, k))
            t.daemon = True
            t.start()

        timer = Timer(self.timeout)
        ip = None
        while not timer.finished() and ip is None:
            if q.qsize() > 0:
                ip = q.get()

        return ip

    def _get_ip_public(self, queue_target, url, json=False, key=None):
        """Request the url service and put the result in the queue_target"""
        try:
            response = urlopen(url, timeout=self.timeout).read().decode('utf-8')
        except Exception as e:
            logger.debug("IP plugin - Cannot open URL {} ({})".format(url, e))
            queue_target.put(None)
        else:
            # Request depend on service
            try:
                if not json:
                    queue_target.put(response)
                else:
                    queue_target.put(loads(response)[key])
            except ValueError:
                queue_target.put(None)

1	# -- coding: utf-8 --
2	#
3	# This file is part of Glances.
4	#
5	# Copyright (C) 2017 Nicolargo <nicolas@nicolargo.com>
6	#
7	# Glances is free software; you can redistribute it and/or modify
8	# it under the terms of the GNU Lesser General Public License as published by
9	# the Free Software Foundation, either version 3 of the License, or
10	# (at your option) any later version.
11	#
12	# Glances is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU Lesser General Public License for more details.
16	#
17	# You should have received a copy of the GNU Lesser General Public License
18	# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20	"""IP plugin."""	5✔
21
22	import threading	5✔
23	from json import loads	5✔
24
25	from glances.compat import iterkeys, urlopen, queue	5✔
26	from glances.globals import BSD	5✔
27	from glances.logger import logger	5✔
28	from glances.timer import Timer	5✔
29	from glances.plugins.glances_plugin import GlancesPlugin	5✔
30
31	# XXX *BSDs: Segmentation fault (core dumped)
32	# -- https://bitbucket.org/al45tair/netifaces/issues/15
33	# Also used in the ports_list script
34	if not BSD:	5✔
35	try:	5✔
36	import netifaces	5✔
37	netifaces_tag = True	×
38	except ImportError:
39	netifaces_tag = False
40	else:
41	netifaces_tag = False	×
42
43	# List of online services to retreive public IP address
44	# List of tuple (url, json, key)
45	# - url: URL of the Web site
46	# - json: service return a JSON (True) or string (False)
47	# - key: key of the IP addresse in the JSON structure
48	urls = [('http://ip.42.pl/raw', False, None),	5✔
49	('http://httpbin.org/ip', True, 'origin'),
50	('http://jsonip.com', True, 'ip'),
51	('https://api.ipify.org/?format=json', True, 'ip')]
52
53
54	class Plugin(GlancesPlugin):	5✔
55
56	"""Glances IP Plugin.
57
58	stats is a dict
59	"""
60
61	def __init__(self, args=None):	5✔
62	"""Init the plugin."""
63	super(Plugin, self).__init__(args=args)	5✔
64
65	# We want to display the stat in the curse interface
66	self.display_curse = True	5✔
67
68	# Get the public IP address once
69	self.public_address = PublicIpAddress().get()	5✔
70
71	# Init the stats
72	self.reset()	5✔
73
74	def reset(self):	5✔
75	"""Reset/init the stats."""
76	self.stats = {}	5✔
77
78	@GlancesPlugin._check_decorator	5✔
79	@GlancesPlugin._log_result_decorator	5✔
80	def update(self):
81	"""Update IP stats using the input method.
82
83	Stats is dict
84	"""
85	# Reset stats
86	self.reset()	5✔
87
88	if self.input_method == 'local' and netifaces_tag:	5✔
89	# Update stats using the netifaces lib
90	try:	×
91	default_gw = netifaces.gateways()['default'][netifaces.AF_INET]	×
92	except (KeyError, AttributeError) as e:	×
93	logger.debug("Cannot grab the default gateway ({})".format(e))	×
94	else:
95	try:	×
96	self.stats['address'] = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']	×
97	self.stats['mask'] = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']	×
98	self.stats['mask_cidr'] = self.ip_to_cidr(self.stats['mask'])	×
99	self.stats['gateway'] = netifaces.gateways()['default'][netifaces.AF_INET][0]	×
100	# !!! SHOULD be done once, not on each refresh
101	self.stats['public_address'] = self.public_address	×
102	except (KeyError, AttributeError) as e:	×
103	logger.debug("Cannot grab IP information: {}".format(e))	×
104	elif self.input_method == 'snmp':	5✔
105	# Not implemented yet
106	pass	×
107
108	return self.stats	5✔
109
110	def update_views(self):	5✔
111	"""Update stats views."""
112	# Call the father's method
113	super(Plugin, self).update_views()	5✔
114
115	# Add specifics informations
116	# Optional
117	for key in iterkeys(self.stats):	5✔
118	self.views[key]['optional'] = True	×
119
120	def msg_curse(self, args=None):	5✔
121	"""Return the dict to display in the curse interface."""
122	# Init the return message
123	ret = []	×
124
125	# Only process if stats exist and display plugin enable...
126	if not self.stats or self.is_disable():	×
127	return ret	×
128
129	# Build the string message
130	msg = ' - '	×
131	ret.append(self.curse_add_line(msg))	×
132	msg = 'IP '	×
133	ret.append(self.curse_add_line(msg, 'TITLE'))	×
134	msg = '{}'.format(self.stats['address'])	×
135	ret.append(self.curse_add_line(msg))	×
136	if 'mask_cidr' in self.stats:	×
137	# VPN with no internet access (issue #842)
138	msg = '/{}'.format(self.stats['mask_cidr'])	×
139	ret.append(self.curse_add_line(msg))	×
140	try:	×
141	msg_pub = '{}'.format(self.stats['public_address'])	×
142	except UnicodeEncodeError:	×
143	pass	×
144	else:
145	if self.stats['public_address'] is not None:	×
146	msg = ' Pub '	×
147	ret.append(self.curse_add_line(msg, 'TITLE'))	×
148	ret.append(self.curse_add_line(msg_pub))	×
149
150	return ret	×
151
152	@staticmethod	5✔
153	def ip_to_cidr(ip):
154	"""Convert IP address to CIDR.
155
156	Example: '255.255.255.0' will return 24
157	"""
158	return sum([int(x) << 8 for x in ip.split('.')]) // 8128	×
159
160
161	class PublicIpAddress(object):	5✔
162	"""Get public IP address from online services"""
163
164	def __init__(self, timeout=2):	5✔
165	self.timeout = timeout	5✔
166
167	def get(self):	5✔
168	"""Get the first public IP address returned by one of the online services"""
169	q = queue.Queue()	5✔
170
171	for u, j, k in urls:	5✔
172	t = threading.Thread(target=self._get_ip_public, args=(q, u, j, k))	5✔
173	t.daemon = True	5✔
174	t.start()	5✔
175
176	timer = Timer(self.timeout)	5✔
177	ip = None	5✔
178	while not timer.finished() and ip is None:	5✔
179	if q.qsize() > 0:	5✔
180	ip = q.get()	5✔
181
182	return ip	5✔
183
184	def _get_ip_public(self, queue_target, url, json=False, key=None):	5✔
185	"""Request the url service and put the result in the queue_target"""
186	try:	5✔
187	response = urlopen(url, timeout=self.timeout).read().decode('utf-8')	5✔
188	except Exception as e:	×
189	logger.debug("IP plugin - Cannot open URL {} ({})".format(url, e))	×
190	queue_target.put(None)	×
191	else:
192	# Request depend on service
193	try:	5✔
194	if not json:	5✔
195	queue_target.put(response)	5✔
196	else:
197	queue_target.put(loads(response)[key])	5✔
198	except ValueError:	×
199	queue_target.put(None)	×

nicolargo / glances / 2356

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous