diff --git a/bumblebee/modules/load.py b/bumblebee/modules/load.py new file mode 100644 index 0000000..8c51e9f --- /dev/null +++ b/bumblebee/modules/load.py @@ -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 diff --git a/tests/modules/test_load.py b/tests/modules/test_load.py new file mode 100644 index 0000000..d8120e2 --- /dev/null +++ b/tests/modules/test_load.py @@ -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