"""Git radar, bulk tasks and templates, the watchdog, policies and Windows notifications.""" import os import re import shutil import subprocess import sys import unittest os.environ["HARBOR_NO_TOASTS"] = "1" from helpers import ROOT, env_for, http, make_home, temp_dir, wait_for_port from harbor import notify, projects from harbor.core import Harbor, HarborError def git(cwd, *args): subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, env={**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"}) class Clock: def __init__(self, t=1_790_000_000.0): self.t = t def __call__(self): return self.t class Flow(unittest.TestCase): def setUp(self): self.dir = temp_dir() repo = self.dir / "repo" repo.mkdir() git(repo, "init", "-q", "-b", "main") (repo / "a.c").write_text("int a;\n", encoding="utf-8") git(repo, "add", "a.c") git(repo, "commit", "-q", "-m", "base") git(repo, "worktree", "add", "-q", "-b", "ann/work", str(self.dir / "wt-ann")) for i in range(3): (self.dir / "wt-ann" / f"f{i}.c").write_text(f"int f{i};\n", encoding="utf-8") git(self.dir / "wt-ann", "add", f"f{i}.c") git(self.dir / "wt-ann", "commit", "-q", "-m", f"draft {i}") (self.dir / "wt-ann" / "scratch.txt").write_text("x", encoding="utf-8") data = {"name": "flow", "coordinators": ["Lead"], "reviewers": ["Lead"], "database": str(self.dir / "f.db"), "policy": {"land_requires": ["Rev"], "all_per_hour": 2}, "templates": {"decomp": {"steps": [ {"name": "draft", "title": "Make {scope} readable", "assignee": "{owner}"}, {"name": "names", "title": "Name what {scope} calls", "assignee": "{owner}", "after": ["draft"]}]}}, "agents": [{"name": "Lead", "workdir": str(repo)}, {"name": "Ann", "workdir": str(self.dir / "wt-ann")}, {"name": "Rev"}]} self.clock = Clock() self.h = Harbor(projects.project_from_dict(data), clock=self.clock) def tearDown(self): self.h.close() shutil.rmtree(self.dir, ignore_errors=True) def test_git_radar(self): g = self.h.git_status("Ann", max_age=0) self.assertEqual((g["branch"], g["dirty"], g["behind_main"], g["ahead_of_main"]), ("ann/work", 1, 0, 3)) self.assertEqual(len(g["unsubmitted"]), 3) sha = g["last_commit"]["sha"] self.h.submit("Ann", "drafts", commit=sha) self.assertEqual([c["sha"] for c in self.h.git_status("Ann", max_age=0)["unsubmitted"]][:1] != [sha], True) self.assertEqual(self.h.git_status("Lead", max_age=0)["unsubmitted"], []) # main is the landing branch self.assertIsNone(self.h.git_status("Rev")) def test_templates_bulk_and_band_routing(self): self.h.claim("Lead", "range:0x440000-0x47FFFF", kind="band", on_behalf_of="Ann") steps = self.h.run_template("Lead", "decomp", scope="fn:0x445310") self.assertEqual([(t["title"], t["assignee"]) for t in steps], [("Make fn:0x445310 readable", "Ann"), ("Name what fn:0x445310 calls", "Ann")]) self.assertEqual(steps[1]["waiting_on"], [steps[0]["id"]]) r = self.h.bulk_tasks("Lead", "# held functions\n0x445310 again\nfn:0x450000 | RouteCellCost\n0x500000\n" "Write the README\n") self.assertEqual([t["title"] for t in r["created"]], ["Make 0x450000 readable (RouteCellCost)", "Make 0x500000 readable", "Write the README"]) self.assertEqual([t["assignee"] for t in r["created"]], ["Ann", None, None]) self.assertEqual(r["skipped"], ["fn:0x445310 already has an open task"]) with self.assertRaises(HarborError): self.h.run_template("Lead", "nope") def test_policies(self): head, older = subprocess.run(["git", "-C", str(self.dir / "wt-ann"), "rev-parse", "HEAD", "HEAD~1"], capture_output=True, text=True, check=True).stdout.split() for bad in ("HEAD", "PLACEHOLDER", "abc", "0123456789abcdef"): # not a hash, or not in git with self.assertRaises(HarborError): self.h.submit("Ann", "x", commit=bad) item = self.h.submit("Ann", "x", commit=head[:8]) self.assertEqual(self.h.item(item["id"])[0]["commit_hash"], head) # stored in full self.h.transition("Lead", item["id"], "approved") with self.assertRaises(HarborError): self.h.transition("Lead", item["id"], "landed", commit="b4") # Rev has not approved self.h.transition("Rev", item["id"], "approved") if False else None self.h.transition("Lead", item["id"], "changes_requested") self.h.transition("Ann", item["id"], "submitted", commit=older) self.h.transition("Rev", item["id"], "approved") self.h.transition("Lead", item["id"], "landed", commit="b4") self.h.send("Ann", "All", "one") self.h.send("Ann", "All", "two") with self.assertRaises(HarborError): self.h.send("Ann", "All", "three") # capped at 2 per hour self.h.send("Ann", ["Lead"], "direct is fine") self.h.send("Lead", "All", "coordinators are not capped") def test_watchdog(self): for _ in range(160): self.h.activity("Ann", "shell") self.assertEqual([b["agent"] for b in self.h.busy_without_progress()], ["Ann"]) self.h.doing("Ann", "still on the proof") self.assertEqual(self.h.busy_without_progress(), []) def test_project_validation(self): bad = {"name": "x", "agents": [{"name": "A"}], "templates": {"t": {"steps": [{"title": "a", "after": ["zz"]}]}}} with self.assertRaises(projects.ProjectError): projects.project_from_dict(bad, home=self.dir) class Notifications(unittest.TestCase): def test_script_and_selection(self): script = notify.toast_script("Codex asks you", "Rename & 'y'?") self.assertIn("<x> & ''y''", script) d = temp_dir() h = Harbor(make_home(d)["test"]) try: h.send("Codex", "All", "before") notify.set_enabled(h, True) self.assertTrue(notify.enabled(h)) h.send("Codex", "All", "a broadcast") h.send("Codex", ["User"], "a direct message") h.ask("Kimi", "Ship?", ["Yes", "No"]) pending, top = notify.pending_for_human(h) self.assertEqual([m["body"].split("\n")[0] for m in pending], ["a direct message", "Ship?"]) notify.mark(h, top) self.assertEqual(notify.pending_for_human(h)[0], []) self.assertFalse(notify.toast("t", "b")) # disabled in tests finally: h.close() shutil.rmtree(d, ignore_errors=True) class ServerFlow(unittest.TestCase): PORT = 8793 @classmethod def setUpClass(cls): cls.dir = temp_dir() cls.project = make_home(cls.dir)["test"] cls.proc = subprocess.Popen([sys.executable, "-B", "-m", "harbor", "serve", "--port", str(cls.PORT)], cwd=cls.dir, env=env_for(cls.dir, PYTHONPATH=str(ROOT)), stdout=subprocess.PIPE, stderr=subprocess.STDOUT) assert wait_for_port(cls.PORT) cls.token = re.search(r'name="harbor-token" content="([^"]+)"', http(cls.PORT, "/")[1]).group(1) @classmethod def tearDownClass(cls): cls.proc.terminate() cls.proc.wait(timeout=10) cls.proc.stdout.close() shutil.rmtree(cls.dir, ignore_errors=True) def test_bulk_and_toasts(self): status, r = http(self.PORT, "/api/bulk", {"text": "0x445310 One\n0x445320 Two"}, token=self.token) self.assertEqual((status, r["created"]), (200, 2)) self.assertEqual(len(http(self.PORT, "/api/overview")[1]["tasks"]), 2) self.assertEqual(http(self.PORT, "/api/toasts", {"on": True}, token=self.token)[1]["on"], True) self.assertTrue(http(self.PORT, "/api/overview")[1]["toasts"]) self.assertEqual(http(self.PORT, "/api/toasts", {"on": False}, token=self.token)[1]["on"], False) if __name__ == "__main__": unittest.main()