Asger Geel Weirsøe
69cb255120
Installable PWA showing rigs, seats (working, idle, waiting for an answer), queue items and recent commits; seat screens with replies; push through ntfy. Reads state through the rig CLI only, Python standard library, no build step, styled entirely with the shared weirsoe-theme. Claude-Session: https://claude.ai/code/session_01QhLkBGEip8wQdaZ54z4AqJ
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
import unittest
|
||
|
||
import server
|
||
|
||
|
||
def snap(commits=(), queue=(), waiting=False):
|
||
return {
|
||
"rigs": [{"name": "team", "seats": [{"name": "dev-qa@team", "waiting": waiting}]}],
|
||
"queue": list(queue),
|
||
"repos": [{"name": "repo", "commits": [{"hash": h, "subject": f"commit {h}"} for h in commits]}],
|
||
}
|
||
|
||
|
||
def q(qid, state, dest="orch-lead@team"):
|
||
return {"id": qid, "state": state, "dest": dest, "source": "operator-agent@kernel", "summary": f"task {qid}"}
|
||
|
||
|
||
class WatcherTest(unittest.TestCase):
|
||
def test_first_snapshot_only_seeds(self):
|
||
w = server.Watcher()
|
||
self.assertEqual(w.events(snap(commits=["a"], queue=[q("1", "in-progress")])), [])
|
||
|
||
def test_new_commits_grouped_per_repo(self):
|
||
w = server.Watcher()
|
||
w.events(snap(commits=["a"]))
|
||
ev = w.events(snap(commits=["a", "b", "c"]))
|
||
self.assertEqual(len(ev), 1)
|
||
self.assertIn("2 ny commits", ev[0][0])
|
||
|
||
def test_lead_task_done_notifies_once(self):
|
||
w = server.Watcher()
|
||
w.events(snap(queue=[q("1", "in-progress"), q("2", "in-progress", "dev-build@team")]))
|
||
ev = w.events(snap(queue=[q("1", "done"), q("2", "done", "dev-build@team")]))
|
||
self.assertEqual([e[0] for e in ev], ["Opgave done: orch-lead@team"])
|
||
self.assertEqual(w.events(snap(queue=[q("1", "done")])), [])
|
||
|
||
def test_new_request_to_human_notifies(self):
|
||
w = server.Watcher()
|
||
w.events(snap())
|
||
ev = w.events(snap(queue=[q("9", "in-progress", "operator-agent@kernel")]))
|
||
self.assertEqual(ev[0][0], "Teamet spørger dig")
|
||
|
||
def test_prompt_must_survive_two_polls(self):
|
||
w = server.Watcher()
|
||
self.assertEqual(w.events(snap(waiting=True)), [])
|
||
ev = w.events(snap(waiting=True))
|
||
self.assertEqual(ev[0][0], "dev-qa@team venter på svar")
|
||
self.assertEqual(w.events(snap(waiting=True)), [])
|
||
w.events(snap(waiting=False))
|
||
w.events(snap(waiting=True))
|
||
self.assertEqual(len(w.events(snap(waiting=True))), 1)
|
||
|
||
def test_prompt_pattern(self):
|
||
self.assertTrue(server.PROMPT.search(" Do you want to proceed?\n ❯ 1. Yes"))
|
||
self.assertFalse(server.PROMPT.search("❯ yes, launch it"))
|
||
|
||
def test_seat_names(self):
|
||
self.assertTrue(server.SEAT.match("dev-qa@ubl-workshop"))
|
||
self.assertFalse(server.SEAT.match("x; rm -rf /"))
|
||
self.assertFalse(server.SEAT.match("../etc@x/y"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|