forked from Keelrleek/Harbor
173 lines
8.6 KiB
Python
173 lines
8.6 KiB
Python
"""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 <x> & '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()
|