[publicip] fix tests and bugs
This commit is contained in:
parent
61ebc3aea6
commit
a78403d3e8
2 changed files with 61 additions and 55 deletions
|
@ -38,50 +38,14 @@ import util.format
|
||||||
import util.location
|
import util.location
|
||||||
|
|
||||||
|
|
||||||
def update_publicip_information(module):
|
|
||||||
widget = module.widget()
|
|
||||||
__previous_default_route = None
|
|
||||||
__current_default_route = None
|
|
||||||
__interval = 5 # Interval between default route change checks
|
|
||||||
|
|
||||||
while True:
|
|
||||||
__current_default_route = netifaces.gateways()["default"][2]
|
|
||||||
|
|
||||||
# Updates public ip information if a change to default route is detected
|
|
||||||
if __current_default_route != __previous_default_route:
|
|
||||||
# Sets __previous_default_route in preparation for next change check
|
|
||||||
__previous_default_route = __current_default_route
|
|
||||||
|
|
||||||
# Refresh location information
|
|
||||||
util.location.reset()
|
|
||||||
|
|
||||||
# Fetch fresh location information
|
|
||||||
__info = util.location.location_info()
|
|
||||||
|
|
||||||
# Contstruct coordinates string
|
|
||||||
__lat = "{:.2f}".format(__info["latitude"])
|
|
||||||
__lon = "{:.2f}".format(__info["longitude"])
|
|
||||||
__coords = __lat + "°N" + "," + " " + __lon + "°E"
|
|
||||||
|
|
||||||
# Set widget values
|
|
||||||
widget.set("public_ip", __info["public_ip"])
|
|
||||||
widget.set("country_name", __info["country"])
|
|
||||||
widget.set("country_code", __info["country_code"])
|
|
||||||
widget.set("city_name", __info["city_name"])
|
|
||||||
widget.set("coordinates", __coords)
|
|
||||||
|
|
||||||
# Update widget values
|
|
||||||
core.event.trigger("update", [widget.module.id], redraw_only=True)
|
|
||||||
|
|
||||||
# Wait __interval seconds before checking for default route changes again
|
|
||||||
time.sleep(__interval)
|
|
||||||
|
|
||||||
class Module(core.module.Module):
|
class Module(core.module.Module):
|
||||||
@core.decorators.every(minutes=60)
|
@core.decorators.every(minutes=60)
|
||||||
def __init__(self, config, theme):
|
def __init__(self, config, theme):
|
||||||
super().__init__(config, theme, core.widget.Widget(self.publicip))
|
super().__init__(config, theme, core.widget.Widget(self.publicip))
|
||||||
|
|
||||||
self.__thread = None
|
self.__previous_default_route = None
|
||||||
|
self.__current_default_route = None
|
||||||
|
self.background = True
|
||||||
|
|
||||||
# Immediate update (override default) when left click on widget
|
# Immediate update (override default) when left click on widget
|
||||||
core.input.register(self, button=core.input.LEFT_MOUSE, cmd=self.__click_update)
|
core.input.register(self, button=core.input.LEFT_MOUSE, cmd=self.__click_update)
|
||||||
|
@ -90,6 +54,8 @@ class Module(core.module.Module):
|
||||||
self._format = self.parameter("format", "{ip} ({country_code})")
|
self._format = self.parameter("format", "{ip} ({country_code})")
|
||||||
|
|
||||||
def publicip(self, widget):
|
def publicip(self, widget):
|
||||||
|
if widget.get("public_ip") == None:
|
||||||
|
return "n/a"
|
||||||
return self._format.format(
|
return self._format.format(
|
||||||
ip=widget.get("public_ip", "-"),
|
ip=widget.get("public_ip", "-"),
|
||||||
country_name=widget.get("country_name", "-"),
|
country_name=widget.get("country_name", "-"),
|
||||||
|
@ -102,12 +68,38 @@ class Module(core.module.Module):
|
||||||
util.location.reset()
|
util.location.reset()
|
||||||
|
|
||||||
def update(self):
|
def update(self):
|
||||||
if self.__thread is not None and self.__thread.is_alive():
|
widget = self.widget()
|
||||||
return
|
|
||||||
self.__thread = threading.Thread(
|
self.__current_default_route = netifaces.gateways()["default"][2]
|
||||||
target=update_publicip_information, args=(self,)
|
|
||||||
)
|
try:
|
||||||
self.__thread.start()
|
# Updates public ip information if a change to default route is detected
|
||||||
|
if self.__current_default_route != self.__previous_default_route:
|
||||||
|
# Sets __previous_default_route in preparation for next change check
|
||||||
|
self.__previous_default_route = self.__current_default_route
|
||||||
|
|
||||||
|
# Refresh location information
|
||||||
|
util.location.reset()
|
||||||
|
|
||||||
|
# Fetch fresh location information
|
||||||
|
__info = util.location.location_info()
|
||||||
|
|
||||||
|
# Contstruct coordinates string
|
||||||
|
__lat = "{:.2f}".format(__info["latitude"])
|
||||||
|
__lon = "{:.2f}".format(__info["longitude"])
|
||||||
|
__coords = __lat + "°N" + "," + " " + __lon + "°E"
|
||||||
|
|
||||||
|
# Set widget values
|
||||||
|
widget.set("public_ip", __info["public_ip"])
|
||||||
|
widget.set("country_name", __info["country"])
|
||||||
|
widget.set("country_code", __info["country_code"])
|
||||||
|
widget.set("city_name", __info["city_name"])
|
||||||
|
widget.set("coordinates", __coords)
|
||||||
|
|
||||||
|
# Update widget values
|
||||||
|
core.event.trigger("update", [widget.module.id], redraw_only=True)
|
||||||
|
except:
|
||||||
|
widget.set("public_ip", None)
|
||||||
|
|
||||||
def state(self, widget):
|
def state(self, widget):
|
||||||
return widget.get("state", None)
|
return widget.get("state", None)
|
||||||
|
|
|
@ -17,27 +17,41 @@ class PublicIPTest(TestCase):
|
||||||
def test_load_module(self):
|
def test_load_module(self):
|
||||||
__import__("modules.contrib.publicip")
|
__import__("modules.contrib.publicip")
|
||||||
|
|
||||||
@mock.patch('util.location.public_ip')
|
@mock.patch('util.location.location_info')
|
||||||
def test_public_ip(self, public_ip_mock):
|
def test_public_ip(self, location_mock):
|
||||||
public_ip_mock.return_value = '5.12.220.2'
|
location_mock.return_value = {
|
||||||
|
'public_ip': '5.12.220.2',
|
||||||
|
'latitude': 0,
|
||||||
|
'longitude': 2,
|
||||||
|
'country': 'some country',
|
||||||
|
'country_code': 'sc',
|
||||||
|
'city_name': '???',
|
||||||
|
}
|
||||||
|
|
||||||
module = build_module()
|
module = build_module()
|
||||||
module.update()
|
module.update()
|
||||||
|
|
||||||
assert widget(module).full_text() == '5.12.220.2'
|
assert widget(module).full_text() == '5.12.220.2 (sc)'
|
||||||
|
|
||||||
@mock.patch('util.location.public_ip')
|
@mock.patch('util.location.location_info')
|
||||||
def test_public_ip(self, public_ip_mock):
|
def test_public_ip2(self, location_mock):
|
||||||
public_ip_mock.return_value = None
|
location_mock.return_value = {
|
||||||
|
'public_ip': None,
|
||||||
|
'latitude': 0,
|
||||||
|
'longitude': 2,
|
||||||
|
'country': 'some country',
|
||||||
|
'country_code': 'sc',
|
||||||
|
'city_name': '???',
|
||||||
|
}
|
||||||
|
|
||||||
module = build_module()
|
module = build_module()
|
||||||
module.update()
|
module.update()
|
||||||
|
|
||||||
assert widget(module).full_text() == 'n/a'
|
assert widget(module).full_text() == 'n/a'
|
||||||
|
|
||||||
@mock.patch('util.location.public_ip')
|
@mock.patch('util.location.location_info')
|
||||||
def test_public_ip_with_exception(self, public_ip_mock):
|
def test_public_ip_with_exception(self, location_mock):
|
||||||
public_ip_mock.side_effect = Exception
|
location_mock.side_effect = Exception
|
||||||
|
|
||||||
module = build_module()
|
module = build_module()
|
||||||
module.update()
|
module.update()
|
||||||
|
|
Loading…
Reference in a new issue