678 lines
36 KiB
Python
678 lines
36 KiB
Python
"""Local recovery is task-scoped and cannot grant delivery readiness."""
|
||
import json
|
||
import os
|
||
import shlex
|
||
import subprocess
|
||
import sys
|
||
import unittest
|
||
|
||
# 允许 `python3 -m unittest tests/test_task_recovery.py` 直接运行（与 discover 模式一致地导入同目录夹具）。
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import test_task as lifecycle # noqa: E402
|
||
|
||
|
||
class TaskRecoveryTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.fixture = lifecycle.TaskLifecycleTests(methodName="runTest")
|
||
self.fixture.setUp()
|
||
self.addCleanup(self.fixture.doCleanups)
|
||
self.wt = self.fixture.worktree()
|
||
self.card = self.fixture.record_task_card(self.wt)
|
||
self.receipt_number = 0
|
||
|
||
def precheck(self, mode="local", cwd=None):
|
||
f = self.fixture
|
||
return f.task(cwd or self.wt, "precheck", check=False,
|
||
env=f.env | {"TASK_WORK_MODE": mode})
|
||
|
||
def capture(self, mode="local"):
|
||
f = self.fixture
|
||
self.receipt_number += 1
|
||
path = f.root / f"recovery-{self.receipt_number}.json"
|
||
result = f.run_cmd(self.wt, "bash", "scripts/evidence.sh", check=False,
|
||
env=f.env | {"TASK_WORK_MODE": mode, "RECEIPT": str(path),
|
||
"CMD": "make precheck"})
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
return path
|
||
|
||
def resume(self, receipt, mode="local", cwd=None):
|
||
f = self.fixture
|
||
return f.task(cwd or self.wt, "resume", check=False,
|
||
env=f.env | {"TASK_WORK_MODE": mode, "PRECHECK_RECEIPT": str(receipt)})
|
||
|
||
def save_card_change(self):
|
||
f = self.fixture
|
||
f.git(self.wt, "add", str(self.card.relative_to(self.wt)))
|
||
f.git(self.wt, "commit", "-m", "Update fixture registration")
|
||
|
||
def local_authorization_scope(self):
|
||
"""Capture the explicit registered in-progress task used as local-mode authorization.
|
||
|
||
The card carries the task scope (worktree path, branch, target branch, and R0
|
||
initial baseline) that authorize LOCAL_READY under a registered task, regardless
|
||
of whether the authoritative origin is currently reachable. This small synthetic
|
||
fixture exposes that scope as a single value so offline-local tests can name it."""
|
||
f = self.fixture
|
||
branch = f.git(self.wt, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
return {
|
||
"wt": self.wt,
|
||
"card": self.card,
|
||
"branch": branch,
|
||
"target_branch": "main",
|
||
"r0": r0,
|
||
}
|
||
|
||
def test_local_precheck_allows_known_lag_without_changing_head(self):
|
||
f = self.fixture
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
f.advance_remote()
|
||
strict = self.precheck(mode="delivery")
|
||
self.assertNotEqual(strict.returncode, 0)
|
||
self.assertIn("SYNC_REQUIRED", strict.stdout)
|
||
local = self.precheck()
|
||
self.assertEqual(local.returncode, 0, local.stdout)
|
||
self.assertIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertNotIn("[verdict] READY", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
|
||
def test_local_precheck_never_advertises_delivery_even_when_current(self):
|
||
result = self.precheck()
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("[verdict] LOCAL_READY", result.stdout)
|
||
self.assertNotIn("[verdict] READY", result.stdout)
|
||
|
||
def test_local_precheck_rejects_main_and_missing_registration(self):
|
||
result = self.precheck(cwd=self.fixture.repo)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.card.unlink()
|
||
self.save_card_change()
|
||
result = self.precheck()
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
def test_local_precheck_rejects_foreign_registration(self):
|
||
self.card.write_text(self.card.read_text().replace(str(self.wt), str(self.fixture.repo)))
|
||
self.save_card_change()
|
||
result = self.precheck()
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
def test_local_precheck_rejects_wrong_branch_registration(self):
|
||
self.card.write_text(self.card.read_text().replace("分支：feature/test", "分支：feature/other"))
|
||
self.save_card_change()
|
||
self.assertNotEqual(self.precheck().returncode, 0)
|
||
|
||
def test_local_precheck_rejects_invalid_baseline(self):
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
self.card.write_text(self.card.read_text().replace(r0, "0" * 40))
|
||
self.save_card_change()
|
||
self.assertNotEqual(self.precheck().returncode, 0)
|
||
|
||
def test_local_precheck_distinguishes_dirty_local_ready_and_remote_unverified(self):
|
||
"""A registered in-progress task with explicit authorization and scope distinguishes
|
||
SHARED_WORKSPACE_BUSY (dirty), LOCAL_READY (offline, registered, HEAD unchanged),
|
||
and REMOTE_UNVERIFIED (delivery, origin unavailable)."""
|
||
f = self.fixture
|
||
scope = self.local_authorization_scope()
|
||
self.assertEqual(scope["branch"], "feature/test")
|
||
self.assertTrue(scope["card"].is_file())
|
||
self.assertEqual(scope["target_branch"], "main")
|
||
|
||
# (1) Dirty workspace still yields SHARED_WORKSPACE_BUSY under the registered scope
|
||
# even when the authoritative origin is reachable. Clean before (2).
|
||
(self.wt / "tracked").write_text("preserve current edit\n")
|
||
dirty = self.precheck()
|
||
self.assertNotEqual(dirty.returncode, 0, dirty.stdout)
|
||
self.assertIn("[verdict] SHARED_WORKSPACE_BUSY", dirty.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", dirty.stdout)
|
||
f.git(self.wt, "commit", "-am", "Save owner edit")
|
||
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
|
||
|
||
# (2) Same unavailable origin; local mode against a registered in-progress task
|
||
# returns LOCAL_READY and leaves HEAD untouched.
|
||
local = self.precheck(mode="local")
|
||
self.assertEqual(local.returncode, 0, local.stdout)
|
||
self.assertIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertNotIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
# (3) Same unavailable origin; delivery mode continues to refuse with
|
||
# REMOTE_UNVERIFIED because no read has refreshed the authoritative target.
|
||
delivery = self.precheck(mode="delivery")
|
||
self.assertNotEqual(delivery.returncode, 0, delivery.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", delivery.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", delivery.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
def test_local_precheck_rejects_closed_task(self):
|
||
"""A closed registered task (已完成（合入前闭合）) must not advertise LOCAL_READY.
|
||
|
||
The current local precheck emits LOCAL_READY for any registered in-progress
|
||
card text and never inspects the card status, so this test records the
|
||
expected red result: the verdict is wrong and the exit code is wrong."""
|
||
f = self.fixture
|
||
self.card.write_text(self.card.read_text().replace("状态：进行中",
|
||
"状态：已完成（合入前闭合）"))
|
||
self.save_card_change()
|
||
self.assertIn("已完成（合入前闭合）", self.card.read_text())
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
local = self.precheck()
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
def test_local_precheck_rejects_reachable_target_rewind(self):
|
||
"""Origin is reachable but its target was rewound after the task cached a
|
||
newer descendant. Local precheck must not emit LOCAL_READY under that
|
||
divergence — the cached refs/remotes/origin/main is no longer trustworthy.
|
||
|
||
The current implementation lets fetch_target() surface a non-FFREMOTE_UNVERIFIED
|
||
line, then unconditionally prints LOCAL_READY in local mode. The exit code
|
||
is wrong and the verdict is wrong; this test records both expected failures."""
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# (1) Advance the remote and let local precheck refresh the cached origin/main.
|
||
f.advance_remote()
|
||
refresh = self.precheck()
|
||
self.assertEqual(refresh.returncode, 0, refresh.stdout)
|
||
forward = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
self.assertNotEqual(forward, r0)
|
||
# (2) Rewind the bare remote target back to R0; the worktree cache still
|
||
# points at the forward descendant because we have not fetched since.
|
||
f.git(f.remote, "update-ref", "refs/heads/main", r0, forward)
|
||
self.assertEqual(f.git(f.remote, "rev-parse", "refs/heads/main").stdout.strip(), r0)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "origin/main").stdout.strip(), forward)
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
# (3) Local precheck sees a reachable origin whose target is older than the
|
||
# cache. fetch_target() refuses the non-fast-forward update; the precheck
|
||
# must still refuse LOCAL_READY.
|
||
local = self.precheck()
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
# After the failed fetch the cache still points at the forward descendant
|
||
# so a follow-up run can repeat the same check.
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "origin/main").stdout.strip(), forward)
|
||
|
||
def test_local_precheck_rejects_reachable_target_divergence(self):
|
||
"""Origin is reachable but its target was replaced by a divergent commit
|
||
that shares R0 with the cached forward target without descending from it.
|
||
Local precheck must refuse LOCAL_READY and surface the divergence so the
|
||
owner can reconcile the target history instead of trusting the cache.
|
||
|
||
Both the cached forward target and the divergent commit are locally
|
||
available so target_rewound() can compare them via merge-base."""
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# (1) Advance the remote and let local precheck refresh the cached origin/main.
|
||
f.advance_remote()
|
||
refresh = self.precheck()
|
||
self.assertEqual(refresh.returncode, 0, refresh.stdout)
|
||
forward = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
self.assertNotEqual(forward, r0)
|
||
# (2) Build a divergent commit on top of R0 in the main checkout; it shares
|
||
# R0 with the cached forward target but does not descend from it.
|
||
f.git(f.repo, "switch", "--detach", r0)
|
||
(f.repo / "divergent-main").write_text("different target history\n")
|
||
f.git(f.repo, "add", "divergent-main")
|
||
f.git(f.repo, "commit", "-m", "Create divergent fixture history")
|
||
divergent = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
||
# Both targets must share R0 (otherwise they would just be unrelated refs).
|
||
self.assertEqual(f.git(f.repo, "merge-base", forward, divergent).stdout.strip(), r0)
|
||
# The cached forward target must NOT be an ancestor of the divergent commit.
|
||
self.assertEqual(f.git(f.repo, "merge-base", "--is-ancestor", forward,
|
||
divergent, check=False).returncode, 1)
|
||
# Both commits must be locally available so target_rewound() can inspect them.
|
||
self.assertEqual(f.git(self.wt, "cat-file", "-e", forward + "^{commit}",
|
||
check=False).returncode, 0)
|
||
self.assertEqual(f.git(self.wt, "cat-file", "-e", divergent + "^{commit}",
|
||
check=False).returncode, 0)
|
||
# (3) Push the divergent commit so the bare remote's object database sees it,
|
||
# then replace refs/heads/main on the bare remote with the divergent commit.
|
||
f.git(f.repo, "push", "origin", "HEAD:refs/heads/divergent")
|
||
f.git(f.remote, "update-ref", "refs/heads/main", divergent, forward)
|
||
self.assertEqual(f.git(f.remote, "rev-parse", "refs/heads/main").stdout.strip(), divergent)
|
||
# The worktree cache still points at the forward descendant because we have
|
||
# not fetched since.
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "origin/main").stdout.strip(), forward)
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
# (4) Local precheck: fetch_target() refuses the non-fast-forward update to
|
||
# the divergent commit, target_rewound() then surfaces the divergence, and
|
||
# the precheck must not advertise LOCAL_READY.
|
||
local = self.precheck()
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertIn("rewound or diverged", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
# After the failed fetch the cache still points at the forward descendant
|
||
# so a follow-up run can repeat the same check.
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "origin/main").stdout.strip(), forward)
|
||
|
||
def test_unknown_work_mode_is_rejected(self):
|
||
result = self.precheck(mode="anything")
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
def test_local_resume_imports_original_receipt_without_sync(self):
|
||
f = self.fixture
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
f.advance_remote()
|
||
result = self.resume(receipt)
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("[resumed]", result.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(receipt.read_bytes(), original)
|
||
self.assertIn(json.loads(original)["receipt_sha256"], self.card.read_text())
|
||
current = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
self.assertNotEqual(json.loads(original)["after"]["target_head"], current)
|
||
|
||
def assert_history_change_rejected(self, receipt):
|
||
f = self.fixture
|
||
before = self.card.read_bytes()
|
||
original = receipt.read_bytes()
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
result = self.resume(receipt)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("Precheck target history has rewound or diverged", result.stdout)
|
||
self.assertNotIn("[resumed]", result.stdout)
|
||
self.assertEqual(self.card.read_bytes(), before)
|
||
self.assertEqual(receipt.read_bytes(), original)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
|
||
def test_local_resume_rejects_target_rewind_before_import(self):
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
f.advance_remote()
|
||
receipt = self.capture()
|
||
observed = json.loads(receipt.read_text())["after"]["target_head"]
|
||
self.assertNotEqual(observed, r0)
|
||
# Only this test's disposable bare remote is rewritten.
|
||
f.git(f.remote, "update-ref", "refs/heads/main", r0, observed)
|
||
# Reproduce recovery after another client operation refreshed the target.
|
||
f.git(self.wt, "fetch", "origin", "+refs/heads/main:refs/remotes/origin/main")
|
||
self.assert_history_change_rejected(receipt)
|
||
|
||
def test_local_resume_rejects_divergent_target_before_import(self):
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
f.advance_remote()
|
||
receipt = self.capture()
|
||
observed = json.loads(receipt.read_text())["after"]["target_head"]
|
||
f.git(f.repo, "switch", "--detach", r0)
|
||
(f.repo / "divergent-main").write_text("different target history\n")
|
||
f.git(f.repo, "add", "divergent-main")
|
||
f.git(f.repo, "commit", "-m", "Create divergent fixture history")
|
||
divergent = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
||
f.git(f.repo, "push", "origin", "HEAD:refs/heads/divergent")
|
||
self.assertEqual(f.git(f.repo, "merge-base", "--is-ancestor", observed,
|
||
divergent, check=False).returncode, 1)
|
||
# Both targets retain R0; only normal ancestry may permit recovery.
|
||
f.git(f.remote, "update-ref", "refs/heads/main", divergent, observed)
|
||
f.git(self.wt, "fetch", "origin", "+refs/heads/main:refs/remotes/origin/main")
|
||
self.assert_history_change_rejected(receipt)
|
||
|
||
def test_make_resume_exports_the_selected_mode_and_receipt(self):
|
||
f = self.fixture
|
||
receipt = self.capture()
|
||
result = f.run_cmd(self.wt, "make", "-f", str(lifecycle.SOURCE / "Makefile"),
|
||
"task-resume", check=False,
|
||
env=f.env | {"TASK_WORK_MODE": "local", "PRECHECK_RECEIPT": str(receipt)})
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("[resumed]", result.stdout)
|
||
|
||
def test_local_receipt_cannot_start_a_task_or_resume_for_delivery(self):
|
||
f = self.fixture
|
||
receipt = self.capture()
|
||
before = self.card.read_bytes()
|
||
delivery = self.resume(receipt, mode="delivery")
|
||
self.assertNotEqual(delivery.returncode, 0, delivery.stdout)
|
||
self.assertEqual(self.card.read_bytes(), before)
|
||
start = f.task(self.wt, "start", "feature/bypass", check=False,
|
||
env=f.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_WORK_MODE": "local",
|
||
"VALIDATE_RECEIPT_PURPOSE": "resume-local",
|
||
"TRIAGE_RECEIPT": str(f.triage_receipt())})
|
||
self.assertNotEqual(start.returncode, 0, start.stdout)
|
||
self.assertFalse((f.root / "repo-wt-feature-bypass").exists())
|
||
|
||
def test_strict_resume_rejects_target_advance(self):
|
||
f = self.fixture
|
||
receipt = self.capture(mode="delivery")
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
f.advance_remote()
|
||
rejected = self.resume(receipt, mode="delivery")
|
||
self.assertNotEqual(rejected.returncode, 0)
|
||
self.assertNotIn("[resumed]", rejected.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
|
||
def test_strict_resume_imports_current_ready_receipt(self):
|
||
receipt = self.capture(mode="delivery")
|
||
result = self.resume(receipt, mode="delivery")
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
|
||
def test_resume_rejects_changed_source_head(self):
|
||
receipt = self.capture()
|
||
f = self.fixture
|
||
(self.wt / "tracked").write_text("new task revision\n")
|
||
f.git(self.wt, "commit", "-am", "Change source after precheck")
|
||
before = self.card.read_bytes()
|
||
result = self.resume(receipt)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(self.card.read_bytes(), before)
|
||
|
||
def test_resume_rejects_another_worktree_with_same_head(self):
|
||
f = self.fixture
|
||
other = f.root / "other-worktree"
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
other_card = self.wt / ".task/feature-other.md"
|
||
other_card.write_text(
|
||
"# Other task\n任务主记录：fixture；状态：进行中；owner：Lifecycle Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {r0}\n"
|
||
f"- 工作区：{other}（worktree）；分支：feature/other\n")
|
||
f.git(self.wt, "add", ".task/feature-other.md")
|
||
f.git(self.wt, "commit", "-m", "Register second fixture task")
|
||
f.git(f.repo, "worktree", "add", "-b", "feature/other", str(other), "HEAD")
|
||
# The main checkout has an older HEAD; align this owned fixture branch to the task HEAD.
|
||
f.git(other, "merge", "--ff-only", "feature/test")
|
||
receipt = self.capture(mode="delivery")
|
||
self.assertEqual(f.git(other, "rev-parse", "HEAD").stdout,
|
||
f.git(self.wt, "rev-parse", "HEAD").stdout)
|
||
result = self.resume(receipt, mode="delivery", cwd=other)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
|
||
def test_unknown_validation_purpose_is_rejected(self):
|
||
receipt = self.capture(mode="delivery")
|
||
f = self.fixture
|
||
result = f.run_cmd(self.wt, "bash", "scripts/evidence.sh", check=False,
|
||
env=f.env | {"VALIDATE_RECEIPT": str(receipt),
|
||
"VALIDATE_RECEIPT_PURPOSE": "ignore"})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
def test_local_precheck_rejects_reachable_uncomparable_target(self):
|
||
"""Origin advertises a target SHA but the advertised object is
|
||
unavailable locally, so ancestry cannot be compared against the
|
||
cached refs/remotes/origin/main. Local precheck must not advertise
|
||
LOCAL_READY under that unverifiable state — the cached target
|
||
cannot be trusted as authoritative without an ancestry check.
|
||
|
||
A disposable test-only git wrapper fails `fetch`, makes
|
||
`ls-remote origin refs/heads/main` advertise a synthetic SHA, and
|
||
makes `cat-file -e <synthetic>^{commit}` fail; all other git calls
|
||
delegate to the real binary so the registered in-progress task
|
||
context stays observable.
|
||
|
||
The current implementation lets target_rewound() bail out at the
|
||
cat-file -e step (return 1), then unconditionally prints
|
||
LOCAL_READY in local mode because the `&& target_rewound` guard
|
||
short-circuits. The verdict is wrong; this test records the
|
||
expected red result."""
|
||
f = self.fixture
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
cached_before = f.git(self.wt, "rev-parse", "refs/remotes/origin/main").stdout
|
||
|
||
# Find the real git to delegate everything but the three intercepted calls.
|
||
real_git = subprocess.check_output(
|
||
["bash", "-c", "command -v git"], env=f.env, text=True).strip()
|
||
synthetic = "1" * 40
|
||
bin_dir = f.root / "uncomparable-bin"
|
||
bin_dir.mkdir(exist_ok=True)
|
||
wrapper = bin_dir / "git"
|
||
wrapper.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"# Disposable fixture wrapper for the reachable-but-uncomparable\n"
|
||
"# scenario: `fetch` fails, `ls-remote origin refs/heads/main`\n"
|
||
"# advertises a synthetic SHA, `cat-file -e <synthetic>^{commit}`\n"
|
||
"# fails. Anything else delegates to the real git so the\n"
|
||
"# registered task context is observable normally.\n"
|
||
f"REAL_GIT={shlex.quote(real_git)}\n"
|
||
f"SYNTHETIC={shlex.quote(synthetic)}\n"
|
||
"WRAPPER_BIN=$(cd \"$(dirname \"$0\")\" && pwd)\n"
|
||
"stripped=\"\"\n"
|
||
"old_ifs=\"$IFS\"\n"
|
||
"IFS=':'\n"
|
||
"for p in $PATH; do\n"
|
||
" [ \"$p\" = \"$WRAPPER_BIN\" ] && continue\n"
|
||
" if [ -z \"$stripped\" ]; then stripped=\"$p\"; else stripped=\"$stripped:$p\"; fi\n"
|
||
"done\n"
|
||
"IFS=\"$old_ifs\"\n"
|
||
"export PATH=\"$stripped\"\n"
|
||
"case \"$1\" in\n"
|
||
" fetch)\n"
|
||
" echo 'fatal: unable to access origin' >&2\n"
|
||
" exit 128 ;;\n"
|
||
" ls-remote)\n"
|
||
" if [ \"${2:-}\" = origin ] && [ \"${3:-}\" = refs/heads/main ]; then\n"
|
||
" printf '%s\\trefs/heads/main\\n' \"$SYNTHETIC\"\n"
|
||
" exit 0\n"
|
||
" fi\n"
|
||
" exec \"$REAL_GIT\" \"$@\" ;;\n"
|
||
" cat-file)\n"
|
||
" if [ \"${2:-}\" = -e ] && [ \"${3:-}\" = \"${SYNTHETIC}^{commit}\" ]; then\n"
|
||
" echo \"fatal: Not a valid object name ${SYNTHETIC}^{commit}.\" >&2\n"
|
||
" exit 1\n"
|
||
" fi\n"
|
||
" exec \"$REAL_GIT\" \"$@\" ;;\n"
|
||
"esac\n"
|
||
"exec \"$REAL_GIT\" \"$@\"\n")
|
||
wrapper.chmod(0o755)
|
||
wrapper_path = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = f.env | {"PATH": wrapper_path}
|
||
local = f.task(self.wt, "precheck", check=False,
|
||
env=wrapper_env | {"TASK_WORK_MODE": "local"})
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "refs/remotes/origin/main").stdout,
|
||
cached_before)
|
||
|
||
def write_no_cache_wrapper(self, bin_dir, advertised, log_name="git.log"):
|
||
"""Create a disposable git wrapper for the missing-tracking-ref scenarios.
|
||
|
||
Intercepts:
|
||
fetch -> fail (origin unreachable; the wrapper itself exits 128
|
||
and writes the standard "unable to access origin" stderr line
|
||
so the precheck captures the same `fetch_err` it would observe
|
||
against a real unreachable remote)
|
||
ls-remote origin refs/heads/main -> configured ADVERTISED (or
|
||
nothing when ADVERTISED is empty, simulating an origin that
|
||
has no configured target branch)
|
||
branch --merged origin/main -> empty output (the cached
|
||
tracking ref is deleted locally; the real git would otherwise
|
||
fail the pipeline under set -e/pipefail and the precheck would
|
||
silently exit before the verdict is printed)
|
||
|
||
All other git calls delegate to the real binary via the resolved
|
||
REAL_GIT absolute path (and a stripped PATH so the wrapper cannot
|
||
recurse into itself) so the registered task context stays
|
||
observable normally.
|
||
|
||
Returns (bin_dir, wrapper_log_path)."""
|
||
f = self.fixture
|
||
real_git = subprocess.check_output(
|
||
["bash", "-c", "command -v git"], env=f.env, text=True).strip()
|
||
bin_dir.mkdir(exist_ok=True)
|
||
wrapper_log = bin_dir / log_name
|
||
wrapper = bin_dir / "git"
|
||
wrapper.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"# Disposable fixture wrapper for the missing-tracking-ref\n"
|
||
"# scenarios:\n"
|
||
"# fetch -> fail (128)\n"
|
||
"# ls-remote origin refs/heads/main -> $ADVERTISED\n"
|
||
"# branch --merged origin/main -> empty\n"
|
||
"# Anything else delegates to the real binary so the registered\n"
|
||
"# task context (HEAD, branch, R0, worktree, .task/) stays\n"
|
||
"# observable normally.\n"
|
||
f"REAL_GIT={shlex.quote(real_git)}\n"
|
||
f"LOG={shlex.quote(str(wrapper_log))}\n"
|
||
f"ADVERTISED={shlex.quote(advertised)}\n"
|
||
"WRAPPER_BIN=$(cd \"$(dirname \"$0\")\" && pwd)\n"
|
||
"echo \"configured advertised-target=$ADVERTISED\" >> \"$LOG\"\n"
|
||
"stripped=\"\"\n"
|
||
"old_ifs=\"$IFS\"\n"
|
||
"IFS=':'\n"
|
||
"for p in $PATH; do\n"
|
||
" [ \"$p\" = \"$WRAPPER_BIN\" ] && continue\n"
|
||
" if [ -z \"$stripped\" ]; then stripped=\"$p\"; else stripped=\"$stripped:$p\"; fi\n"
|
||
"done\n"
|
||
"IFS=\"$old_ifs\"\n"
|
||
"export PATH=\"$stripped\"\n"
|
||
"case \"$1\" in\n"
|
||
" fetch)\n"
|
||
" echo \"fetch-intercepted\" >> \"$LOG\"\n"
|
||
" echo 'fatal: unable to access origin' >&2\n"
|
||
" exit 128 ;;\n"
|
||
" ls-remote)\n"
|
||
" if [ \"${2:-}\" = origin ] && [ \"${3:-}\" = refs/heads/main ]; then\n"
|
||
" echo \"ls-remote-served advertised-target=$ADVERTISED\" >> \"$LOG\"\n"
|
||
" if [ -n \"$ADVERTISED\" ]; then\n"
|
||
" printf '%s\\trefs/heads/main\\n' \"$ADVERTISED\"\n"
|
||
" fi\n"
|
||
" exit 0\n"
|
||
" fi\n"
|
||
" exec \"$REAL_GIT\" \"$@\" ;;\n"
|
||
" branch)\n"
|
||
" # refs/remotes/origin/main is deleted locally; the real git\n"
|
||
" # `git branch --merged origin/main` would fail and the\n"
|
||
" # precheck pipeline (with pipefail + set -e) would silently\n"
|
||
" # exit. Stub it so the verdict is actually emitted.\n"
|
||
" if [ \"${2:-}\" = --merged ] && [ \"${3:-}\" = origin/main ]; then\n"
|
||
" echo \"branch-merged-stub no-merged-branches\" >> \"$LOG\"\n"
|
||
" exit 0\n"
|
||
" fi\n"
|
||
" exec \"$REAL_GIT\" \"$@\" ;;\n"
|
||
"esac\n"
|
||
"exec \"$REAL_GIT\" \"$@\"\n")
|
||
wrapper.chmod(0o755)
|
||
return bin_dir, wrapper_log
|
||
|
||
def test_local_precheck_without_cached_ref_rejects_uncomparable_remote(self):
|
||
"""Cached refs/remotes/origin/main is absent; origin advertises a
|
||
SHA whose object is unavailable locally. Local precheck must not
|
||
advertise LOCAL_READY because ancestry cannot be compared against
|
||
any locally known target — the cached target cannot be trusted
|
||
as authoritative without an ancestry check, and the advertised
|
||
SHA is not locally reachable so no forward-descendant
|
||
relationship can be established either.
|
||
|
||
The current implementation lets target_history() short-circuit at
|
||
the missing cached-ref step (return "ok") and unconditionally
|
||
print LOCAL_READY in local mode. The verdict is wrong; this test
|
||
records the expected red result."""
|
||
f = self.fixture
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# Detach the registered branch from its (now stale) upstream so
|
||
# no in-process code accidentally queries HEAD@{u} / remote.merge
|
||
# against the deleted cached ref.
|
||
f.git(self.wt, "branch", "--unset-upstream", "feature/test")
|
||
f.git(self.wt, "update-ref", "-d", "refs/remotes/origin/main")
|
||
|
||
synthetic = "1" * 40
|
||
bin_dir, wrapper_log = self.write_no_cache_wrapper(
|
||
f.root / "no-cache-uncomparable-bin", synthetic)
|
||
path_env = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = f.env | {"PATH": path_env}
|
||
|
||
local = f.task(self.wt, "precheck", check=False,
|
||
env=wrapper_env | {"TASK_WORK_MODE": "local"})
|
||
|
||
# Proof: the wrapper was loaded (configured marker) and served
|
||
# the failure case (fetch intercepted). ls-remote is served by
|
||
# the wrapper if and when the post-fix implementation queries it
|
||
# for the missing cached-ref path; the current implementation
|
||
# short-circuits before ls-remote, so the assertion below does
|
||
# not require an ls-remote entry.
|
||
log = wrapper_log.read_text()
|
||
self.assertIn(f"configured advertised-target={synthetic}", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
def test_local_precheck_without_cached_ref_rejects_missing_remote_target(self):
|
||
"""Cached refs/remotes/origin/main is absent; origin returns no
|
||
configured target branch (ls-remote outputs nothing). Local
|
||
precheck must not advertise LOCAL_READY because nothing can be
|
||
verified against the registered task's R0 — the cached target
|
||
is the only handle on what origin claimed, and without it (and
|
||
without an ls-remote response) the precheck has no
|
||
authoritative signal to confirm the target even exists.
|
||
|
||
The current implementation lets target_history() short-circuit at
|
||
the missing cached-ref step (return "ok") and unconditionally
|
||
print LOCAL_READY in local mode. The verdict is wrong; this test
|
||
records the expected red result."""
|
||
f = self.fixture
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
f.git(self.wt, "branch", "--unset-upstream", "feature/test")
|
||
f.git(self.wt, "update-ref", "-d", "refs/remotes/origin/main")
|
||
|
||
bin_dir, wrapper_log = self.write_no_cache_wrapper(
|
||
f.root / "no-cache-missing-target-bin", "")
|
||
path_env = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = f.env | {"PATH": path_env}
|
||
|
||
local = f.task(self.wt, "precheck", check=False,
|
||
env=wrapper_env | {"TASK_WORK_MODE": "local"})
|
||
|
||
log = wrapper_log.read_text()
|
||
# Empty advertised value means the wrapper emits no ls-remote line.
|
||
self.assertIn("configured advertised-target=", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
|
||
self.assertNotEqual(local.returncode, 0, local.stdout)
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
def test_local_precheck_without_cached_ref_allows_comparable_forward_target(self):
|
||
"""Cached refs/remotes/origin/main is absent; origin advertises
|
||
the task's R0 (or a forward descendant that is locally
|
||
available). Local precheck must continue to advertise
|
||
LOCAL_READY — the explicit missing-tracking-ref continuation
|
||
contract — because the registered task context is the same as
|
||
the cached ref would have represented, and the ls-remote
|
||
response can be compared against the registered baseline to
|
||
confirm origin is at or past the task's initial state.
|
||
|
||
The current implementation short-circuits at the missing
|
||
cached-ref step (return "ok"), which coincidentally returns
|
||
LOCAL_READY for any registered task. The verdict is right; this
|
||
test records the expected green result so the post-fix
|
||
implementation preserves the contract when the ls-remote check
|
||
is added against the registered R0."""
|
||
f = self.fixture
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
f.git(self.wt, "branch", "--unset-upstream", "feature/test")
|
||
f.git(self.wt, "update-ref", "-d", "refs/remotes/origin/main")
|
||
|
||
bin_dir, wrapper_log = self.write_no_cache_wrapper(
|
||
f.root / "no-cache-comparable-bin", r0)
|
||
path_env = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = f.env | {"PATH": path_env}
|
||
|
||
local = f.task(self.wt, "precheck", check=False,
|
||
env=wrapper_env | {"TASK_WORK_MODE": "local"})
|
||
|
||
log = wrapper_log.read_text()
|
||
self.assertIn(f"configured advertised-target={r0}", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
|
||
self.assertEqual(local.returncode, 0, local.stdout)
|
||
self.assertIn("[verdict] LOCAL_READY", local.stdout)
|
||
self.assertNotIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|