[modules/load] Re-enable load module
Display system load and show warning/critical error when load is above a certain threshold (compared to the number of available CPUs). see #23
This commit is contained in:
parent
4bd13c2f63
commit
c8fc75a401
2 changed files with 90 additions and 0 deletions
43
bumblebee/modules/load.py
Normal file
43
bumblebee/modules/load.py
Normal file
|
@ -0,0 +1,43 @@
|
|||
# pylint: disable=C0111,R0903
|
||||
|
||||
"""Displays system load.
|
||||
|
||||
Parameters:
|
||||
* load.warning : Warning threshold for the one-minute load average (defaults to 70% of the number of CPUs)
|
||||
* load.critical: Critical threshold for the one-minute load average (defaults to 80% of the number of CPUs)
|
||||
"""
|
||||
|
||||
import os
|
||||
import multiprocessing
|
||||
|
||||
import bumblebee.input
|
||||
import bumblebee.output
|
||||
import bumblebee.engine
|
||||
|
||||
class Module(bumblebee.engine.Module):
|
||||
def __init__(self, engine, config):
|
||||
super(Module, self).__init__(engine, config,
|
||||
bumblebee.output.Widget(full_text=self.load)
|
||||
)
|
||||
try:
|
||||
self._cpus = multiprocessing.cpu_count()
|
||||
except multiprocessing.NotImplementedError as e:
|
||||
self._cpus = 1
|
||||
engine.input.register_callback(self, button=bumblebee.input.LEFT_MOUSE,
|
||||
cmd="gnome-system-monitor")
|
||||
|
||||
def load(self):
|
||||
return "{:.02f}/{:.02f}/{:.02f}".format(
|
||||
self._load[0], self._load[1], self._load[2]
|
||||
)
|
||||
|
||||
def update(self, widgets):
|
||||
self._load = os.getloadavg()
|
||||
|
||||
def state(self, widget):
|
||||
if self._load[0] > float(self.parameter("critical", self._cpus*0.8)):
|
||||
return "critical"
|
||||
if self._load[0] > float(self.parameter("warning", self._cpus*0.7)):
|
||||
return "warning"
|
||||
|
||||
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
47
tests/modules/test_load.py
Normal file
47
tests/modules/test_load.py
Normal file
|
@ -0,0 +1,47 @@
|
|||
# pylint: disable=C0103,C0111
|
||||
|
||||
import json
|
||||
import unittest
|
||||
import mock
|
||||
|
||||
import bumblebee.input
|
||||
from bumblebee.input import I3BarInput
|
||||
from bumblebee.modules.load import Module
|
||||
from tests.util import MockEngine, MockConfig, assertStateContains, assertMouseEvent
|
||||
|
||||
class TestLoadModule(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.engine = MockEngine()
|
||||
self.engine.input = I3BarInput()
|
||||
self.engine.input.need_event = True
|
||||
self.config = MockConfig()
|
||||
self.module = Module(engine=self.engine, config={ "config": self.config })
|
||||
|
||||
@mock.patch("select.select")
|
||||
@mock.patch("subprocess.Popen")
|
||||
@mock.patch("sys.stdin")
|
||||
def test_leftclick(self, mock_input, mock_output, mock_select):
|
||||
assertMouseEvent(mock_input, mock_output, mock_select, self.engine,
|
||||
self.module, bumblebee.input.LEFT_MOUSE,
|
||||
"gnome-system-monitor"
|
||||
)
|
||||
|
||||
@mock.patch("multiprocessing.cpu_count")
|
||||
@mock.patch("os.getloadavg")
|
||||
def test_warning(self, mock_loadavg, mock_cpucount):
|
||||
self.config.set("load.critical", "1")
|
||||
self.config.set("load.warning", "0.8")
|
||||
mock_cpucount.return_value = 1
|
||||
mock_loadavg.return_value = [ 0.9, 0, 0 ]
|
||||
assertStateContains(self, self.module, "warning")
|
||||
|
||||
@mock.patch("multiprocessing.cpu_count")
|
||||
@mock.patch("os.getloadavg")
|
||||
def test_critical(self, mock_loadavg, mock_cpucount):
|
||||
self.config.set("load.critical", "1")
|
||||
self.config.set("load.warning", "0.8")
|
||||
mock_cpucount.return_value = 1
|
||||
mock_loadavg.return_value = [ 1.1, 0, 0 ]
|
||||
assertStateContains(self, self.module, "critical")
|
||||
|
||||
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
Loading…
Reference in a new issue