istos
a1124ec991
Against a real repo with a real branch: a real merge, a real conflict, the wrong-branch refusal, a crash, and a second complete arriving from inside the first one's merge — which is refused, starts no second merge, and does not release the run it lost to. Plus the registry's own rules and the card-face invariants in board.html.
363 lines
16 KiB
Python
363 lines
16 KiB
Python
"""A card being merged and cleaned up looks like it, and holds still (task 38).
|
|
|
|
"Merge & clean up" parks the drive, merges, deletes a worktree and a
|
|
branch and moves the card — seconds to a minute of destructive work the
|
|
card used to render as "waiting on you", fully draggable, every action
|
|
armed. The state is server-held (`state.COMPLETING`), so these run the
|
|
real thing against a real git repo: a real merge, a real conflict, and a
|
|
real second request arriving mid-run.
|
|
|
|
python3 -m unittest discover -s tests -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO / "manager" / "core"))
|
|
|
|
import config # noqa: E402
|
|
import drive as drive_mod # noqa: E402
|
|
import github # noqa: E402
|
|
import state # noqa: E402
|
|
|
|
BOARD_HTML = REPO / "manager" / "core" / "board.html"
|
|
FILENAME = "38-a-card-being-completed.md"
|
|
STEM = FILENAME[:-3]
|
|
BRANCH = f"task/{STEM}"
|
|
|
|
|
|
def git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["git", "-C", str(cwd), *args],
|
|
capture_output=True, text=True)
|
|
|
|
|
|
CARD = ("# 38 — A card with work on it\n\n"
|
|
"**Status:** Review\n**Priority:** Medium\n\n"
|
|
"Body text long enough that git sees a rename rather than a delete\n"
|
|
"and an add when the file moves between stage directories.\n")
|
|
|
|
|
|
class TheRegistry(unittest.TestCase):
|
|
"""state.COMPLETING: one claim per card, the latest narrated step, and
|
|
a release that always publishes."""
|
|
|
|
def setUp(self):
|
|
state.COMPLETING.clear()
|
|
self.addCleanup(state.COMPLETING.clear)
|
|
state.BOARD_EVENTS.clear()
|
|
self.addCleanup(state.BOARD_EVENTS.clear)
|
|
self.sent: list[dict] = []
|
|
original = state.broadcast
|
|
self.addCleanup(setattr, state, "broadcast", original)
|
|
state.broadcast = self.sent.append
|
|
# persist() writes the event log; keep the test off the real one
|
|
self.tmp = Path(tempfile.mkdtemp(prefix="bench-registry-"))
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
self.addCleanup(setattr, config, "SESSIONS_DIR", config.SESSIONS_DIR)
|
|
config.SESSIONS_DIR = self.tmp
|
|
|
|
def published(self) -> list[dict]:
|
|
return [m["completing"] for m in self.sent if m["type"] == "completing"]
|
|
|
|
def test_a_second_claim_is_refused_rather_than_queued(self):
|
|
self.assertTrue(state.claim_completing(FILENAME, "merging…"))
|
|
self.assertFalse(state.claim_completing(FILENAME, "merging again…"),
|
|
"one completion per card, ever")
|
|
self.assertEqual(state.COMPLETING[FILENAME]["step"], "merging…",
|
|
"the refused claim must not overwrite the live one")
|
|
|
|
def test_the_claim_reaches_the_browser_immediately(self):
|
|
state.claim_completing(FILENAME, "merging and cleaning up…")
|
|
self.assertEqual(self.published()[-1][FILENAME]["step"],
|
|
"merging and cleaning up…")
|
|
|
|
def test_narrated_steps_become_the_card_s_line(self):
|
|
state.claim_completing(FILENAME, "merging and cleaning up…")
|
|
state.record_board_event({"kind": "agent", "actor": "board",
|
|
"file": FILENAME, "summary": "parking the drive"})
|
|
self.assertEqual(state.COMPLETING[FILENAME]["step"], "parking the drive")
|
|
self.assertEqual(self.published()[-1][FILENAME]["step"], "parking the drive")
|
|
|
|
def test_another_card_s_events_are_not_this_card_s_step(self):
|
|
state.claim_completing(FILENAME, "merging and cleaning up…")
|
|
before = len(self.published())
|
|
state.record_board_event({"kind": "agent", "actor": "board",
|
|
"file": "07-something-else.md",
|
|
"summary": "merged task/07 into main"})
|
|
self.assertEqual(state.COMPLETING[FILENAME]["step"], "merging and cleaning up…")
|
|
self.assertEqual(len(self.published()), before,
|
|
"an unrelated event must not republish the registry")
|
|
|
|
def test_release_empties_it_and_says_so(self):
|
|
state.claim_completing(FILENAME, "merging…")
|
|
state.release_completing(FILENAME)
|
|
self.assertEqual(state.COMPLETING, {})
|
|
self.assertEqual(self.published()[-1], {})
|
|
|
|
def test_releasing_what_was_never_claimed_is_silent(self):
|
|
state.release_completing(FILENAME)
|
|
self.assertEqual(self.published(), [])
|
|
|
|
def test_the_public_view_is_a_copy(self):
|
|
state.claim_completing(FILENAME, "merging…")
|
|
snapshot = state.completing_public()
|
|
snapshot[FILENAME]["step"] = "tampered"
|
|
self.assertEqual(state.COMPLETING[FILENAME]["step"], "merging…")
|
|
|
|
|
|
class ACompletionInFlight(unittest.TestCase):
|
|
"""One board, one card in review/ with a branch and a commit on it.
|
|
Sync off, so the merge is the local one."""
|
|
|
|
def setUp(self):
|
|
# resolve(): macOS tempdirs sit behind the /var → /private/var
|
|
# symlink and git reports the resolved path.
|
|
self.tmp = Path(tempfile.mkdtemp(prefix="bench-completing-")).resolve()
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
self.board = self.tmp / "board"
|
|
self.board.mkdir()
|
|
subprocess.run(["git", "init", "-q", "-b", "main", str(self.board)],
|
|
check=True, capture_output=True)
|
|
git(self.board, "config", "user.name", "ada")
|
|
git(self.board, "config", "user.email", "ada@example.com")
|
|
for slug in config.STAGE_DIRS:
|
|
(self.board / "tasks" / slug).mkdir(parents=True)
|
|
(self.board / "code.txt").write_text("shipped\n", encoding="utf-8")
|
|
git(self.board, "add", "-A")
|
|
git(self.board, "commit", "-q", "-m", "root")
|
|
(self.board / "tasks" / "review" / FILENAME).write_text(CARD, encoding="utf-8")
|
|
|
|
self.patch(REPO=self.board, TASKS=self.board / "tasks",
|
|
WORKTREES=self.tmp / "worktrees", SESSIONS_DIR=self.tmp / "sessions",
|
|
SYNC=False, COMMIT_MOVES=False)
|
|
self.worktree = config.WORKTREES / STEM
|
|
git(self.board, "worktree", "add", "-q", "-b", BRANCH, str(self.worktree))
|
|
(self.worktree / "feature.txt").write_text("the work\n", encoding="utf-8")
|
|
git(self.worktree, "add", "-A")
|
|
git(self.worktree, "commit", "-q", "-m", "the work")
|
|
|
|
state.COMPLETING.clear()
|
|
self.addCleanup(state.COMPLETING.clear)
|
|
state.BOARD_EVENTS.clear()
|
|
self.addCleanup(state.BOARD_EVENTS.clear)
|
|
state.EXPECTED_MOVES.clear()
|
|
self.addCleanup(state.EXPECTED_MOVES.clear)
|
|
self.addCleanup(setattr, drive_mod, "DRIVE", drive_mod.DRIVE)
|
|
drive_mod.DRIVE = None
|
|
self.sent: list[dict] = []
|
|
self.addCleanup(setattr, state, "broadcast", state.broadcast)
|
|
state.broadcast = self.sent.append
|
|
|
|
def patch(self, **values) -> None:
|
|
for attr, value in values.items():
|
|
self.addCleanup(setattr, config, attr, getattr(config, attr))
|
|
setattr(config, attr, value)
|
|
|
|
def stage_of(self) -> str | None:
|
|
for slug in config.STAGE_DIRS:
|
|
if (self.board / "tasks" / slug / FILENAME).is_file():
|
|
return slug
|
|
return None
|
|
|
|
def steps(self) -> list[str]:
|
|
return [m["completing"][FILENAME]["step"] for m in self.sent
|
|
if m["type"] == "completing" and FILENAME in m["completing"]]
|
|
|
|
def narrated(self) -> list[str]:
|
|
return [e["summary"] for e in state.BOARD_EVENTS if e.get("file") == FILENAME]
|
|
|
|
def conflict(self) -> None:
|
|
"""The same line, two ways: main and the branch cannot be merged."""
|
|
for where, text in ((self.board, "main's version\n"),
|
|
(self.worktree, "the branch's version\n")):
|
|
(where / "contested.txt").write_text(text, encoding="utf-8")
|
|
git(where, "add", "-A")
|
|
git(where, "commit", "-q", "-m", f"contested in {where.name}")
|
|
|
|
# — the card is busy from the first step to the last —
|
|
|
|
def test_the_claim_is_held_before_the_drive_is_parked(self):
|
|
seen = {}
|
|
drive_mod.DRIVE = {"task": FILENAME, "status": "up", "proc": None,
|
|
"pgid": None, "started": 0, "log": str(self.tmp / "d.log")}
|
|
self.addCleanup(setattr, drive_mod, "stop", drive_mod.stop)
|
|
|
|
def fake_stop():
|
|
seen["claimed"] = FILENAME in state.COMPLETING
|
|
drive_mod.DRIVE["status"] = "parked"
|
|
drive_mod.stop = fake_stop
|
|
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertTrue(seen.get("claimed"),
|
|
"the card must already be busy when the drive is parked")
|
|
|
|
def test_the_card_tracks_the_steps_the_ticker_narrates(self):
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
steps = self.steps()
|
|
self.assertEqual(steps[0], "merging and cleaning up…")
|
|
self.assertTrue(any("merged" in s and "into main" in s for s in steps),
|
|
f"the merge never reached the card: {steps}")
|
|
self.assertTrue(any(s.startswith("cleaned up:") for s in steps),
|
|
f"the cleanup never reached the card: {steps}")
|
|
for summary in self.narrated():
|
|
self.assertIn(summary, steps,
|
|
"every step narrated to the ticker lands on the card too")
|
|
|
|
def test_a_finished_completion_gives_the_card_back(self):
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertEqual(self.stage_of(), "done")
|
|
self.assertEqual(state.COMPLETING, {})
|
|
self.assertEqual([m["completing"] for m in self.sent
|
|
if m["type"] == "completing"][-1], {},
|
|
"the last word to the browser is 'nothing is busy'")
|
|
|
|
# — every failure gives it back too —
|
|
|
|
def test_a_merge_conflict_releases_the_card(self):
|
|
self.conflict()
|
|
|
|
with self.assertRaises(ValueError) as caught:
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertIn("merge conflict", str(caught.exception))
|
|
self.assertEqual(self.stage_of(), "review", "the card stays put")
|
|
self.assertEqual(state.COMPLETING, {}, "…and comes back to life")
|
|
self.assertEqual(git(self.board, "rev-parse", "--verify", "--quiet",
|
|
BRANCH).returncode, 0, "the branch survives a conflict")
|
|
|
|
def test_the_wrong_branch_refusal_releases_the_card(self):
|
|
git(self.board, "checkout", "-q", "-b", "side")
|
|
|
|
with self.assertRaises(ValueError) as caught:
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertIn("not main", str(caught.exception))
|
|
self.assertEqual(state.COMPLETING, {})
|
|
|
|
def test_a_crash_mid_completion_releases_the_card(self):
|
|
self.addCleanup(setattr, github, "_merge_locally", github._merge_locally)
|
|
|
|
def boom(filename, branch):
|
|
raise OSError("git went away")
|
|
github._merge_locally = boom
|
|
|
|
with self.assertRaises(OSError):
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertEqual(state.COMPLETING, {}, "a finally, not a release per exit")
|
|
|
|
# — and the second request is refused, not run —
|
|
|
|
def test_a_second_request_mid_completion_starts_no_second_merge(self):
|
|
real = github._merge_locally
|
|
self.addCleanup(setattr, github, "_merge_locally", real)
|
|
calls, refused = [], []
|
|
|
|
def merge_then_race(filename, branch):
|
|
calls.append(branch)
|
|
try:
|
|
github.complete_task(FILENAME, "review")
|
|
except ValueError as exc:
|
|
refused.append(str(exc))
|
|
real(filename, branch)
|
|
github._merge_locally = merge_then_race
|
|
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertEqual(len(refused), 1, "the second request must be refused")
|
|
self.assertIn("already being completed", refused[0])
|
|
self.assertEqual(len(calls), 1, "and no second merge attempted")
|
|
self.assertEqual(self.stage_of(), "done", "the first run finished normally")
|
|
|
|
def test_the_refusal_does_not_release_the_run_it_lost_to(self):
|
|
state.claim_completing(FILENAME, "merging and cleaning up…")
|
|
|
|
with self.assertRaises(ValueError):
|
|
github.complete_task(FILENAME, "review")
|
|
|
|
self.assertIn(FILENAME, state.COMPLETING,
|
|
"a refused caller must never release what it did not take")
|
|
self.assertEqual(self.stage_of(), "review")
|
|
|
|
|
|
class ServedToTheBrowser(unittest.TestCase):
|
|
def test_state_payload_carries_the_registry(self):
|
|
import httpd
|
|
self.assertIn("completing", httpd.state_payload())
|
|
|
|
def test_a_restart_leaves_nothing_stuck(self):
|
|
"""The registry is memory, never disk: a board killed mid-merge
|
|
comes back with every card renderable from the files alone."""
|
|
source = (REPO / "manager" / "core" / "state.py").read_text(encoding="utf-8")
|
|
body = re.search(r"def claim_completing.*?\n(?=\ndef )", source, re.S)
|
|
self.assertIsNotNone(body, "claim_completing is gone")
|
|
for persistent in ("persist(", "write_text", "open("):
|
|
self.assertNotIn(persistent, body.group(0),
|
|
"the claim must not outlive the process")
|
|
|
|
|
|
class TheCardFace(unittest.TestCase):
|
|
"""board.html is a single file with no frontend runner — these are the
|
|
source-level invariants of the surface this card adds."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.html = BOARD_HTML.read_text(encoding="utf-8")
|
|
cls.card = re.search(r"function cardFor\(task\) \{.*?\n\}", cls.html, re.S).group(0)
|
|
|
|
def test_the_card_renders_from_the_server_s_truth(self):
|
|
self.assertIn("(S.state.completing || {})[task.file]", self.card,
|
|
"the busy state comes from /api/state, not from a click")
|
|
self.assertIn("msg.type === 'completing'", self.html,
|
|
"…and follows the SSE stream between full loads")
|
|
|
|
def test_it_wears_the_working_vocabulary_and_no_new_colour(self):
|
|
completing = re.search(r"if \(completing\) \{\n\s*//.*?\n\s*//.*?\n(.*?)\n \}",
|
|
self.card, re.S).group(1)
|
|
self.assertIn("text: 'completing'", completing)
|
|
self.assertIn("var(--accent)", completing)
|
|
for colour in ("--alarm", "--calm", "--idle"):
|
|
self.assertNotIn(colour, completing, "busy is not an alarm or a verdict")
|
|
flat = self.html.replace(" ", "").replace("\n", "")
|
|
self.assertIn(".card.completing{border-color:color-mix(inoklab,var(--accent)55%,var(--border))", flat)
|
|
self.assertIn(".card.pill.status.breathing{animation:breathe", flat)
|
|
self.assertIn("""class="pill status${completing ? ' breathing' : ''}\"""", self.card)
|
|
|
|
def test_it_says_which_step(self):
|
|
self.assertIn("esc(completing.step", self.card,
|
|
"the card carries the latest narrated step")
|
|
|
|
def test_it_holds_still(self):
|
|
self.assertIn("el.draggable = !completing;", self.card)
|
|
|
|
def test_its_actions_are_suppressed_not_merely_ignored(self):
|
|
self.assertRegex(self.card, r"if \(completing\) \{\n(\s*//[^\n]*\n)+\s*\} else if \(agent\)",
|
|
"the action builder must fall through to nothing")
|
|
self.assertIn("if (task.stage === 'review' && !completing) {", self.card,
|
|
"the drive chips go with the worktree being removed")
|
|
self.assertIn("if (hasBranch && !completing &&", self.card,
|
|
"…and so do the project's command chips")
|
|
|
|
def test_the_sheet_still_only_opens_for_a_card_with_work(self):
|
|
"""Untouched path: no branch and no PR means no sheet, so no
|
|
completion and no flash of busy."""
|
|
mover = re.search(r"async function move\(file, from, to\) \{.*?\n\}",
|
|
self.html, re.S).group(0)
|
|
self.assertIn("task && (task.pr || (S.state.branches || []).includes(stem))", mover)
|
|
self.assertIn("completeSheet(task, from)", mover)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|