[core] Rework core implementation

Experimental re-implementation of core functionality with the aim:
- Depend only on the Python Standard Library for core
- If modules are missing elsewhere, *never* throw
- Unit test *everything*
- Cleaner and more minimal implementation
- Better integration points for existing implementations (charts,
  braille, etc.)
- Full backwards-compatibility with existing module system (except where
  modules can be vastly simplified)
This commit is contained in:
Tobias Witek 2020-01-19 13:29:34 +01:00
parent 72f88b3409
commit e931bb93c6
214 changed files with 45 additions and 13040 deletions

View file

@ -1,109 +0,0 @@
# pylint: disable=C0103,C0111
import sys
import mock
import unittest
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
import tests.mocks as mocks
from bumblebee.modules.battery import Module
from bumblebee.config import Config
class TestBatteryModule(unittest.TestCase):
def setUp(self):
self._stdout = mock.patch("sys.stdout", new_callable=StringIO)
self._exists = mock.patch("bumblebee.modules.battery.os.path.exists")
self._open = mock.patch("bumblebee.modules.battery.open", create=True)
self.stdout = self._stdout.start()
self.exists = self._exists.start()
self.open = self._open.start()
self.file = mock.Mock()
self.file.__enter__ = lambda x: self.file
self.file.__exit__ = lambda x, a, b, c: ""
self.file.read.return_value = "120"
self.open.return_value = self.file
self.exists.return_value = True
self.engine = mock.Mock()
self.config = Config()
self.config.set("battery.showremaining", "false")
self.module = Module(engine=self.engine, config={"config":self.config})
self.config.set("battery.critical", "20")
self.config.set("battery.warning", "25")
self.criticalValue = "19"
self.warningValue = "21"
self.normalValue = "26"
self.chargedValue = "96"
self.module.widgets()[0]
def tearDown(self):
self._stdout.stop()
self._exists.stop()
self._open.stop()
def test_format(self):
for widget in self.module.widgets():
self.assertEquals(len(widget.full_text()), len("100%"))
def test_critical(self):
self.file.read.return_value = self.criticalValue
self.module.update_all()
self.assertTrue("critical" in self.module.state(self.module.widgets()[0]))
def test_warning(self):
self.file.read.return_value = self.warningValue
self.module.update_all()
self.assertTrue("warning" in self.module.state(self.module.widgets()[0]))
def test_normal(self):
self.file.read.return_value = self.normalValue
self.module.update_all()
self.assertTrue(not "warning" in self.module.state(self.module.widgets()[0]))
self.assertTrue(not "critical" in self.module.state(self.module.widgets()[0]))
def test_overload(self):
self.file.read.return_value = "120"
self.module.update_all()
self.assertTrue(not "warning" in self.module.state(self.module.widgets()[0]))
self.assertTrue(not "critical" in self.module.state(self.module.widgets()[0]))
self.assertEquals(self.module.capacity(self.module.widgets()[0]), "100%")
def test_ac(self):
self.exists.return_value = False
self.file.read.return_value = "120"
self.module.update_all()
self.assertEquals(self.module.capacity(self.module.widgets()[0]), "ac")
self.assertTrue("AC" in self.module.state(self.module.widgets()[0]))
def test_error(self):
self.file.read.side_effect = IOError("failed to read")
self.module.update_all()
self.assertEquals(self.module.capacity(self.module.widgets()[0]), "n/a")
self.assertTrue("critical" in self.module.state(self.module.widgets()[0]))
self.assertTrue("unknown" in self.module.state(self.module.widgets()[0]))
def test_charging(self):
self.file.read.return_value = self.chargedValue
self.module.update_all()
self.assertTrue("charged" in self.module.state(self.module.widgets()[0]))
self.file.read.return_value = self.normalValue
self.module.update_all()
self.assertTrue("charging" in self.module.state(self.module.widgets()[0]))
def test_discharging(self):
for limit in [ 10, 25, 50, 80, 100 ]:
value = limit - 1
self.file.read.return_value = str(value)
self.module.update_all()
self.file.read.return_value = "Discharging"
self.assertTrue("discharging-{}".format(limit) in self.module.state(self.module.widgets()[0]))
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,69 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
try:
FileNotFoundError
except NameError:
FileNotFoundError = IOError
import tests.mocks as mocks
import bumblebee.util
from bumblebee.config import Config
from bumblebee.input import WHEEL_UP, WHEEL_DOWN
from bumblebee.modules.brightness import Module
class TestBrightnessModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
self.tool = ""
self.up = ""
self.down = ""
if bumblebee.util.which("light"):
self.tool = "light"
self.up = "-A {}%"
self.down = "-U {}%"
elif bumblebee.util.which("brightnessctl"):
self.tool = "brightnessctl"
self.up = "s {}%+"
self.down = "s {}%-"
else:
self.tool = "xbacklight"
self.up = "+{}%"
self.down = "-{}%"
def tearDown(self):
mocks.teardown_test(self)
# def test_format(self):
# for widget in self.module.widgets():
# self.assertEquals(len(widget.full_text()), len("100%"))
def test_wheel_up(self):
mocks.mouseEvent(stdin=self.stdin, button=WHEEL_UP, inp=self.input, module=self.module)
self.popen.assert_call("{} {}".format(self.tool, self.up.format(2)))
def test_wheel_down(self):
mocks.mouseEvent(stdin=self.stdin, button=WHEEL_DOWN, inp=self.input, module=self.module)
self.popen.assert_call("{} {}".format(self.tool, self.down.format(2)))
def test_custom_step(self):
self.config.set("brightness.step", "10")
module = Module(engine=self.engine, config={"config": self.config})
mocks.mouseEvent(stdin=self.stdin, button=WHEEL_DOWN, inp=self.input, module=module)
self.popen.assert_call("{} {}".format(self.tool, self.down.format(10)))
@mock.patch('bumblebee.modules.brightness.open', create=True)
def test_error(self,mock_open):
mock_open.side_effect = FileNotFoundError
self.module.update_all()
self.assertEquals(self.module.brightness(self.anyWidget), "n/a")
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,50 +0,0 @@
# pylint: disable=C0103,C0111
import unittest
from mock import patch
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE
from bumblebee.modules.caffeine import Module
class TestCaffeineModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
def tearDown(self):
mocks.teardown_test(self)
def test_check_requirements(self):
with patch('bumblebee.util.which', side_effect=['', 'xprop', 'xdg-screensaver']):
self.assertTrue(['xdotool'] == self.module._check_requirements())
def test_get_i3bar_xid_returns_digit(self):
self.popen.mock.communicate.return_value = ("8388614", None)
self.assertTrue(self.module._get_i3bar_xid().isdigit())
def test_get_i3bar_xid_returns_error_string(self):
self.popen.mock.communicate.return_value = ("Some error message", None)
self.assertTrue(self.module._get_i3bar_xid() is None)
def test_get_i3bar_xid_returns_empty_string(self):
self.popen.mock.communicate.return_value = ("", None)
self.assertTrue(self.module._get_i3bar_xid() is None)
def test_suspend_screensaver_success(self):
with patch.object(self.module, '_get_i3bar_xid', return_value=8388614):
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.assertTrue(self.module._suspend_screensaver() is True)
def test_suspend_screensaver_fail(self):
with patch.object(self.module, '_get_i3bar_xid', return_value=None):
self.module._active = False
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.assertTrue(self.module._suspend_screensaver() is False)
def test_resume_screensaver(self):
with patch.object(self.module, '_check_requirements', return_value=[]):
self.module._active = True
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.assertTrue(self.module._resume_screensaver() is True)
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,114 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE
from bumblebee.modules.cmus import Module
class TestCmusModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
self.songTemplate = """
status {status}
file /path/to/file
duration {duration}
position {position}
tag title {title}
tag artist {artist}
tag album {album}
tag tracknumber 1
tag date 1984
tag comment comment
"""
def tearDown(self):
mocks.teardown_test(self)
def test_read_song(self):
self.popen.mock.communicate.return_value = ("song", None)
self.module.update_all()
self.popen.assert_call("cmus-remote -Q")
def test_handle_runtimeerror(self):
self.popen.mock.communicate.side_effect = RuntimeError("error loading song")
self.module.update_all()
self.assertEquals(self.module.description(self.anyWidget), " - /")
def test_format(self):
self.popen.mock.communicate.return_value = (self.songTemplate.format(
artist="an artist", title="a title", duration="100", position="20",
album="an album", status="irrelevant"
), None)
self.module.update_all()
self.anyWidget.set("theme.width", 1000)
self.assertEquals(self.module.description(self.anyWidget),
"an artist - a title 00:20/01:40"
)
def test_scrollable_format(self):
self.popen.mock.communicate.return_value = (self.songTemplate.format(
artist="an artist", title="a title", duration="100", position="20",
album="an album", status="irrelevant"
), None)
self.module.update_all()
self.anyWidget.set("theme.width", 10)
self.assertEquals(self.module.description(self.anyWidget),
"an artist - a title 00:20/01:40"[:10]
)
def test_repeat(self):
self.popen.mock.communicate.return_value = ("set repeat false", None)
self.module.update_all()
self.assertTrue("repeat-off" in self.module.state(self.module.widget("cmus.repeat")))
self.popen.mock.communicate.return_value = ("set repeat true", None)
self.module.update_all()
self.assertTrue("repeat-on" in self.module.state(self.module.widget("cmus.repeat")))
def test_shuffle(self):
self.popen.mock.communicate.return_value = ("set shuffle false", None)
self.module.update_all()
self.assertTrue("shuffle-off" in self.module.state(self.module.widget("cmus.shuffle")))
self.popen.mock.communicate.return_value = ("set shuffle true", None)
self.module.update_all()
self.assertTrue("shuffle-on" in self.module.state(self.module.widget("cmus.shuffle")))
def test_prevnext(self):
self.assertTrue("prev" in self.module.state(self.module.widget("cmus.prev")))
self.assertTrue("next" in self.module.state(self.module.widget("cmus.next")))
def test_main(self):
self.popen.mock.communicate.return_value = ("status paused", None)
self.module.update_all()
self.assertTrue("paused" in self.module.state(self.module.widget("cmus.main")))
self.popen.mock.communicate.return_value = ("status playing", None)
self.module.update_all()
self.assertTrue("playing" in self.module.state(self.module.widget("cmus.main")))
self.popen.mock.communicate.return_value = ("status stopped", None)
self.module.update_all()
self.assertTrue("stopped" in self.module.state(self.module.widget("cmus.main")))
def test_widget(self):
self.assertEquals(len(self.module.widgets()), 5)
for idx, val in enumerate(["prev", "main", "next", "shuffle", "repeat"]):
self.assertEquals(self.module.widgets()[idx].name, "cmus.{}".format(val))
def test_interaction(self):
events = [
{"widget": "cmus.shuffle", "action": "cmus-remote -S"},
{"widget": "cmus.repeat", "action": "cmus-remote -R"},
{"widget": "cmus.next", "action": "cmus-remote -n"},
{"widget": "cmus.prev", "action": "cmus-remote -r"},
{"widget": "cmus.main", "action": "cmus-remote -u"},
]
for event in events:
mocks.mouseEvent(stdin=self.stdin, inp=self.input, module=self.module, instance=self.module.widget(event["widget"]).id, button=LEFT_MOUSE)
self.popen.assert_call(event["action"])
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,45 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE
from bumblebee.modules.cpu import Module
class TestCPUModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
self._psutil = mock.patch("bumblebee.modules.cpu.psutil")
self.psutil = self._psutil.start()
def tearDown(self):
self._psutil.stop()
mocks.teardown_test(self)
def test_format(self):
self.psutil.cpu_percent.return_value = 21.0
self.module.update_all()
for widget in self.module.widgets():
self.assertEquals(len(widget.full_text()), len("99.0%"))
def test_leftclick(self):
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.popen.assert_call("gnome-system-monitor")
def test_warning(self):
self.config.set("cpu.critical", "20")
self.config.set("cpu.warning", "18")
self.psutil.cpu_percent.return_value = 19.0
self.module.update_all()
self.assertTrue("warning" in self.module.state(self.anyWidget))
def test_critical(self):
self.config.set("cpu.critical", "20")
self.config.set("cpu.warning", "19")
self.psutil.cpu_percent.return_value = 21.0
self.module.update_all()
self.assertTrue("critical" in self.module.state(self.anyWidget))
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,48 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE
from bumblebee.modules.disk import Module
class MockVFS(object):
def __init__(self, perc):
self.f_blocks = 1024*1024
self.f_frsize = 1
self.f_bfree = self.f_blocks*(1.0 - perc/100.0)
class TestDiskModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
self._os = mock.patch("bumblebee.modules.disk.os")
self.os = self._os.start()
self.config.set("disk.path", "somepath")
self.config.set("disk.open", "nautilus")
def tearDown(self):
self._os.stop()
mocks.teardown_test(self)
def test_leftclick(self):
module = Module(engine=self.engine, config={"config":self.config})
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=module)
self.popen.assert_call("nautilus {}".format(self.module.parameter("path")))
def test_warning(self):
self.config.set("disk.critical", "80")
self.config.set("disk.warning", "70")
self.os.statvfs.return_value = MockVFS(75.0)
self.module.update_all()
self.assertTrue("warning" in self.module.state(self.anyWidget))
def test_critical(self):
self.config.set("disk.critical", "80")
self.config.set("disk.warning", "70")
self.os.statvfs.return_value = MockVFS(85.0)
self.module.update_all()
self.assertTrue("critical" in self.module.state(self.anyWidget))
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,59 +0,0 @@
# -*- coding: utf-8 -*-
import mock
import unittest
from bumblebee.config import Config
import bumblebee.modules.hddtemp
class TestHddtempModule(unittest.TestCase):
def setUp(self):
config = Config()
self.module = bumblebee.modules.hddtemp.Module(
engine=mock.Mock(), config={"config": config})
self.data_line = "|/dev/sda|TOSHIBA DT01ACA100 <20>|35|C||/dev/sdb|TOSHIBA DT01ACA100 <20>|37|C|"
self.expected_parts = [
"/dev/sda",
"TOSHIBA DT01ACA100 <20>",
"35",
"C",
"",
"/dev/sdb",
"TOSHIBA DT01ACA100 <20>",
"37",
"C",
""]
self.expected_per_disk = [
["/dev/sda",
"TOSHIBA DT01ACA100 <20>",
"35",
"C",
""],
["/dev/sdb",
"TOSHIBA DT01ACA100 <20>",
"37",
"C",
""]]
self.device_record = self.expected_per_disk[0]
self.expected_name_and_temp = ("sda", "35")
self.expected_hddtemp = "sda+35°C"
def test_get_parts(self):
self.assertEqual(
self.expected_parts, self.module._get_parts(self.data_line))
def test_partition_parts(self):
self.assertEqual(
self.expected_per_disk,
self.module._partition_parts(self.expected_parts))
def test_get_name_and_temp(self):
self.assertEqual(
self.expected_name_and_temp,
self.module._get_name_and_temp(self.device_record))
def test_get_hddtemp(self):
self.assertEqual(
self.expected_hddtemp,
self.module._get_hddtemp(self.expected_name_and_temp))

View file

@ -1,49 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
from bumblebee.modules.http_status import Module
from bumblebee.config import Config
class TestHttpStatusModule(unittest.TestCase):
def test_status_success(self):
config = Config()
config.set("http_status.target", "http://example.org")
self.module = Module(engine=mock.Mock(), config={"config":config})
self.assertTrue(not "warning" in self.module.state(self.module.widgets()[0]))
self.assertTrue(not "critical" in self.module.state(self.module.widgets()[0]))
self.assertEqual(self.module.getStatus(), "200")
self.assertEqual(self.module.getOutput(), "200")
def test_status_error(self):
config = Config()
config.set("http_status.expect", "not a 200")
config.set("http_status.target", "http://example.org")
self.module = Module(engine=mock.Mock(), config={"config":config})
self.assertTrue(not "warning" in self.module.state(self.module.widgets()[0]))
self.assertTrue("critical" in self.module.state(self.module.widgets()[0]))
self.assertEqual(self.module.getStatus(), "200")
self.assertEqual(self.module.getOutput(), "200 != not a 200")
def test_label(self):
config = Config()
config.set("http_status.label", "example")
config.set("http_status.target", "http://example.org")
self.module = Module(engine=mock.Mock(), config={"config":config})
self.assertEqual(self.module.getOutput(), "example: 200")
def test_unknow(self):
config = Config()
config.set("http_status.target", "invalid target")
self.module = Module(engine=mock.Mock(), config={"config":config})
self.assertTrue("warning" in self.module.state(self.module.widgets()[0]))
self.assertEqual(self.module.getStatus(), "UNK")
self.assertEqual(self.module.getOutput(), "UNK != 200")
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,57 +0,0 @@
# pylint: disable=C0103,C0111
import json
import mock
import unittest
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE
from bumblebee.modules.load import Module
class TestLoadModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
self._mp = mock.patch("bumblebee.modules.load.multiprocessing")
self._os = mock.patch("bumblebee.modules.load.os")
self.mp = self._mp.start()
self.os = self._os.start()
self.mp.cpu_count.return_value = 1
def tearDown(self):
self._mp.stop()
self._os.stop()
mocks.teardown_test(self)
def test_leftclick(self):
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.popen.assert_call("gnome-system-monitor")
def test_load_format(self):
self.os.getloadavg.return_value = [ 5.9, 1.2, 0.8 ]
self.module.update_all()
self.assertEquals(self.module.load(self.anyWidget), "5.90/1.20/0.80")
def test_warning(self):
self.config.set("load.critical", "1")
self.config.set("load.warning", "0.8")
self.os.getloadavg.return_value = [ 0.9, 0, 0 ]
self.module.update_all()
self.assertTrue("warning" in self.module.state(self.anyWidget))
def test_critical(self):
self.config.set("load.critical", "1")
self.config.set("load.warning", "0.8")
self.os.getloadavg.return_value = [ 1.1, 0, 0 ]
self.module.update_all()
self.assertTrue("critical" in self.module.state(self.anyWidget))
def test_assume_single_core(self):
self.mp.cpu_count.side_effect = NotImplementedError
module = Module(engine=self.engine, config={"config": mock.Mock() })
self.assertEquals(1, module._cpus)
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

View file

@ -1,34 +0,0 @@
# pylint: disable=C0103,C0111
import mock
import unittest
import tests.mocks as mocks
from bumblebee.input import LEFT_MOUSE, RIGHT_MOUSE, WHEEL_UP, WHEEL_DOWN
from bumblebee.modules.pulseaudio import Module
class TestPulseAudioModule(unittest.TestCase):
def setUp(self):
mocks.setup_test(self, Module)
def tearDown(self):
mocks.teardown_test(self)
def test_leftclick(self):
mocks.mouseEvent(stdin=self.stdin, button=LEFT_MOUSE, inp=self.input, module=self.module)
self.popen.assert_call("pactl set-source-mute @DEFAULT_SOURCE@ toggle")
def test_rightclick(self):
mocks.mouseEvent(stdin=self.stdin, button=RIGHT_MOUSE, inp=self.input, module=self.module)
self.popen.assert_call("pavucontrol")
def test_wheelup(self):
mocks.mouseEvent(stdin=self.stdin, button=WHEEL_UP, inp=self.input, module=self.module)
self.popen.assert_call("pactl set-source-volume @DEFAULT_SOURCE@ +2%")
def test_wheeldown(self):
mocks.mouseEvent(stdin=self.stdin, button=WHEEL_DOWN, inp=self.input, module=self.module)
self.popen.assert_call("pactl set-source-volume @DEFAULT_SOURCE@ -2%")
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4