GodopuandClaude Opus 5
d907359746
- --worktree <path> (CLI over MAM_LOOP_WORKTREE): the source root. Only the
diff, base, head and the empty-set check use it; the wrapper, registry,
checkpoint, loop guard, promotion and cleanup stay on the control root. The
collector is the control root's diff_collect.sh (never the worktree's
unreviewed copy), its arguments shlex-quoted. Validation: an existing
directory, the top of its working tree, the same git common dir as the
control root (".git" is resolved against the directory it came from).
- --base <ref> (CLI only, no env - a stale sha would silently widen the next
range): resolved to a full sha, must be an ancestor of the source HEAD.
- --review-only: requires --base, refuses --plan; Phase 2 is skipped and
logged; correction and comply jobs still go to the creator.
- With --worktree or --base every delegated prompt starts with a context
block (path, branch, base and its origin, head, base..head commit count,
uncommitted changes); without them prompts are byte-identical. All call
sites in run() go through one _dispatch helper.
- Fail-closed guard (only with --worktree/--base): before each review round,
self-review included, an empty change set or one git cannot determine
(exit codes, never output parsing) ends the loop with rc 1 instead of
asking for reviews. This closes the fail-open where a loop started after
an out-of-loop commit sent reviewers "(no changes since base commit)".
- The runner honours MAM_JOBS_DIR and passes --registry-dir to the wrapper.
- run_loop.sh accepts the three options; the legacy shell loop
(MAM_LOOP_PY=0) refuses them before the freeze and the loop lock.
Tests T2-1..T2-14.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017R3Eg7MEpSWnm1oB9ZPP6G
489 lines
20 KiB
Python
489 lines
20 KiB
Python
"""Issue #21 phase-1 prerequisites for the worktree operating model.
|
|
|
|
§1 (item 1) delegate-job wrapper: absolute brief / start-command registry /
|
|
audit-log root / Output Report Path, runnable from any cwd.
|
|
§2 (item 2) loop runner --worktree / --base / --review-only and the
|
|
fail-closed empty change-set guard; the control plane stays in the main
|
|
checkout.
|
|
§3 (item 3) control-plane binding persisted on the session row and forwarded
|
|
to the pane; status / reconcile / stop honour MAM_JOBS_DIR.
|
|
|
|
Every test writes only under tmp_path / mam_sandbox (Issue #20).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
RUN_LOOP_SH = SKILLS / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
|
GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
|
|
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}
|
|
|
|
|
|
def _env(home: Path, extra=None, drop=()) -> dict:
|
|
env = dict(os.environ)
|
|
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
env["AGENT_PYTHON_BIN"] = sys.executable
|
|
env["HOME"] = str(home)
|
|
env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
|
|
env["PYTHONPATH"] = str(SKILLS)
|
|
for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD",
|
|
"MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"):
|
|
env.pop(k, None)
|
|
env.update(UNREACHABLE)
|
|
if extra:
|
|
env.update({k: str(v) for k, v in extra.items()})
|
|
for k in drop:
|
|
env.pop(k, None)
|
|
return env
|
|
|
|
|
|
def _only_job_id(reg: Path) -> str:
|
|
ids = [p.stem for p in reg.glob("*.json")]
|
|
assert len(ids) == 1, ids
|
|
return ids[0]
|
|
|
|
|
|
def _kill_timer(reg: Path, jid: str) -> None:
|
|
try:
|
|
pid = int((reg / f"{jid}.heartbeat.pid").read_text().strip())
|
|
os.kill(pid, signal.SIGKILL)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
|
|
def _dry_submit(cwd: Path, env: dict, *args) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--prompt", "p21",
|
|
"--dry-run", *args],
|
|
capture_output=True, text=True, env=env, cwd=str(cwd), timeout=60)
|
|
|
|
|
|
def _injected(stdout: str, name: str) -> str:
|
|
m = re.search(name + r"='(.*?)'", stdout)
|
|
assert m, stdout
|
|
return m.group(1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §1 item 1 — the wrapper's injected paths are absolute
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_t1_1_injected_paths_are_absolute(tmp_path):
|
|
a = tmp_path / "A"
|
|
a.mkdir()
|
|
real_a = os.path.realpath(a)
|
|
env = _env(tmp_path, {"WORKSPACE_ROOT": a, "DELEGATE_JOB_LOGS_DIR": a / "logs"})
|
|
res = _dry_submit(a, env, "--workdir", str(a))
|
|
assert res.returncode == 0, res.stderr
|
|
jid = _only_job_id(a / ".mam" / "jobs")
|
|
brief = f"{real_a}/.mam/jobs/{jid}/brief.md"
|
|
assert res.stdout.count(brief) == 2, res.stdout
|
|
start = _injected(res.stdout, "start")
|
|
assert f"--registry-dir {real_a}/.mam/jobs " in start
|
|
assert start.startswith(f"DELEGATE_JOB_LOGS_DIR={real_a}/logs "), start
|
|
# no cwd-relative registry path is left anywhere in the injected text
|
|
assert res.stdout.count(f".mam/jobs/{jid}") == res.stdout.count(f"{real_a}/.mam/jobs/{jid}")
|
|
|
|
|
|
def test_t1_2_start_and_success_commands_run_from_another_cwd(tmp_path):
|
|
"""The core regression: a registry and an audit-log root with a space in
|
|
them, the commands executed as written from a different cwd with no
|
|
workspace env - the job advances, the audit trail stays in the control
|
|
root, and nothing is written under the other cwd."""
|
|
ctl = tmp_path / "ctl root"
|
|
reg, logs, wt = ctl / "jobs", ctl / "logs", tmp_path / "wt"
|
|
for d in (reg, logs, wt):
|
|
d.mkdir(parents=True)
|
|
env = _env(tmp_path, {"WORKSPACE_ROOT": ctl, "DELEGATE_JOB_LOGS_DIR": logs})
|
|
res = _dry_submit(ctl, env, "--registry-dir", str(reg), "--workdir", str(ctl),
|
|
"--heartbeat", "7", "--ack-timeout", "11")
|
|
assert res.returncode == 0, res.stderr
|
|
jid = _only_job_id(reg)
|
|
start = _injected(res.stdout, "start")
|
|
success = _injected(res.stdout, "success").replace("<summary>", "done")
|
|
agent_env = _env(tmp_path) # no WORKSPACE_ROOT / MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR
|
|
try:
|
|
r1 = subprocess.run(["bash", "-c", start + " --attempts 1"], capture_output=True, text=True,
|
|
env=agent_env, cwd=str(wt), timeout=120)
|
|
# rc 2 = broker unreachable; the event is recorded locally first (test18 contract)
|
|
assert r1.returncode == 2, r1.stderr
|
|
assert json.loads((reg / f"{jid}.json").read_text())["status"] != "pending"
|
|
r2 = subprocess.run(["bash", "-c", success + " --attempts 1"], capture_output=True, text=True,
|
|
env=agent_env, cwd=str(wt), timeout=120)
|
|
assert r2.returncode == 2, r2.stderr
|
|
assert json.loads((reg / f"{jid}.json").read_text())["status"] == "completed"
|
|
finally:
|
|
_kill_timer(reg, jid)
|
|
audit = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
|
|
published = [json.dumps(e) for e in audit if e.get("event") == "published"]
|
|
assert len(published) == 2, audit
|
|
assert '"started"' in published[0] and '"completed"' in published[1], published
|
|
assert not (wt / ".mam").exists(), "the agent's cwd must not grow a .mam"
|
|
|
|
|
|
def test_t1_3_mam_jobs_dir_is_the_default_registry(tmp_path):
|
|
x, y = tmp_path / "X", tmp_path / "Y"
|
|
env = _env(tmp_path, {"MAM_JOBS_DIR": x, "WORKSPACE_ROOT": tmp_path, "DELEGATE_JOB_LOGS_DIR": tmp_path / "logs"})
|
|
res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path))
|
|
assert res.returncode == 0, res.stderr
|
|
assert len(list(x.glob("*.json"))) == 1
|
|
assert not (tmp_path / ".mam" / "jobs").exists()
|
|
|
|
res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path), "--registry-dir", str(y))
|
|
assert res.returncode == 0, res.stderr
|
|
assert len(list(y.glob("*.json"))) == 1, "an explicit --registry-dir wins over MAM_JOBS_DIR"
|
|
assert len(list(x.glob("*.json"))) == 1
|
|
|
|
|
|
def test_t1_4_wrapper_source_guard():
|
|
"""The loop job type never builds its text under --dry-run (it returns
|
|
early), so both paths are pinned to the shared helpers here."""
|
|
src = WRAPPER.read_text()
|
|
assert ".mam/jobs/$JOB_ID" not in src
|
|
assert src.count('instructions="$(_instructions "$pub")"') == 1
|
|
assert src.count('instructions="$(_instructions "$pub" "$iteration")"') == 1
|
|
assert src.count('pub="$(_pub_cmd ') == 2
|
|
report_lines = [l for l in src.splitlines() if "**Output Report Path**" in l]
|
|
assert len(report_lines) == 2 and all("$REGISTRY_DIR/$JOB_ID/" in l for l in report_lines)
|
|
|
|
|
|
def _sent_texts(state) -> list:
|
|
out = []
|
|
|
|
def walk(o):
|
|
if isinstance(o, dict):
|
|
for k, v in o.items():
|
|
if k == "sent_text" and isinstance(v, str):
|
|
out.append(v)
|
|
else:
|
|
walk(v)
|
|
elif isinstance(o, list):
|
|
for v in o:
|
|
walk(v)
|
|
walk(state)
|
|
return out
|
|
|
|
|
|
def test_t1_5_onboarding_publish_command_is_absolute(mam_sandbox, mock_herdr, mock_agents):
|
|
tmp = mam_sandbox
|
|
script = tmp / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
res = subprocess.run(["bash", str(script), "--workspace", str(tmp), "--agent", "claude",
|
|
"--role", "creator", "--onboard"],
|
|
capture_output=True, text=True, cwd=str(tmp), timeout=180)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
texts = [t for t in _sent_texts(json.loads(Path(mock_herdr).read_text())) if "publish_event.py" in t]
|
|
assert texts, "the onboarding instructions were not injected"
|
|
text = "".join(texts)
|
|
real = os.path.realpath(tmp)
|
|
assert f" {real}/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py " in text, text
|
|
assert f"--registry-dir {real}/.mam/jobs " in text, text
|
|
assert f"DELEGATE_JOB_LOGS_DIR={real}/.mam/delegate_job_logs " in text, text
|
|
assert " .agents/skills/" not in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# §2 item 2 — loop runner --worktree / --base / --review-only, empty-set guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _git(cwd: Path, *args: str) -> str:
|
|
return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True,
|
|
env={**os.environ, **GIT_ID}).stdout.strip()
|
|
|
|
|
|
@pytest.fixture
|
|
def wt_repo(tmp_path, monkeypatch):
|
|
"""main checkout (control root) + a linked worktree on branch feat."""
|
|
main, wt = tmp_path / "main", tmp_path / "wt"
|
|
main.mkdir()
|
|
_git(main, "init", "-q")
|
|
(main / "a.txt").write_text("a\n")
|
|
_git(main, "add", "a.txt")
|
|
_git(main, "commit", "-q", "-m", "init")
|
|
_git(main, "worktree", "add", "-q", "-b", "feat", str(wt))
|
|
for k, v in GIT_ID.items():
|
|
monkeypatch.setenv(k, v)
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(main))
|
|
monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(main / ".mam" / "jobs"))
|
|
monkeypatch.delenv("MAM_LOOP_WORKTREE", raising=False)
|
|
monkeypatch.delenv("MAM_LOOP_BASE", raising=False)
|
|
monkeypatch.setenv("MAM_STATE_JSON", json.dumps({"herdr_sessions": [
|
|
{"name": "creator-claude-01", "agent": "claude", "role": "creator", "status": "running"},
|
|
{"name": "reviewer-claude-01", "agent": "claude", "role": "reviewer", "status": "running"},
|
|
]}))
|
|
return main, wt
|
|
|
|
|
|
def _runner(*extra):
|
|
from lib_py.loop import runner as lr
|
|
return lr.LoopRunner(["--creator", "creator-claude-01", "--task", "t21", *extra])
|
|
|
|
|
|
def _drive(monkeypatch, tmp_path, argv, on_worker=None):
|
|
"""run() end to end with the delegation stubbed: every job 'completes' and
|
|
every report is a PASS. Returns (rc, [(role, prompt)], runner)."""
|
|
from lib_py.loop import runner as lr
|
|
monkeypatch.setattr(lr.signal, "signal", lambda *a, **k: None)
|
|
r = lr.LoopRunner(["--creator", "creator-claude-01", "--task", "t21",
|
|
"--reviewer", "reviewer-claude-01", "--max-loop", "1", *argv])
|
|
calls = []
|
|
report = tmp_path / "pass-report.md"
|
|
report.write_text("looks fine\n\n[VERDICT: PASS]\n")
|
|
|
|
def fake_delegate(session, agent_type, role, prompt):
|
|
calls.append((role, prompt))
|
|
if role == "Worker" and on_worker:
|
|
on_worker(r)
|
|
return f"j{len(calls)}"
|
|
|
|
monkeypatch.setattr(r, "delegate_job", fake_delegate)
|
|
monkeypatch.setattr(r, "wait_for_job", lambda *a, **k: True)
|
|
monkeypatch.setattr(r, "find_report", lambda jid: report)
|
|
return r.run(), calls, r
|
|
|
|
|
|
def _roles(calls, role):
|
|
return [p for (ro, p) in calls if ro == role]
|
|
|
|
|
|
def test_t2_1_diff_and_head_come_from_the_worktree(wt_repo):
|
|
main, wt = wt_repo
|
|
(wt / "a.txt").write_text("a\nworktree change\n")
|
|
(main / "b.txt").write_text("main-only change\n")
|
|
r = _runner("--worktree", str(wt))
|
|
r.parse_args()
|
|
r.base_commit = r._source_head()
|
|
assert r._source_head() == _git(wt, "rev-parse", "HEAD")
|
|
diff = r.collect_changes_diff()
|
|
assert "worktree change" in diff
|
|
assert "main-only change" not in diff and "b.txt" not in diff
|
|
|
|
|
|
def test_t2_2_invalid_worktrees_are_refused(wt_repo, tmp_path):
|
|
main, wt = wt_repo
|
|
(wt / "sub").mkdir()
|
|
(tmp_path / "plain").mkdir()
|
|
other = tmp_path / "other"
|
|
other.mkdir()
|
|
_git(other, "init", "-q")
|
|
for bad in (tmp_path / "missing", tmp_path / "plain", wt / "sub", other):
|
|
with pytest.raises(SystemExit) as e:
|
|
_runner("--worktree", str(bad)).parse_args()
|
|
assert e.value.code == 1, bad
|
|
|
|
|
|
def test_t2_3_delegation_stays_on_the_control_root(wt_repo, monkeypatch):
|
|
main, wt = wt_repo
|
|
from lib_py.loop import runner as lr
|
|
monkeypatch.setenv("MAM_HEARTBEAT_SEC", "20")
|
|
monkeypatch.setenv("MAM_HEARTBEAT_MISSES", "3")
|
|
monkeypatch.setenv("MAM_ACK_TIMEOUT_S", "45")
|
|
r = _runner("--worktree", str(wt))
|
|
r.parse_args()
|
|
seen = []
|
|
|
|
def fake_run(cmd, **kw):
|
|
seen.append((cmd, kw))
|
|
|
|
class P:
|
|
returncode = 0
|
|
stdout = "registered job: cafe21\n"
|
|
stderr = ""
|
|
return P()
|
|
|
|
monkeypatch.setattr(lr.subprocess, "run", fake_run)
|
|
assert r.delegate_job("creator-claude-01", "claude", "Worker", "go") == "cafe21"
|
|
cmd, kw = seen[0]
|
|
assert os.path.realpath(kw["cwd"]) == os.path.realpath(main)
|
|
assert cmd[1].endswith("multi-agent-mux-delegate-job/multi-agent-mux-delegate-job")
|
|
assert not cmd[1].startswith(os.path.realpath(wt))
|
|
assert cmd[cmd.index("--registry-dir") + 1] == str(main / ".mam" / "jobs")
|
|
assert cmd[-6:] == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]
|
|
|
|
|
|
def test_t2_4_checkpoint_lives_in_the_control_root(wt_repo):
|
|
main, wt = wt_repo
|
|
r = _runner("--worktree", str(wt))
|
|
r.parse_args()
|
|
r._record_checkpoint()
|
|
data = json.loads((main / ".mam" / "loop-checkpoint.json").read_text())
|
|
assert data["worktree_root"] == os.path.realpath(wt)
|
|
assert data["base_source"] == "head" and data["review_only"] is False
|
|
assert data["head_commit"] == _git(wt, "rev-parse", "HEAD")
|
|
assert not (wt / ".mam").exists()
|
|
|
|
|
|
def test_t2_5_prompts_carry_the_worktree_context(wt_repo, monkeypatch, tmp_path):
|
|
main, wt = wt_repo
|
|
head0 = _git(wt, "rev-parse", "HEAD")
|
|
|
|
def implement(_r):
|
|
(wt / "a.txt").write_text("a\nimplemented in wt\n")
|
|
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)], on_worker=implement)
|
|
assert rc == 0
|
|
worker, review = _roles(calls, "Worker")[0], _roles(calls, "Reviewer")[0]
|
|
for p in (worker, review):
|
|
assert p.startswith("[워크트리 컨텍스트]\n")
|
|
assert f"- 소스 경로: {os.path.realpath(wt)}\n" in p
|
|
assert f"- base: {head0} (출처: 루프 시작 시 HEAD)" in p
|
|
assert f"- head: {head0} (base..head 커밋: 0개" in p
|
|
assert "미커밋 변경: 있음" in review and "implemented in wt" in review
|
|
|
|
|
|
def test_t2_5b_prompts_unchanged_without_options(wt_repo, monkeypatch, tmp_path):
|
|
main, _ = wt_repo
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(main))
|
|
|
|
def implement(_r):
|
|
(main / "a.txt").write_text("a\nimplemented in main\n")
|
|
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, [], on_worker=implement)
|
|
assert rc == 0
|
|
assert all("[워크트리 컨텍스트]" not in p for _, p in calls)
|
|
assert _roles(calls, "Worker")[0].startswith("계획서가 존재하지 않으므로")
|
|
assert _roles(calls, "Reviewer")[0].startswith("다음 구현 사항(작업 목표: t21)")
|
|
|
|
|
|
def test_t2_6_worktree_env_fallback_and_no_base_env(wt_repo, tmp_path):
|
|
main, wt = wt_repo
|
|
wt2 = tmp_path / "wt2"
|
|
_git(main, "worktree", "add", "-q", "-b", "feat2", str(wt2))
|
|
os.environ["MAM_LOOP_WORKTREE"] = str(wt)
|
|
os.environ["MAM_LOOP_BASE"] = _git(main, "rev-parse", "HEAD")
|
|
try:
|
|
r = _runner()
|
|
r.parse_args()
|
|
assert r.worktree_root == Path(os.path.realpath(wt)) and r.worktree_given
|
|
assert r.base_ref is None and r.base_sha == "" and r.base_source == "head", "MAM_LOOP_BASE is ignored (D17)"
|
|
r = _runner("--worktree", str(wt2))
|
|
r.parse_args()
|
|
assert r.worktree_root == Path(os.path.realpath(wt2)), "the CLI wins over MAM_LOOP_WORKTREE"
|
|
finally:
|
|
os.environ.pop("MAM_LOOP_WORKTREE", None)
|
|
os.environ.pop("MAM_LOOP_BASE", None)
|
|
|
|
|
|
def test_t2_7_runner_reads_mam_jobs_dir(wt_repo, tmp_path, monkeypatch):
|
|
x = tmp_path / "elsewhere" / "jobs"
|
|
(x / "jid21" / "rev-reports").mkdir(parents=True)
|
|
(x / "jid21.json").write_text(json.dumps({"status": "completed"}))
|
|
(x / "jid21" / "rev-reports" / "report-final.md").write_text("r\n")
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(x))
|
|
r = _runner()
|
|
r.parse_args()
|
|
assert r.wait_for_job("jid21", max_wait=2) is True
|
|
assert r.find_report("jid21") == x / "jid21" / "rev-reports" / "report-final.md"
|
|
|
|
|
|
@pytest.mark.parametrize("opts,env", [
|
|
(["--worktree", "/nowhere"], {}),
|
|
(["--base", "HEAD"], {}),
|
|
(["--review-only"], {}),
|
|
([], {"MAM_LOOP_WORKTREE": "/nowhere"}),
|
|
])
|
|
def test_t2_8_legacy_shell_loop_refuses_the_new_options(tmp_path, opts, env):
|
|
marker = tmp_path / "loop-marker"
|
|
run_env = {**os.environ, "MAM_LOOP_PY": "0", "MAM_LOOP_NO_FREEZE": "1",
|
|
"MAM_LOOP_MARKER": str(marker), **env}
|
|
res = subprocess.run(["bash", str(RUN_LOOP_SH), "--creator", "c", "--task", "t", *opts],
|
|
capture_output=True, text=True, env=run_env, cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 1
|
|
assert "require the Python loop (MAM_LOOP_PY=1)" in res.stderr
|
|
assert not marker.exists(), "refused before the loop lock"
|
|
|
|
|
|
def _commit_impl(wt: Path) -> str:
|
|
"""Implementation committed in the worktree outside the loop; returns the pre-implementation sha."""
|
|
before = _git(wt, "rev-parse", "HEAD")
|
|
(wt / "impl.txt").write_text("impl line\n")
|
|
_git(wt, "add", "impl.txt")
|
|
_git(wt, "commit", "-q", "-m", "impl")
|
|
return before
|
|
|
|
|
|
def test_t2_9a_committed_impl_without_base_is_refused(wt_repo, monkeypatch, tmp_path, capsys):
|
|
main, wt = wt_repo
|
|
_commit_impl(wt)
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)])
|
|
assert rc == 1
|
|
assert _roles(calls, "Reviewer") == []
|
|
assert "empty change set since base" in capsys.readouterr().err
|
|
|
|
|
|
def test_t2_9b_review_only_reviews_the_committed_range(wt_repo, monkeypatch, tmp_path):
|
|
main, wt = wt_repo
|
|
before = _commit_impl(wt)
|
|
rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", before, "--review-only"])
|
|
assert rc == 0
|
|
assert _roles(calls, "Worker") == [], "Phase 2 is skipped"
|
|
review = _roles(calls, "Reviewer")[0]
|
|
assert "+impl line" in review
|
|
assert f"- base: {before} (출처: --base)" in review
|
|
assert "(base..head 커밋: 1개, 미커밋 변경: 없음)" in review
|
|
|
|
|
|
def test_t2_10_base_must_be_an_ancestor(wt_repo):
|
|
main, wt = wt_repo
|
|
_git(main, "commit", "-q", "--allow-empty", "-m", "side") # on main's branch, not in feat
|
|
side = _git(main, "rev-parse", "HEAD")
|
|
for ref in (side, "no-such-ref"):
|
|
with pytest.raises(SystemExit) as e:
|
|
_runner("--worktree", str(wt), "--base", ref).parse_args()
|
|
assert e.value.code == 1, ref
|
|
|
|
|
|
def test_t2_11_review_only_needs_base_and_no_plan(wt_repo):
|
|
main, wt = wt_repo
|
|
head = _git(wt, "rev-parse", "HEAD")
|
|
for extra in (["--review-only"], ["--review-only", "--base", head, "--plan"]):
|
|
with pytest.raises(SystemExit) as e:
|
|
_runner("--worktree", str(wt), *extra).parse_args()
|
|
assert e.value.code == 1, extra
|
|
|
|
|
|
def test_t2_12_guard_is_off_without_options(wt_repo, monkeypatch, tmp_path):
|
|
"""Compatibility: no --worktree/--base, nothing changed - the review is
|
|
still requested exactly as before."""
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, [])
|
|
assert rc == 0
|
|
assert len(_roles(calls, "Reviewer")) == 1
|
|
|
|
|
|
def test_t2_13_base_without_review_only_still_implements(wt_repo, monkeypatch, tmp_path):
|
|
main, wt = wt_repo
|
|
before = _commit_impl(wt)
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", before])
|
|
assert rc == 0
|
|
assert len(_roles(calls, "Worker")) == 1
|
|
assert "+impl line" in _roles(calls, "Reviewer")[0]
|
|
|
|
|
|
def test_t2_14_undeterminable_change_set_is_refused(wt_repo, monkeypatch, tmp_path, capsys):
|
|
main, wt = wt_repo
|
|
|
|
def corrupt(r):
|
|
r.base_commit = "0" * 40 # not an object: git diff exits 128
|
|
|
|
rc, calls, _ = _drive(monkeypatch, tmp_path, ["--worktree", str(wt)], on_worker=corrupt)
|
|
assert rc == 1
|
|
assert _roles(calls, "Reviewer") == []
|
|
assert "refusing to request reviews on no evidence" in capsys.readouterr().err
|