commit
a678241a70
2 changed files with 235 additions and 0 deletions
128
bumblebee_status/modules/contrib/network.py
Normal file
128
bumblebee_status/modules/contrib/network.py
Normal file
|
@ -0,0 +1,128 @@
|
|||
"""
|
||||
A module to show the currently active network connection (ethernet or wifi) and connection strength if the connection is wireless.
|
||||
|
||||
Requires the Python netifaces package and iw installed on Linux.
|
||||
|
||||
A simpler take on nic and network_traffic. No extra config necessary!
|
||||
|
||||
"""
|
||||
|
||||
|
||||
import util.cli
|
||||
import util.format
|
||||
|
||||
import core.module
|
||||
import core.widget
|
||||
import core.input
|
||||
|
||||
import netifaces
|
||||
import socket
|
||||
|
||||
|
||||
class Module(core.module.Module):
|
||||
@core.decorators.every(seconds=5)
|
||||
def __init__(self, config, theme):
|
||||
super().__init__(config, theme, core.widget.Widget(self.network))
|
||||
self.__is_wireless = False
|
||||
self.__is_connected = False
|
||||
self.__interface = None
|
||||
self.__message = None
|
||||
self.__signal = -110
|
||||
|
||||
# Get network information to display to the user
|
||||
def network(self, widgets):
|
||||
# Determine whether there is an internet connection
|
||||
self.__is_connected = self.__attempt_connection()
|
||||
|
||||
# Attempt to extract a valid network interface device
|
||||
try:
|
||||
self.__interface = netifaces.gateways()["default"][netifaces.AF_INET][1]
|
||||
except Exception:
|
||||
self.__interface = None
|
||||
|
||||
# Check to see if the interface (if connected to the internet) is wireless
|
||||
if self.__is_connected and self.__interface:
|
||||
self.__is_wireless = self.__interface_is_wireless(self.__interface)
|
||||
|
||||
# setup message to send to the user
|
||||
if not self.__is_connected or not self.__interface:
|
||||
self.__message = "No connection"
|
||||
elif not self.__is_wireless:
|
||||
# Assuming that if user is connected via non-wireless means that it will be ethernet
|
||||
self.__signal = -30
|
||||
self.__message = "Ethernet"
|
||||
else:
|
||||
# We have a wireless connection
|
||||
iw_dat = util.cli.execute("iwgetid")
|
||||
has_ssid = "ESSID" in iw_dat
|
||||
signal = self.__compute_signal(self.__interface)
|
||||
|
||||
# If signal is None, that means that we can't compute the default interface's signal strength
|
||||
self.__signal = (
|
||||
util.format.asint(signal, minimum=-110, maximum=-30) if signal else None
|
||||
)
|
||||
|
||||
ssid = (
|
||||
iw_dat[iw_dat.index(":") + 1 :].replace('"', "").strip()
|
||||
if has_ssid
|
||||
else "Unknown"
|
||||
)
|
||||
self.__message = self.__generate_wireles_message(ssid, self.__signal)
|
||||
|
||||
return self.__message
|
||||
|
||||
# State determined by signal strength
|
||||
def state(self, widget):
|
||||
if self.__compute_strength(self.__signal) < 50:
|
||||
return "critical"
|
||||
if self.__compute_strength(self.__signal) < 75:
|
||||
return "warning"
|
||||
|
||||
return None
|
||||
|
||||
# manually done for better granularity / ease of parsing strength data
|
||||
def __generate_wireles_message(self, ssid, signal):
|
||||
computed_strength = self.__compute_strength(signal)
|
||||
strength_str = str(computed_strength) if computed_strength else "?"
|
||||
|
||||
return "{} {}%".format(ssid, strength_str)
|
||||
|
||||
def __compute_strength(self, signal):
|
||||
return int(100 * ((signal + 100) / 70.0)) if signal else None
|
||||
|
||||
# get signal strength in decibels/milliwat
|
||||
def __compute_signal(self, interface):
|
||||
# Get connection strength
|
||||
cmd = "iwconfig {}".format(interface)
|
||||
config_dat = " ".join(util.cli.execute(cmd).split())
|
||||
config_tokens = config_dat.replace("=", " ").split()
|
||||
|
||||
# handle weird output
|
||||
try:
|
||||
signal = config_tokens[config_tokens.index("level") + 1]
|
||||
except Exception:
|
||||
signal = None
|
||||
|
||||
return signal
|
||||
|
||||
def __attempt_connection(self):
|
||||
can_connect = False
|
||||
try:
|
||||
socket.create_connection(("1.1.1.1", 53))
|
||||
can_connect = True
|
||||
except Exception:
|
||||
can_connect = False
|
||||
|
||||
return can_connect
|
||||
|
||||
def __interface_is_wireless(self, interface):
|
||||
is_wireless = False
|
||||
try:
|
||||
with open("/proc/net/wireless", "r") as f:
|
||||
is_wireless = interface in f.read()
|
||||
f.close()
|
||||
except Exception:
|
||||
is_wireless = False
|
||||
|
||||
return is_wireless
|
||||
|
107
tests/modules/contrib/test_network.py
Normal file
107
tests/modules/contrib/test_network.py
Normal file
|
@ -0,0 +1,107 @@
|
|||
from unittest import TestCase, mock
|
||||
import pytest
|
||||
|
||||
import core.config
|
||||
import core.widget
|
||||
import modules.contrib.network
|
||||
|
||||
import socket
|
||||
|
||||
pytest.importorskip("netifaces")
|
||||
|
||||
|
||||
def build_module():
|
||||
config = core.config.Config([])
|
||||
return modules.contrib.network.Module(config=config, theme=None)
|
||||
|
||||
|
||||
def wireless_default():
|
||||
return {"default": {1: ("10.0.1.12", "wlan3")}}
|
||||
|
||||
|
||||
def wired_default():
|
||||
return {"default": {18: ("10.0.1.12", "eth3")}}
|
||||
|
||||
|
||||
def exec_side_effect_valid(*args, **kwargs):
|
||||
if args[0] == "iwgetid":
|
||||
return "ESSID: bumblefoo"
|
||||
if "iwconfig" in args[0]:
|
||||
return "level=-30"
|
||||
return mock.DEFAULT
|
||||
|
||||
|
||||
def exec_side_effect_invalid(*args, **kwargs):
|
||||
return "invalid gibberish, can't parse for info"
|
||||
|
||||
|
||||
class TestNetworkUnit(TestCase):
|
||||
def test_load_module(self):
|
||||
__import__("modules.contrib.network")
|
||||
|
||||
@pytest.mark.allow_hosts(["127.0.0.1"])
|
||||
def test_no_internet(self):
|
||||
module = build_module()
|
||||
assert module.widgets()[0].full_text() == "No connection"
|
||||
|
||||
@mock.patch("util.cli.execute")
|
||||
@mock.patch("netifaces.gateways")
|
||||
@mock.patch("socket.create_connection")
|
||||
@mock.patch("netifaces.AF_INET", 1)
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data="wlan3"))
|
||||
def test_valid_wireless_connection(self, socket_mock, gateways_mock, execute_mock):
|
||||
socket_mock.return_value = mock.MagicMock()
|
||||
fake_ssid = "bumblefoo"
|
||||
gateways_mock.return_value = wireless_default()
|
||||
execute_mock.side_effect = exec_side_effect_valid
|
||||
|
||||
module = build_module()
|
||||
|
||||
assert fake_ssid in module.widgets()[0].full_text()
|
||||
|
||||
@mock.patch("netifaces.gateways")
|
||||
@mock.patch("socket.create_connection")
|
||||
@mock.patch("netifaces.AF_INET", 18)
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data="wlan3"))
|
||||
def test_valid_wired_connection(self, socket_mock, gateways_mock):
|
||||
gateways_mock.return_value = wired_default()
|
||||
socket_mock.return_value = mock.MagicMock()
|
||||
|
||||
module = build_module()
|
||||
|
||||
assert module.widgets()[0].full_text() == "Ethernet"
|
||||
|
||||
@mock.patch("netifaces.gateways")
|
||||
@mock.patch("socket.create_connection")
|
||||
def test_invalid_gateways(self, socket_mock, gateways_mock):
|
||||
socket_mock.return_value = mock.Mock()
|
||||
gateways_mock.return_value = {"xyz": "abc"}
|
||||
|
||||
module = build_module()
|
||||
assert module.widgets()[0].full_text() == "No connection"
|
||||
|
||||
@mock.patch("util.cli.execute")
|
||||
@mock.patch("socket.create_connection")
|
||||
@mock.patch("netifaces.gateways")
|
||||
@mock.patch("netifaces.AF_INET", 1)
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data="wlan3"))
|
||||
def test_invalid_execs(self, gateways_mock, socket_mock, execute_mock):
|
||||
execute_mock.side_effect = exec_side_effect_invalid
|
||||
socket_mock.return_value = mock.MagicMock()
|
||||
gateways_mock.return_value = wireless_default()
|
||||
|
||||
module = build_module()
|
||||
|
||||
assert module.widgets()[0].full_text() == "Unknown ?%"
|
||||
|
||||
@mock.patch("builtins.open", **{"return_value.raiseError.side_effect": Exception()})
|
||||
@mock.patch("socket.create_connection")
|
||||
@mock.patch("netifaces.gateways")
|
||||
@mock.patch("netifaces.AF_INET", 18)
|
||||
@mock.patch("builtins.open", mock.mock_open(read_data="wlan3"))
|
||||
def test_no_wireless_file(self, gateways_mock, socket_mock, mock_open):
|
||||
gateways_mock.return_value = wired_default()
|
||||
socket_mock.return_value = mock.MagicMock()
|
||||
module = build_module()
|
||||
|
||||
assert module.widgets()[0].full_text() == "Ethernet"
|
Loading…
Reference in a new issue