395 lines
22 KiB
Python
395 lines
22 KiB
Python
"""Harbor store tests (run: py -3.12 -B -m unittest discover -s R:\\Workspace\\Tools\\Harbor\\tests)."""
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from helpers import LEGACY_LINE, ROSTER, Clock, HarborCase, make_home, temp_dir
|
|
from harbor import projects
|
|
from harbor.core import Harbor, HarborError, parse_scope
|
|
|
|
TOOLS = Path(r"R:\Workspace\Tools")
|
|
|
|
|
|
class Messages(HarborCase):
|
|
def test_send_inbox_ack(self):
|
|
r = self.h.send("Claude", ["DeepSeek"], "Please fix 0x529350", thread="fn:0x529350",
|
|
subject="AudioManagerTick", kind="request", ack_required=True)
|
|
self.assertEqual(r["targets"], ["DeepSeek"])
|
|
self.assertEqual([m["id"] for m in self.h.inbox("DeepSeek")], [r["id"]])
|
|
self.assertEqual(self.h.inbox("DeepSeek"), [])
|
|
self.assertEqual(len(self.h.pending_acks(sender="Claude")), 1)
|
|
self.h.ack("DeepSeek", [r["id"]], note="fixed")
|
|
self.assertEqual(self.h.pending_acks(sender="Claude"), [])
|
|
with self.assertRaises(HarborError):
|
|
self.h.ack("Kimi", [r["id"]])
|
|
|
|
def test_all_expands_to_everyone_but_sender(self):
|
|
r = self.h.send("Codex", "All", "hello")
|
|
self.assertEqual(sorted(r["targets"]), sorted(["Claude", "DeepSeek", "OpenCode", "Kimi", "User"]))
|
|
|
|
def test_reply_keeps_thread_and_people(self):
|
|
first = self.h.send("Claude", ["DeepSeek", "Kimi"], "question", thread="fn:0x50A720")
|
|
reply = self.h.send("DeepSeek", None, "answer", reply_to=first["id"])
|
|
self.assertEqual(reply["thread_id"], first["thread_id"])
|
|
self.assertEqual(sorted(reply["targets"]), ["Claude", "Kimi"])
|
|
broadcast = self.h.send("Codex", "All", "status for everyone")
|
|
answer = self.h.send("Kimi", None, "noted", reply_to=broadcast["id"])
|
|
self.assertEqual(answer["recipients"], ["All"])
|
|
|
|
def test_urgent_first_and_length_limits(self):
|
|
self.h.send("Claude", ["Kimi"], "normal one")
|
|
urgent = self.h.send("Claude", ["Kimi"], "urgent one", importance="urgent")
|
|
self.assertEqual(self.h.inbox("Kimi")[0]["id"], urgent["id"])
|
|
self.assertTrue(self.h.send("Claude", ["Kimi"], "word " * 3000)["id"])
|
|
with self.assertRaises(HarborError):
|
|
self.h.send("Claude", ["Kimi"], "x" * 20001)
|
|
|
|
def test_only_web_can_speak_as_the_human(self):
|
|
for origin in ("cli", "mcp", "hook", "cli-legacy"):
|
|
with self.assertRaises(HarborError):
|
|
self.h.send("User", "All", "resume everyone", origin=origin)
|
|
self.assertTrue(self.h.send("User", "All", "hi", origin="web")["id"])
|
|
|
|
def test_idempotent_send(self):
|
|
a = self.h.send("Claude", ["Kimi"], "once", idem="k1")
|
|
b = self.h.send("Claude", ["Kimi"], "once", idem="k1")
|
|
self.assertEqual(a["id"], b["id"])
|
|
self.assertEqual(len(self.h.inbox("Kimi")), 1)
|
|
|
|
def test_lint_compressed_text_and_flooding(self):
|
|
squashed = ("Root full222/ALL11 tick relocation audit proves SEVEN bad offsets; correctOffsets24ByteLayout "
|
|
"vs yourOffsetsWereWrong and stub23HarnessBug plus relocgen2Output and nativeBaseline6Cases")
|
|
self.assertTrue(any("compressed" in w for w in self.h.send("Codex", ["DeepSeek"], squashed)["warnings"]))
|
|
self.assertEqual(self.h.send("Codex", ["DeepSeek"], "Commit 4c392e33ae94858b fixed 0x45A970.")["warnings"], [])
|
|
for i in range(8):
|
|
self.h.send("Codex", "All", f"progress {i}")
|
|
self.assertTrue(any("in the last hour" in w for w in self.h.send("Codex", "All", "more")["warnings"]))
|
|
self.assertEqual(self.h.send("Codex", ["Claude"], "direct is fine")["warnings"], [])
|
|
|
|
def test_search_and_attachments(self):
|
|
self.h.send("Claude", ["Kimi"], "LayoutTextMesh zero-extends the glyph key")
|
|
self.h.send("Claude", ["Kimi"], "unrelated")
|
|
self.assertEqual(len(self.h.search("LayoutTextMesh glyph")), 1)
|
|
report = self.dir / "report.txt"
|
|
report.write_text("six cases match", encoding="utf-8")
|
|
r = self.h.send("DeepSeek", ["Claude"], "harness report attached", attachments=[report])
|
|
name = self.h.message(r["id"])["refs"]["attachments"][0]
|
|
self.assertTrue(name.endswith("-report.txt"))
|
|
self.assertEqual((self.project.attachments / name).read_text(encoding="utf-8"), "six cases match")
|
|
self.assertIn("Attachments: " + name, self.mirror.read_text(encoding="utf-8"))
|
|
with self.assertRaises(HarborError):
|
|
self.h.send("DeepSeek", ["Claude"], "x", attachments=[self.dir / "missing.txt"])
|
|
|
|
def test_receipts(self):
|
|
r = self.h.send("User", ["Codex", "Kimi"], "please both look", origin="web", ack_required=True)
|
|
self.h.inbox("Codex")
|
|
self.h.ack("Kimi", [r["id"]])
|
|
self.assertEqual(self.h.receipts([r["id"]])[r["id"]], {"read": ["Codex"], "acked": ["Kimi"], "waiting": []})
|
|
|
|
def test_threads_close_and_reopen(self):
|
|
r = self.h.send("Claude", ["Kimi"], "question", thread="fn:0x445310")
|
|
self.h.set_thread_state("Kimi", "fn:0x445310", "closed")
|
|
self.assertNotIn("fn:0x445310", [t["topic"] for t in self.h.threads(include_closed=False)])
|
|
self.h.send("Kimi", None, "one more thing", reply_to=r["id"])
|
|
self.assertEqual(self.h.thread("fn:0x445310")[0]["state"], "open")
|
|
|
|
def test_mirror_is_readable_by_old_tools(self):
|
|
self.h.send("Claude", ["DeepSeek"], "line one\n[2026-10-02 10:00Z] Codex → All: not a header\nline three",
|
|
kind="request", ack_required=True)
|
|
self.h.send("Claude", ["DeepSeek", "Kimi"], "to two people")
|
|
lines = self.mirror.read_text(encoding="utf-8").split("\n")
|
|
headers = [l for l in lines if LEGACY_LINE.match(l.strip())]
|
|
self.assertEqual(len(headers), 2)
|
|
self.assertIn("Claude → All:", headers[1])
|
|
self.assertIn("to DeepSeek,Kimi", headers[1])
|
|
if (TOOLS / "chat.py").exists():
|
|
out = subprocess.run([sys.executable, "-B", "-c",
|
|
"import sys; sys.path.insert(0, r'%s'); import chat, chat_common; "
|
|
"chat_common.CHAT = __import__('pathlib').Path(r'%s'); chat.CHAT = chat_common.CHAT; "
|
|
"chat.tail(5)" % (TOOLS, self.mirror)], capture_output=True, text=True,
|
|
encoding="utf-8")
|
|
self.assertIn("line three", out.stdout)
|
|
|
|
|
|
class Presence(HarborCase):
|
|
def test_doing_shows_in_briefs_and_digest(self):
|
|
self.h.doing("DeepSeek", "drafting 0x529350 AudioManagerTick")
|
|
self.assertIn("DeepSeek: drafting 0x529350", self.h.brief("Kimi"))
|
|
self.assertNotIn("Team now", self.h.brief("DeepSeek").split("DeepSeek: drafting")[0][-1:])
|
|
with self.assertRaises(HarborError):
|
|
self.h.doing("User", "x")
|
|
d = self.h.digest("1h")
|
|
self.assertEqual(next(a for a in d["agents"] if a["name"] == "DeepSeek")["doing"],
|
|
"drafting 0x529350 AudioManagerTick")
|
|
self.h.doing("DeepSeek", "")
|
|
self.assertIsNone(next(a for a in self.h.agents() if a["name"] == "DeepSeek")["doing"])
|
|
|
|
def test_wait_returns_on_message_stop_or_timeout(self):
|
|
self.h.clock = __import__("time").time
|
|
self.assertTrue(self.h.wait("Kimi", timeout=0.3, poll=0.05)["timed_out"])
|
|
def later():
|
|
time.sleep(0.4)
|
|
sender = Harbor(self.project)
|
|
sender.send("Claude", ["Kimi"], "your turn")
|
|
sender.close()
|
|
threading.Thread(target=later).start()
|
|
started = time.monotonic()
|
|
r = self.h.wait("Kimi", timeout=10, poll=0.05)
|
|
self.assertLess(time.monotonic() - started, 5)
|
|
self.assertEqual([m["body"] for m in r["messages"]], ["your turn"])
|
|
self.h.set_flag("stop:Kimi", "on", "Claude", "cli")
|
|
self.h.mark_read("Kimi")
|
|
self.assertEqual(self.h.wait("Kimi", timeout=10, poll=0.05)["stopped"], ["stop:Kimi"])
|
|
|
|
def test_digest_and_attention(self):
|
|
self.h.send("Codex", ["Claude"], "a")
|
|
self.h.send("User", ["Codex"], "status please", origin="web")
|
|
item = self.h.submit("DeepSeek", "thing", scope="fn:0x445310", commit="abc")
|
|
self.h.transition("Claude", item["id"], "in_review")
|
|
self.h.send("Claude", ["Kimi"], "please confirm", ack_required=True)
|
|
self.h.touch("OpenCode")
|
|
d = self.h.digest("1h")
|
|
codex = next(a for a in d["agents"] if a["name"] == "Codex")
|
|
self.assertEqual((codex["messages"], codex["threads"]), (1, 1))
|
|
self.assertEqual([e["to_state"] for e in d["item_events"]], ["submitted", "in_review"])
|
|
self.assertEqual(self.h.digest("1h")["to_human"], [])
|
|
self.h.send("Codex", ["User"], "here is the status")
|
|
self.assertEqual([m["body"] for m in self.h.digest("1h")["to_human"]], ["here is the status"])
|
|
self.assertEqual(self.h.attention(), [])
|
|
self.clock.t += 3 * 3600
|
|
kinds = sorted(a["kind"] for a in self.h.attention())
|
|
self.assertEqual(kinds, ["ack_overdue", "agent_quiet", "agent_quiet", "agent_quiet", "agent_quiet", "item_stale"])
|
|
self.h.set_flag("stop:all", "on", "Claude", "cli")
|
|
self.assertNotIn("agent_quiet", [a["kind"] for a in self.h.attention()])
|
|
with self.assertRaises(HarborError):
|
|
self.h.digest("yesterday-ish")
|
|
|
|
|
|
class Claims(HarborCase):
|
|
def test_claims_conflicts_ttl_and_bands(self):
|
|
self.h.claim("Claude", "range:0x440000-0x47FFFF", kind="band", on_behalf_of="DeepSeek")
|
|
self.h.claim("Claude", "range:0x480000-0x4FFFFF", kind="band", on_behalf_of="OpenCode")
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("Kimi", "range:0x400000-0x401000", kind="band")
|
|
d = self.h.claim("DeepSeek", "fn:0x445310", kind="draft")
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("OpenCode", "0x445310", kind="draft")
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("OpenCode", "fn:0x450000", kind="draft")
|
|
self.assertTrue(self.h.claim("OpenCode", "fn:0x490000", kind="draft")["id"])
|
|
self.assertTrue(self.h.claim("Kimi", "fn:0x445310", kind="review")["id"])
|
|
renewed = self.h.claim("DeepSeek", "fn:0x445310", kind="draft", ttl_hours=1)
|
|
self.assertEqual((renewed["renewed"], renewed["id"]), (True, d["id"]))
|
|
self.clock.t += 2 * 3600
|
|
self.assertEqual([c for c in self.h.active_claims(kind="draft") if c["scope"] == "fn:0x445310"], [])
|
|
over = self.h.claim("Claude", "fn:0x450010", kind="draft", override=True)
|
|
self.assertEqual(len(over["conflicts"]), 1)
|
|
self.assertEqual(self.h.release("Claude", scope="fn:0x450010"), [over["id"]])
|
|
self.assertTrue(self.h.claim("Codex", "task:write the README")["id"])
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("Kimi", "task:Write the README")
|
|
|
|
def test_scope_parsing(self):
|
|
self.assertEqual(parse_scope("0x445310")[0], "fn:0x445310")
|
|
self.assertEqual(parse_scope("range:0x440000-0x47ffff")[1:], (0x440000, 0x47FFFF))
|
|
with self.assertRaises(HarborError):
|
|
parse_scope("somewhere")
|
|
|
|
|
|
class Review(HarborCase):
|
|
def test_review_item_lifecycle(self):
|
|
item = self.h.submit("DeepSeek", "RouteCellCost readable", scope="fn:0x445310", commit="abc1234",
|
|
branch="deepseek/decomp")
|
|
self.assertEqual([m["kind"] for m in self.h.inbox("Claude")], ["request"])
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("DeepSeek", item["id"], "approved")
|
|
self.h.transition("Claude", item["id"], "in_review")
|
|
self.h.transition("Claude", item["id"], "changes_requested", note="movsx: signed read")
|
|
self.assertEqual(self.h.inbox("DeepSeek")[0]["kind"], "request")
|
|
self.h.transition("DeepSeek", item["id"], "submitted", commit="def5678")
|
|
self.h.transition("Kimi", item["id"], "approved")
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("Kimi", item["id"], "landed", commit="x")
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("Claude", item["id"], "landed")
|
|
done = self.h.transition("Claude", item["id"], "landed", commit="b4b44f0")
|
|
self.assertEqual((done["state"], done["commit_hash"]), ("landed", "def5678"))
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("Claude", item["id"], "held")
|
|
self.assertEqual([e["to_state"] for e in self.h.item(item["id"])[1]],
|
|
["submitted", "in_review", "changes_requested", "submitted", "approved", "landed"])
|
|
|
|
|
|
class Control(HarborCase):
|
|
def test_resume_one_agent_under_stop_all(self):
|
|
"""Resuming one agent whose own flag is already off still overrides a later 'stop everyone'."""
|
|
self.h.set_flag("stop:Claude", "off", "User", "web")
|
|
self.clock.t += 60
|
|
self.h.set_flag("stop:all", "on", "User", "web")
|
|
self.assertTrue(self.h.stopped("Claude"))
|
|
self.clock.t += 60
|
|
self.h.set_flag("stop:Claude", "off", "User", "web")
|
|
self.assertFalse(self.h.stopped("Claude"))
|
|
self.assertTrue(self.h.stopped("Codex"))
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:Codex", "off", "Claude", "cli") # still the human's alone
|
|
|
|
def test_stop_and_resume_rules(self):
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "on", "User", "cli")
|
|
self.h.set_flag("stop:all", "on", "User", "web", reason="lets stop for now")
|
|
self.assertTrue(self.h.stopped("DeepSeek"))
|
|
self.assertIn("STOP", self.h.brief("DeepSeek"))
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Codex", "cli")
|
|
agent_msg = self.h.send("Kimi", "All", "resume please")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Codex", "cli", relay_of=agent_msg["id"])
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Claude", "cli") # the user set it, not Claude
|
|
user_msg = self.h.send("User", "All", "okay resume everyone", origin="web")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:all", "off", "Codex", "cli", relay_of=user_msg["id"]) # only the human resumes
|
|
self.h.set_flag("stop:all", "off", "User", "web")
|
|
self.assertFalse(self.h.stopped("DeepSeek"))
|
|
self.h.set_flag("stop:Kimi", "on", "Kimi", "cli", reason="out of allowance")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:Kimi", "off", "Kimi", "cli")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:DeepSeek", "on", "Kimi", "cli")
|
|
self.h.set_flag("stop:Codex", "on", "Claude", "cli", reason="mirror the user's stop")
|
|
self.h.set_flag("stop:Codex", "off", "Claude", "cli", reason="set by mistake") # own stop, untouched
|
|
self.h.set_flag("stop:Codex", "on", "Claude", "cli")
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
with self.assertRaises(HarborError):
|
|
self.h.set_flag("stop:Codex", "off", "Claude", "cli") # the user has confirmed it
|
|
|
|
def test_brief_lists_what_needs_attention(self):
|
|
self.h.send("Claude", ["DeepSeek"], "please look", ack_required=True, importance="urgent")
|
|
self.h.claim("DeepSeek", "fn:0x445310", kind="draft", note="writing")
|
|
self.h.submit("DeepSeek", "thing", scope="fn:0x445310", commit="abc")
|
|
text = self.h.brief("DeepSeek")
|
|
for part in ("Unread: 1 (1 urgent)", "Waiting for your acknowledgement", "Your claims", "Your review items"):
|
|
self.assertIn(part, text)
|
|
self.assertIn("Review queue for you: 1 open", self.h.brief("Claude"))
|
|
self.assertNotIn("Review queue for you", self.h.brief("Codex"))
|
|
|
|
def test_legacy_import_is_idempotent_and_read(self):
|
|
legacy = self.dir / "old-chat.md"
|
|
legacy.write_text("# chat\n\n---\n\n[2026-10-01 13:32Z] User → All: rundown please\n\n"
|
|
"[2026-10-01 13:40Z] Codex → User: here it is\nsecond line\n\n"
|
|
"[2026-10-01 13:40Z] Codex → User: here it is\nsecond line\n\n"
|
|
"[2026-10-02 08:00Z] Kimi → Codex: next day\n", encoding="utf-8")
|
|
self.assertEqual(self.h.import_legacy(legacy), {"parsed": 4, "added": 4})
|
|
self.assertEqual(self.h.import_legacy(legacy)["added"], 0)
|
|
self.assertEqual(self.h.inbox("Codex"), [])
|
|
self.assertEqual(sorted(t["topic"] for t in self.h.threads(include_legacy=True)),
|
|
["legacy:2026-10-01", "legacy:2026-10-02"])
|
|
|
|
def test_backup_keeps_newest(self):
|
|
self.h.send("Claude", ["Kimi"], "keep me")
|
|
for i in range(3):
|
|
self.clock.t += 60
|
|
self.h.backup(keep=2)
|
|
copies = sorted(self.project.backups.glob("harbor-*.db"))
|
|
self.assertEqual(len(copies), 2)
|
|
import sqlite3
|
|
with sqlite3.connect(str(copies[-1])) as c:
|
|
self.assertEqual(c.execute("SELECT body FROM messages").fetchone()[0], "keep me")
|
|
|
|
|
|
class Projects(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = temp_dir()
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def test_roster_comes_from_the_project(self):
|
|
p = projects.project_from_dict({"name": "garden", "human": "Kai", "coordinators": ["Lead"],
|
|
"database": str(self.dir / "g.db"),
|
|
"agents": [{"name": "Lead"}, {"name": "Helper"}]})
|
|
h = Harbor(p)
|
|
self.assertEqual(h.agent_names(), ["Lead", "Helper", "Kai"])
|
|
self.assertEqual(sorted(h.send("Lead", "All", "hi")["targets"]), ["Helper", "Kai"])
|
|
with self.assertRaises(HarborError):
|
|
h.send("Kai", "All", "pretend", origin="cli")
|
|
with self.assertRaises(HarborError):
|
|
h.send("Lead", ["Codex"], "not on this team")
|
|
item = h.submit("Helper", "first task") # reviewers default to Lead
|
|
self.assertEqual(h.inbox("Lead")[-1]["refs"]["item"], item["id"])
|
|
h.close()
|
|
smaller = projects.project_from_dict({"name": "garden", "human": "Kai", "database": str(self.dir / "g.db"),
|
|
"agents": [{"name": "Lead"}]})
|
|
h = Harbor(smaller)
|
|
self.assertEqual([a["name"] for a in h.agents()], ["Lead", "Kai"]) # Helper keeps history, inactive
|
|
h.close()
|
|
other = projects.project_from_dict({"name": "other", "database": str(self.dir / "g.db"),
|
|
"agents": [{"name": "Lead"}]})
|
|
with self.assertRaises(HarborError):
|
|
Harbor(other) # the database belongs to garden
|
|
|
|
def test_validation(self):
|
|
for bad in ({"name": "Bad Name"}, {"name": "x", "agents": [{"name": "All"}]},
|
|
{"name": "x", "agents": [{"name": "A"}, {"name": "A"}]},
|
|
{"name": "x", "coordinators": ["Ghost"], "agents": [{"name": "A"}]}):
|
|
with self.assertRaises(projects.ProjectError):
|
|
projects.project_from_dict(bad, home=self.dir)
|
|
|
|
def test_session_identity_for_two_sessions_of_one_app(self):
|
|
made = projects.project_from_dict({
|
|
"name": "pair", "identities": {"wt-codex": "Codex", "wt-codex2": "Codex2", "wt-deepseek": "DeepSeek"},
|
|
"agents": [{"name": "Codex", "host": "codex"}, {"name": "Codex2", "host": "codex"},
|
|
{"name": "DeepSeek", "host": "deepseek"}, {"name": "User", "kind": "human"}]}, home=self.dir)
|
|
two = self.dir / "wt-codex2" / "src"
|
|
two.mkdir(parents=True)
|
|
self.assertEqual(made.session_identity("Codex", two), "Codex2") # same app: the folder decides
|
|
self.assertEqual(made.session_identity("Codex", self.dir), "Codex") # no folder match: installed one
|
|
self.assertEqual(made.session_identity("DeepSeek", two), "DeepSeek") # another app's folder: ignored
|
|
self.assertEqual(made.session_identity(None, two), "Codex2") # nothing pinned: the folder
|
|
|
|
def test_resolution_order_and_identities(self):
|
|
made = make_home(self.dir, names=("first", "second"))
|
|
self.assertEqual(projects.resolve(home=self.dir, cwd=self.dir).name, "first") # home default
|
|
self.assertEqual(projects.resolve("second", home=self.dir).name, "second") # explicit
|
|
tree = self.dir / "repo" / "worktree-deepseek" / "src"
|
|
tree.mkdir(parents=True)
|
|
(self.dir / "repo" / ".harbor.toml").write_text('project = "second"\n', encoding="utf-8")
|
|
found = projects.resolve(home=self.dir, cwd=tree)
|
|
self.assertEqual(found.name, "second") # nearest marker
|
|
self.assertEqual(found.identity_for(tree), "DeepSeek")
|
|
self.assertIsNone(found.identity_for(self.dir))
|
|
with self.assertRaises(projects.ProjectError):
|
|
projects.resolve("missing", home=self.dir)
|
|
self.assertEqual(made["first"].ai_names, ["Claude", "Codex", "DeepSeek", "OpenCode", "Kimi"])
|
|
|
|
|
|
class Concurrency(unittest.TestCase):
|
|
def test_many_processes_write_without_loss(self):
|
|
d = temp_dir()
|
|
project = make_home(d)["test"]
|
|
Harbor(project).close()
|
|
root = Path(__file__).resolve().parents[1]
|
|
script = ("import sys; sys.path.insert(0, r'%s'); from harbor import projects; from harbor.core import Harbor; "
|
|
"h = Harbor(projects.load_named('test', home=r'%s')); "
|
|
"[h.send(sys.argv[1], ['Claude'], f'{sys.argv[1]} message {i}') for i in range(40)]" % (root, d))
|
|
senders = ["Codex", "DeepSeek", "OpenCode", "Kimi", "Codex", "DeepSeek"]
|
|
procs = [subprocess.Popen([sys.executable, "-B", "-c", script, s]) for s in senders]
|
|
self.assertEqual([p.wait(timeout=180) for p in procs], [0] * len(senders))
|
|
h = Harbor(project)
|
|
rows = h.conn.execute("SELECT COUNT(*) n, COUNT(DISTINCT id) u FROM messages").fetchone()
|
|
self.assertEqual((rows["n"], rows["u"]), (240, 240))
|
|
headers = [l for l in project.mirror.read_text(encoding="utf-8").split("\n") if LEGACY_LINE.match(l.strip())]
|
|
self.assertEqual(len(headers), 240)
|
|
self.assertEqual(h.unread_count("Claude")[0], 240)
|
|
h.close()
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|