[modules/caffeine] Re-implement caffeine module

Add caffeine module & add a framework for testing it (no tests yet,
though).

see #23

bumblebee/input.py

@@ -28,10 +28,12 @@ def read_input(inp):
         try:
             event = json.loads(line)
             inp.callback(event)
+            inp.has_valid_event = True
             inp.redraw()
         except ValueError:
             pass
     inp.has_event = True
+    inp.has_valid_event = True
     inp.clean_exit = True
 
 class I3BarInput(object):
@@ -42,12 +44,15 @@ class I3BarInput(object):
         self.clean_exit = False
         self.global_id = str(uuid.uuid4())
         self.need_event = False
+        self.need_valid_event = False
         self.has_event = False
+        self.has_valid_event = False
         self._condition = threading.Condition()
 
     def start(self):
         """Start asynchronous input processing"""
         self.has_event = False
+        self.has_valid_event = False
         self.running = True
         self._condition.acquire()
         self._thread = threading.Thread(target=read_input, args=(self,))
@@ -65,16 +70,20 @@ class I3BarInput(object):
     def wait(self, timeout):
         self._condition.wait(timeout)
 
-    def _wait(self):
+    def _wait(self, valid=False):
         while not self.has_event:
             time.sleep(0.1)
+        if valid:
+            while not self.has_valid_event:
+                time.sleep(0.1)
         self.has_event = False
+        self.has_valid_event = False
 
     def stop(self):
         """Stop asynchronous input processing"""
         self._condition.release()
         if self.need_event:
-            self._wait()
+            self._wait(self.need_valid_event)
         self.running = False
         self._thread.join()
         return self.clean_exit

bumblebee/modules/caffeine.py

@@ -0,0 +1,47 @@
+# pylint: disable=C0111,R0903
+
+"""Enable/disable automatic screen locking.
+"""
+
+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.caffeine)
+        )
+        engine.input.register_callback(self, button=bumblebee.input.LEFT_MOUSE,
+            cmd=self._toggle
+        )
+
+    def caffeine(self):
+        return ""
+
+    def state(self, widget):
+        if self._active():
+            return "activated"
+        return "deactivated"
+
+    def _active(self):
+        for line in bumblebee.util.execute("xset q").split("\n"):
+            if "timeout" in line:
+                timeout = int(line.split(" ")[4])
+                if timeout == 0:
+                    return True
+                return False
+        return False
+
+    def update(self, widgets):
+        pass
+
+    def _toggle(self, widget):
+        if self._active():
+            bumblebee.util.execute("xset s default")
+            bumblebee.util.execute("notify-send \"Out of coffee\"")
+        else:
+            bumblebee.util.execute("xset s off")
+            bumblebee.util.execute("notify-send \"Consuming caffeine\"")
+
+# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

tests/modules/test_caffeine.py

@@ -0,0 +1,21 @@
+# pylint: disable=C0103,C0111
+
+import json
+import unittest
+import mock
+
+import bumblebee.input
+from bumblebee.input import I3BarInput
+from bumblebee.modules.caffeine import Module
+from tests.util import MockEngine, MockConfig, assertPopen
+
+class TestCaffeineModule(unittest.TestCase):
+    def setUp(self):
+        self.engine = MockEngine()
+        self.engine.input = I3BarInput()
+        self.engine.input.need_event = True
+        self.engine.input.need_valid_event = True
+        self.config = MockConfig()
+        self.module = Module(engine=self.engine, config={ "config": self.config })
+        for widget in self.module.widgets():
+            widget.link_module(self.module)

tests/util.py

@@ -11,7 +11,7 @@ def assertWidgetAttributes(test, widget):
 
 def assertPopen(output, cmd):
     res = shlex.split(cmd)
-    output.assert_called_with(res,
+    output.assert_any_call(res,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT
     )