aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--deckit/modules/obs.py18
-rw-r--r--deckit/modules/windows.py10
-rw-r--r--deckit/obsclient.py5
-rw-r--r--deckit/server.py7
-rw-r--r--tests/test_llm.py9
-rw-r--r--tests/test_obs.py9
-rw-r--r--tests/test_obsclient.py22
-rw-r--r--tests/test_server.py7
-rw-r--r--tests/test_theme.py10
-rw-r--r--tests/test_windows.py10
10 files changed, 95 insertions, 12 deletions
diff --git a/deckit/modules/obs.py b/deckit/modules/obs.py
index f4bf71a..96217c4 100644
--- a/deckit/modules/obs.py
+++ b/deckit/modules/obs.py
@@ -18,6 +18,7 @@ def make(cfg: dict) -> Module:
host=o.get("host", "127.0.0.1"), port=o.get("port", 4455), password=o.get("password", ""))
lock = threading.Lock()
stop = threading.Event()
+ last_error = {"msg": None}
def req(kind, data=None):
with lock:
@@ -28,13 +29,14 @@ def make(cfg: dict) -> Module:
if not getattr(client, "connected", False):
try:
client.connect()
- except Exception:
- pass
+ last_error["msg"] = None
+ except Exception as e:
+ last_error["msg"] = str(e) or type(e).__name__
stop.wait(5)
def state():
if not getattr(client, "connected", False):
- return {"connected": False}
+ return {"connected": False, "error": last_error["msg"]}
try:
scenes = req("GetSceneList")
rec = req("GetRecordStatus")
@@ -80,11 +82,17 @@ def make(cfg: dict) -> Module:
raise ModuleError("scene required")
req("SetCurrentProgramScene", {"sceneName": name})
+ def flag(body):
+ on = body.get("on")
+ if not isinstance(on, bool):
+ raise ModuleError('"on" must be true or false')
+ return on
+
def record(body):
- req("StartRecord" if body.get("on") else "StopRecord")
+ req("StartRecord" if flag(body) else "StopRecord")
def stream(body):
- req("StartStream" if body.get("on") else "StopStream")
+ req("StartStream" if flag(body) else "StopStream")
def mute(body):
req("SetInputMute", {"inputName": str(body.get("input")), "inputMuted": bool(body.get("muted"))})
diff --git a/deckit/modules/windows.py b/deckit/modules/windows.py
index 022bdd7..d02fc0c 100644
--- a/deckit/modules/windows.py
+++ b/deckit/modules/windows.py
@@ -45,7 +45,7 @@ def _state(run):
try:
clients = json.loads(run(["hyprctl", "clients", "-j"]) or "[]")
active = json.loads(run(["hyprctl", "activewindow", "-j"]) or "{}").get("address")
- monitors = json.loads(run(["hyprctl", "monitors", "-j"]) or "[]")
+ workspaces = json.loads(run(["hyprctl", "workspaces", "-j"]) or "[]")
except json.JSONDecodeError:
return {"windows": [], "workspaces": [], "activeClass": None}
wins = sorted(
@@ -54,7 +54,8 @@ def _state(run):
for c in clients if c.get("mapped", True) and c["workspace"]["id"] > 0),
key=lambda w: (w["workspace"], w["class"]))
active_class = next((w["class"] for w in wins if w["active"]), None)
- return {"windows": wins, "workspaces": [m["activeWorkspace"]["id"] for m in monitors],
+ return {"windows": wins,
+ "workspaces": sorted(w["id"] for w in workspaces if w.get("id", 0) > 0),
"activeClass": active_class}
@@ -88,5 +89,8 @@ def make(cfg: dict) -> Module:
raise ModuleError("unknown launch key")
_popen(["sh", "-c", command] if isinstance(command, str) else command)
- return Module(name="windows", title="Windows", state=lambda: _state(run),
+ def state():
+ return {**_state(run), "launches": list(launch)}
+
+ return Module(name="windows", title="Windows", state=state,
actions={"workspace": workspace, "focus": focus, "close": close, "launch": launch_key})
diff --git a/deckit/obsclient.py b/deckit/obsclient.py
index 2a8fd8c..181225a 100644
--- a/deckit/obsclient.py
+++ b/deckit/obsclient.py
@@ -49,9 +49,14 @@ class ObsClient:
self._thread.start()
def _recv_loop(self):
+ import websocket
+
while not self._stop.is_set():
try:
msg = json.loads(self.ws.recv())
+ except websocket.WebSocketTimeoutException:
+ # An idle OBS is not a closed one; keep reading.
+ continue
except Exception:
break
op, d = msg.get("op"), msg.get("d", {})
diff --git a/deckit/server.py b/deckit/server.py
index bbd92cf..bddbb00 100644
--- a/deckit/server.py
+++ b/deckit/server.py
@@ -129,14 +129,17 @@ class Handler(BaseHTTPRequestHandler):
return self._send(404, str(e).encode())
if fn is None:
return self._send(404, b"unknown action")
- length = int(self.headers.get("Content-Length", 0) or 0)
+ try:
+ length = int(self.headers.get("Content-Length", 0) or 0)
+ except ValueError:
+ return self._send(400, b"Invalid Content-Length")
if self.headers.get("Content-Type") != "application/json" or not 0 < length <= 4096:
return self._send(400, b"Expected a small JSON body")
try:
result = fn(json.loads(self.rfile.read(length)))
except module.ModuleError as e:
return self._send(400, str(e).encode())
- except (ValueError, AttributeError):
+ except (ValueError, AttributeError, TypeError, KeyError):
return self._send(400, b"Invalid request")
if result is None:
return self._send(204)
diff --git a/tests/test_llm.py b/tests/test_llm.py
index 759666c..9cdfcf6 100644
--- a/tests/test_llm.py
+++ b/tests/test_llm.py
@@ -27,6 +27,15 @@ class Up(unittest.TestCase):
self.assertEqual(st["claude"]["context_pct"], 12)
self.assertEqual(st["opencode"]["total_cost"], 1.5)
+ def test_handles_the_real_latest_shape(self):
+ data = {"/api/claude_limits": {"latest": {"context_pct": 42, "five_hour_pct": 10}},
+ "/api/claude_session": {},
+ "/api/summary": {}}
+ st = llm.make({"llm": {"url": "http://x"},
+ "fetch": lambda u, timeout=2: data[u]}).state()
+ self.assertEqual(st["claude"]["context_pct"], 42)
+ self.assertEqual(st["claude"]["five_hour_pct"], 10)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_obs.py b/tests/test_obs.py
index 876d48b..aeb0791 100644
--- a/tests/test_obs.py
+++ b/tests/test_obs.py
@@ -39,7 +39,14 @@ class Actions(unittest.TestCase):
def test_state_offline(self):
class Off:
connected = False
- self.assertEqual(obs.make({"client": Off()}).state(), {"connected": False})
+ st = obs.make({"client": Off()}).state()
+ self.assertFalse(st["connected"])
+ self.assertIn("error", st)
+
+ def test_record_requires_a_bool(self):
+ for bad in ({}, {"on": "false"}, {"on": 1}):
+ with self.assertRaises(obs.ModuleError, msg=bad):
+ self.m.actions["record"](bad)
class Serialized(unittest.TestCase):
diff --git a/tests/test_obsclient.py b/tests/test_obsclient.py
index 7d1a5ca..54eadd1 100644
--- a/tests/test_obsclient.py
+++ b/tests/test_obsclient.py
@@ -25,5 +25,27 @@ class Disconnected(unittest.TestCase):
oc.ObsClient(host="127.0.0.1", port=4455, password="").request("GetVersion")
+class RecvLoop(unittest.TestCase):
+ def test_a_recv_timeout_does_not_end_the_loop(self):
+ import websocket
+
+ class FakeWS:
+ calls = 0
+
+ def recv(self):
+ FakeWS.calls += 1
+ if FakeWS.calls <= 3:
+ raise websocket.WebSocketTimeoutException("idle")
+ raise RuntimeError("closed")
+
+ c = oc.ObsClient()
+ c.ws = FakeWS()
+ c.connected = True
+ c._recv_loop()
+ # It kept calling recv past three idle timeouts; a timeout must not
+ # look like a closed socket.
+ self.assertEqual(FakeWS.calls, 4)
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_server.py b/tests/test_server.py
index 1583ad2..bf571a6 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -38,7 +38,8 @@ class Server(unittest.TestCase):
srv.TOKEN = TOKEN
srv.REGISTRY = mod.Registry([
mod.Module("demo", "Demo", lambda: {"ok": True},
- {"ping": lambda body: {"got": body.get("n")}}),
+ {"ping": lambda body: {"got": body.get("n")},
+ "boom": lambda body: int(body.get("x"))}),
])
cls.server = srv.UnixHTTPServer(cls.sock, srv.Handler)
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
@@ -86,6 +87,10 @@ class Server(unittest.TestCase):
{**HOST, **ORIGIN, "Content-Type": "application/json"}, big)
self.assertEqual(status, 400)
+ def test_rejects_wrong_typed_body(self):
+ h = {**HOST, **ORIGIN, "Content-Type": "application/json"}
+ self.assertEqual(self.req("POST", "/api/demo/boom", h, '{"x": null}')[0], 400)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_theme.py b/tests/test_theme.py
index 5f2f533..6530af5 100644
--- a/tests/test_theme.py
+++ b/tests/test_theme.py
@@ -29,6 +29,16 @@ class Theme(unittest.TestCase):
self.assertIn("--udt-base: #24273a", out)
self.assertIn("--udt-accent:", out) # fallback accent
+ def test_load_falls_back_when_files_are_missing(self):
+ saved = (theme.PALETTE_PATH, theme.ACCENT_PATH)
+ theme.PALETTE_PATH = Path("/nonexistent/palette.rasi")
+ theme.ACCENT_PATH = Path("/nonexistent/accent.rasi")
+ try:
+ out = theme.load()
+ finally:
+ theme.PALETTE_PATH, theme.ACCENT_PATH = saved
+ self.assertIn("--udt-base: #24273a", out)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_windows.py b/tests/test_windows.py
index 1573a77..6aa7925 100644
--- a/tests/test_windows.py
+++ b/tests/test_windows.py
@@ -18,6 +18,16 @@ class Commands(unittest.TestCase):
self.assertIsNone(windows.ADDR.fullmatch("0x1;rm"))
self.assertTrue(windows.ADDR.fullmatch("0x55897e5c0800"))
+ def test_state_lists_all_workspaces(self):
+ responses = {
+ ("hyprctl", "clients", "-j"): "[]",
+ ("hyprctl", "activewindow", "-j"): "{}",
+ ("hyprctl", "monitors", "-j"): "[]",
+ ("hyprctl", "workspaces", "-j"): '[{"id": 1}, {"id": 2}, {"id": 5}]',
+ }
+ st = windows._state(lambda cmd: responses[tuple(cmd)])
+ self.assertEqual(st["workspaces"], [1, 2, 5])
+
if __name__ == "__main__":
unittest.main()