1296 lines
73 KiB
Python
1296 lines
73 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_resume_continues_when_origin_is_unreachable(self):
|
||
"""Standalone: capture a valid TASK_WORK_MODE=local precheck receipt,
|
||
point the disposable worktree's origin URL at a nonexistent path,
|
||
then assert (a) `task resume` in local mode imports the receipt and
|
||
leaves HEAD unchanged and (b) the same receipt under delivery mode
|
||
rejects with REMOTE_UNVERIFIED and does not import.
|
||
|
||
Currently RED on (a): cmd_resume calls fetch_target unconditionally,
|
||
so the unreachable origin aborts the resume before the receipt can
|
||
be imported (set -e trips, returncode != 0, no [resumed], receipt
|
||
bytes unchanged, card untouched). Post-fix the local path must
|
||
skip or survive the unreachable fetch and import the receipt on
|
||
the registered in-progress task context.
|
||
|
||
Delivery must remain strict under the same offline origin — the
|
||
RED-capture only records what (a) requires; (b) stays correct."""
|
||
f = self.fixture
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
receipt_sha = json.loads(original)["receipt_sha256"]
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
# Break the disposable worktree's origin URL so the offline
|
||
# scenario is real (the fetch in cmd_resume loses its target).
|
||
# The bare-remote fixture and the registered HEAD remain
|
||
# identifiable on the worktree fs; only the authoritative URL
|
||
# is changed for the duration of this test.
|
||
f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
|
||
# (1) Local mode must continue authorized work by importing the
|
||
# captured receipt under the registered in-progress task context.
|
||
local = self.resume(receipt, mode="local")
|
||
self.assertEqual(local.returncode, 0,
|
||
f"local resume should import the captured receipt under the "
|
||
f"registered task (RED: fetch_target aborts the resume before "
|
||
f"IMPORT_RECEIPT; returncode={local.returncode}); stdout={local.stdout}")
|
||
self.assertIn("[resumed]", local.stdout)
|
||
self.assertIn("work_mode=local", local.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(receipt.read_bytes(), original)
|
||
self.assertIn(receipt_sha, self.card.read_text())
|
||
# (2) Same receipt under delivery mode must stay strict under
|
||
# the same offline origin: reject with REMOTE_UNVERIFIED, no
|
||
# `[resumed]` line, the card is not updated to claim delivery
|
||
# authorization, and HEAD remains untouched.
|
||
card_before_delivery = self.card.read_bytes()
|
||
delivery = self.resume(receipt, mode="delivery")
|
||
self.assertNotEqual(delivery.returncode, 0, delivery.stdout)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", delivery.stdout)
|
||
self.assertNotIn("[resumed]", delivery.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(self.card.read_bytes(), card_before_delivery)
|
||
|
||
def test_local_child_resume_uses_local_parent_when_origin_is_unreachable(self):
|
||
"""Child: configure the task card as child with
|
||
immediate_target=feature/parent, create that parent branch locally
|
||
at the recorded parent_head_sha, capture a local receipt, make
|
||
origin unreachable, and assert that `task resume` in local mode
|
||
imports the receipt using the authorized local parent reference
|
||
(the local refs/heads/feature/parent tip = parent_head_sha).
|
||
|
||
Currently RED: cmd_resume reaches fetch_target before resolving
|
||
the parent reference, so the unreachable origin aborts the
|
||
resume before the receipt can be imported (returncode != 0, no
|
||
`[resumed]`, receipt bytes unchanged, child card untouched).
|
||
Post-fix the local path must skip or survive the unreachable
|
||
fetch AND let resolve_parent_ref fall back to the authorized
|
||
local parent branch when origin is unqueriable."""
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# (1) Create the parent initiative branch in the main repo so
|
||
# the parent has a deterministic tip commit, push it to the
|
||
# bare remote so the worktree can fetch it once at the start,
|
||
# then create the child branch on top of the parent inside the
|
||
# worktree and switch the worktree to it.
|
||
f.git(f.repo, "checkout", "-b", "feature/parent")
|
||
(f.repo / "parent-tip").write_text("authorized parent initiative tip\n")
|
||
f.git(f.repo, "add", "parent-tip")
|
||
f.git(f.repo, "commit", "-m", "Parent initiative tip commit")
|
||
parent_head = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
||
f.git(f.repo, "push", "origin", "feature/parent")
|
||
# Bring the parent branch into the worktree, branch off the
|
||
# parent at the recorded tip, then check the worktree out on
|
||
# the child branch.
|
||
f.git(self.wt, "fetch", "origin",
|
||
"feature/parent:refs/remotes/origin/feature/parent")
|
||
f.git(self.wt, "branch", "feature/parent-child", parent_head)
|
||
f.git(self.wt, "checkout", "feature/parent-child")
|
||
# (2) Write the child card on the child branch so branch_card
|
||
# finds it (the file is named after the branch slug) and the
|
||
# resume's immediate_target + parent_head_sha fields resolve
|
||
# to feature/parent and the recorded parent tip.
|
||
child_card = self.wt / ".task" / "feature-parent-child.md"
|
||
self.assertFalse(child_card.exists())
|
||
child_card.parent.mkdir(exist_ok=True)
|
||
child_card.write_text(
|
||
"# feature/parent-child\n"
|
||
"任务主记录：fixture；状态：进行中；owner：Lifecycle Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {r0}\n"
|
||
f"- 工作区：{self.wt}（worktree）；分支：feature/parent-child\n"
|
||
"- 分支类型：child；initiative_id：feature-parent；root_target：main；"
|
||
"immediate_target：feature/parent；accepting_requirements：false；"
|
||
"parent_branch：feature/parent；"
|
||
f"parent_head_sha：{parent_head}；"
|
||
"parent_pr：无；parent_source：local（非正式准入）\n"
|
||
)
|
||
f.git(self.wt, "add", str(child_card.relative_to(self.wt)))
|
||
f.git(self.wt, "commit", "-m", "Register child fixture task")
|
||
# (3) Capture a local receipt on the child branch.
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
receipt_sha = json.loads(original)["receipt_sha256"]
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
# (4) Break the origin URL so the offline scenario is real.
|
||
f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
|
||
# (5) Local resume must import the receipt using the
|
||
# authorized local parent reference and leave HEAD unchanged.
|
||
result = self.resume(receipt, mode="local")
|
||
self.assertEqual(result.returncode, 0,
|
||
f"local child resume should import the captured receipt via the "
|
||
f"local parent ref (RED: fetch_target aborts the resume before "
|
||
f"resolve_parent_ref / IMPORT_RECEIPT; returncode={result.returncode}); "
|
||
f"stdout={result.stdout}")
|
||
self.assertIn("[resumed]", result.stdout)
|
||
self.assertIn("work_mode=local", result.stdout)
|
||
self.assertIn(f"observed_target={parent_head}", result.stdout)
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(receipt.read_bytes(), original)
|
||
self.assertIn(receipt_sha, child_card.read_text())
|
||
|
||
def test_local_child_resume_rejects_reachable_missing_parent(self):
|
||
"""Child: configure the task card as child with
|
||
immediate_target=feature/parent, create that parent branch on the
|
||
bare remote AND locally at the recorded parent_head_sha, capture
|
||
a local receipt, then remove ONLY refs/heads/feature/parent from
|
||
the bare remote while keeping origin reachable (the bare remote
|
||
still serves refs/heads/main). `task resume` in local mode must
|
||
refuse with REMOTE_UNVERIFIED instead of silently accepting the
|
||
local parent reference: the registered immediate parent branch
|
||
is absent remotely, so the local-only reference cannot be
|
||
reconciled against the authoritative origin even though it
|
||
exists in the worktree.
|
||
|
||
Currently RED: resolve_parent_ref() does not distinguish
|
||
`ls-remote rc=2` (ref missing on a reachable origin) from the
|
||
unreachable branch. The function falls through the
|
||
`elif [ $rc -ne 2 ]` guard and lets `local_ref` satisfy the
|
||
resolver, so cmd_resume reaches IMPORT_RECEIPT and prints
|
||
`[resumed]` even though origin is fully reachable and just
|
||
missing the registered parent ref. Post-fix the local path must
|
||
print `[verdict] REMOTE_UNVERIFIED — origin is reachable but
|
||
feature/parent is missing; preserve local work, receipt not
|
||
imported` and refuse to import the receipt.
|
||
|
||
The fixture mirrors
|
||
test_local_child_resume_uses_local_parent_when_origin_is_unreachable
|
||
but adds the local refs/heads/feature/parent branch and only
|
||
deletes the bare remote ref — the worktree URL stays on the
|
||
disposable bare remote and main remains fetchable, so the only
|
||
authority gap is the absent remote parent ref."""
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# (1) Create the parent initiative branch in the main repo and
|
||
# push it to the bare remote so the worktree can fetch it once
|
||
# at the start, then create the local refs/heads/feature/parent
|
||
# branch inside the worktree (the post-fix path must refuse
|
||
# even when this local branch exists), branch off the parent
|
||
# tip into feature/parent-child, then switch the worktree to
|
||
# the child branch.
|
||
f.git(f.repo, "checkout", "-b", "feature/parent")
|
||
(f.repo / "parent-tip").write_text("authorized parent initiative tip\n")
|
||
f.git(f.repo, "add", "parent-tip")
|
||
f.git(f.repo, "commit", "-m", "Parent initiative tip commit")
|
||
parent_head = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
||
f.git(f.repo, "push", "origin", "feature/parent")
|
||
f.git(self.wt, "fetch", "origin",
|
||
"feature/parent:refs/remotes/origin/feature/parent")
|
||
# The shared refs/heads namespace means the worktree already
|
||
# sees refs/heads/feature/parent (it was created in f.repo via
|
||
# `checkout -b feature/parent`); the test must reject even
|
||
# when this local branch exists, so leave it intact.
|
||
self.assertEqual(
|
||
f.git(self.wt, "rev-parse", "--verify",
|
||
"refs/heads/feature/parent").stdout.strip(), parent_head)
|
||
f.git(self.wt, "branch", "feature/parent-child", parent_head)
|
||
f.git(self.wt, "checkout", "feature/parent-child")
|
||
# (2) Write the child card on the child branch so branch_card
|
||
# finds it (the file is named after the branch slug) and the
|
||
# resume's immediate_target + parent_head_sha fields resolve
|
||
# to feature/parent and the recorded parent tip.
|
||
child_card = self.wt / ".task" / "feature-parent-child.md"
|
||
self.assertFalse(child_card.exists())
|
||
child_card.parent.mkdir(exist_ok=True)
|
||
child_card.write_text(
|
||
"# feature/parent-child\n"
|
||
"任务主记录：fixture；状态：进行中；owner：Lifecycle Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {r0}\n"
|
||
f"- 工作区：{self.wt}（worktree）；分支：feature/parent-child\n"
|
||
"- 分支类型：child；initiative_id：feature-parent；root_target：main；"
|
||
"immediate_target：feature/parent；accepting_requirements：false；"
|
||
"parent_branch：feature/parent；"
|
||
f"parent_head_sha：{parent_head}；"
|
||
"parent_pr：无；parent_source：local（非正式准入）\n"
|
||
)
|
||
f.git(self.wt, "add", str(child_card.relative_to(self.wt)))
|
||
f.git(self.wt, "commit", "-m", "Register child fixture task")
|
||
# (3) Capture a local receipt on the child branch BEFORE
|
||
# removing the bare remote ref, so the receipt reflects the
|
||
# state where the remote parent ref is still present.
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
receipt_sha = json.loads(original)["receipt_sha256"]
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
child_card_before = child_card.read_bytes()
|
||
# (4) Remove ONLY refs/heads/feature/parent from the bare
|
||
# remote. The worktree's origin URL still points at the
|
||
# disposable bare remote and refs/heads/main stays intact, so
|
||
# the only gap in authority is the absent remote parent ref.
|
||
# The local refs/heads/feature/parent branch remains in the
|
||
# worktree (this is the "even if a local parent branch
|
||
# exists" half of the scenario).
|
||
self.assertEqual(
|
||
f.git(f.remote, "rev-parse", "--verify",
|
||
"refs/heads/feature/parent").stdout.strip(), parent_head)
|
||
f.git(f.remote, "update-ref", "-d", "refs/heads/feature/parent")
|
||
# Confirm the bare remote now refuses to serve the parent ref
|
||
# while still serving main (so origin is reachable, just
|
||
# missing the registered immediate parent ref).
|
||
self.assertNotEqual(
|
||
f.git(f.remote, "show-ref", "--verify", "--hash",
|
||
"refs/heads/feature/parent", check=False).returncode, 0)
|
||
self.assertEqual(
|
||
f.git(f.remote, "rev-parse", "--verify",
|
||
"refs/heads/main").stdout.strip(), r0)
|
||
# (5) Local resume must refuse with REMOTE_UNVERIFIED instead
|
||
# of silently accepting the local parent ref. No [resumed]
|
||
# line, the receipt is not imported (card bytes unchanged and
|
||
# receipt_sha256 not embedded), the receipt file is byte-for-
|
||
# byte unchanged, and HEAD is preserved.
|
||
result = self.resume(receipt, mode="local")
|
||
self.assertNotEqual(result.returncode, 0,
|
||
f"local child resume must reject when origin is reachable "
|
||
f"but the registered parent ref is missing; current code "
|
||
f"falls through resolve_parent_ref's rc=2 case and accepts "
|
||
f"the local parent, returning 0 and printing [resumed]; "
|
||
f"stdout={result.stdout}")
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", result.stdout,
|
||
"the resume must surface REMOTE_UNVERIFIED for the reachable-but-"
|
||
"missing-parent scenario so the owner can reconcile the parent "
|
||
f"initiative instead of silently resuming on an unsynced tip; "
|
||
f"stdout={result.stdout}")
|
||
self.assertNotIn("[resumed]", result.stdout,
|
||
"the resume must NOT import the receipt under a reachable "
|
||
"origin whose registered parent ref is missing — accepting "
|
||
"the local-only parent would advertise delivery readiness "
|
||
f"for an initiative the worktree cannot reconcile; stdout={result.stdout}")
|
||
self.assertEqual(child_card.read_bytes(), child_card_before,
|
||
"the child card must not be updated with receipt_sha256 when "
|
||
"the resume refuses to import the receipt")
|
||
self.assertEqual(receipt.read_bytes(), original,
|
||
"the receipt file must be byte-for-byte unchanged when the "
|
||
"resume refuses to import it")
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head,
|
||
"the child HEAD must not move when the resume refuses to "
|
||
"import the receipt")
|
||
self.assertNotIn(receipt_sha, child_card.read_text(),
|
||
"the child card must not embed the receipt sha256 when "
|
||
"the resume refuses to import the receipt")
|
||
|
||
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",
|
||
auth_fail=False):
|
||
"""Create a disposable git wrapper for the missing-tracking-ref and
|
||
reachable-auth-failure scenarios. The wrapper's intercept set is
|
||
parameterised so a single helper produces both no-cache (default)
|
||
and auth-failure fixtures; each test still owns its own bin_dir so
|
||
no fixtures are shared across tests.
|
||
|
||
When auth_fail=False (default, no-cache scenario):
|
||
fetch -> fail (128); the wrapper
|
||
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 -> serve ADVERTISED (or
|
||
nothing when ADVERTISED is empty, simulating an origin that
|
||
has no configured target branch)
|
||
branch --merged origin/main -> DELEGATED to REAL_GIT.
|
||
The earlier stub masked a production-pipeline bug: with the
|
||
cached refs/remotes/origin/main deleted, the real git call
|
||
fails under set -e/pipefail and the precheck silently exits
|
||
before the verdict is printed. The stub has been removed so
|
||
the tests now exercise the actual missing-ref strict-shell
|
||
path; the post-fix implementation must reach LOCAL_READY (or
|
||
REMOTE_UNVERIFIED) without the helper stub, and the test
|
||
records the expected red result on current code.
|
||
|
||
When auth_fail=True (reachable-but-auth-failure scenario):
|
||
fetch -> fail (128) and write
|
||
"fatal: Authentication failed for origin (HTTP 401)"
|
||
ls-remote origin refs/heads/main -> fail (128) and write
|
||
"fatal: Authentication failed (HTTP 401)"; no value served.
|
||
This simulates a reachable origin that refuses to disclose
|
||
its target head — the production code must NOT promote the
|
||
registered task to LOCAL_READY (or [resumed] for resume) just
|
||
because ls-remote returned an auth error instead of an
|
||
advertised SHA.
|
||
|
||
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"
|
||
if auth_fail:
|
||
header_log = "echo \"configured auth-failure-mode\" >> \"$LOG\"\n"
|
||
fetch_block = (
|
||
" fetch)\n"
|
||
" echo \"fetch-auth-failure\" >> \"$LOG\"\n"
|
||
" echo 'fatal: Authentication failed for origin (HTTP 401)' >&2\n"
|
||
" exit 128 ;;\n"
|
||
)
|
||
ls_remote_block = (
|
||
" ls-remote)\n"
|
||
" if [ \"${2:-}\" = origin ] && [ \"${3:-}\" = refs/heads/main ]; then\n"
|
||
" echo \"ls-remote-auth-failure advertised-target=\" >> \"$LOG\"\n"
|
||
" echo 'fatal: Authentication failed (HTTP 401)' >&2\n"
|
||
" exit 128\n"
|
||
" fi\n"
|
||
" exec \"$REAL_GIT\" \"$@\" ;;\n"
|
||
)
|
||
else:
|
||
header_log = f"echo \"configured advertised-target=$ADVERTISED\" >> \"$LOG\"\n"
|
||
fetch_block = (
|
||
" fetch)\n"
|
||
" echo \"fetch-intercepted\" >> \"$LOG\"\n"
|
||
" echo 'fatal: unable to access origin' >&2\n"
|
||
" exit 128 ;;\n"
|
||
)
|
||
ls_remote_block = (
|
||
" 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"
|
||
)
|
||
wrapper.write_text(
|
||
"#!/usr/bin/env bash\n"
|
||
"# Disposable fixture wrapper for the missing-tracking-ref and\n"
|
||
"# reachable-auth-failure scenarios. See write_no_cache_wrapper\n"
|
||
"# docstring for the intercept set.\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"
|
||
+ header_log +
|
||
"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_block + ls_remote_block +
|
||
"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 actually used: the configured marker
|
||
# names the advertised SHA the test asked the wrapper to serve,
|
||
# `fetch` was rejected (so the unreachable remote is observable
|
||
# to the production pipeline), and `ls-remote origin refs/heads/main`
|
||
# was answered with the synthetic SHA the wrapper was configured
|
||
# to serve. The last assertion is required because reachable
|
||
# advertised cases must verify the wrapper served the expected
|
||
# ls-remote response, not only that the wrapper initialized.
|
||
log = wrapper_log.read_text()
|
||
self.assertIn(f"configured advertised-target={synthetic}", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
self.assertIn(f"ls-remote-served advertised-target={synthetic}", 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 serves no ls-remote line.
|
||
# The configured marker still records the empty advertised value,
|
||
# and the served marker is logged with an empty advertised-target so
|
||
# the test can confirm the wrapper was actually invoked for the
|
||
# ls-remote call (not only that the wrapper binary was loaded).
|
||
self.assertIn("configured advertised-target=", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
self.assertIn("ls-remote-served advertised-target=", 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
|
||
a forward descendant of the task's R0 (the descendant is
|
||
locally available so ancestry can be compared against the
|
||
registered baseline). 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()
|
||
# Advance the bare remote so the advertised target is a real
|
||
# forward descendant of R0 (not R0 itself). The descendant must
|
||
# be locally available so the wrapper can advertise it AND the
|
||
# production `cat-file -e` check inside target_history() succeeds
|
||
# when the descendant is compared against the registered R0.
|
||
f.advance_remote()
|
||
forward = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
self.assertNotEqual(forward, r0)
|
||
self.assertEqual(f.git(f.repo, "merge-base", r0, forward).stdout.strip(), r0)
|
||
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", forward)
|
||
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={forward}", log)
|
||
self.assertIn("fetch-intercepted", log)
|
||
# Reachable advertised cases must verify the wrapper actually
|
||
# served the expected ls-remote response (a forward descendant of
|
||
# the registered R0), not only that the wrapper binary was loaded.
|
||
self.assertIn(f"ls-remote-served advertised-target={forward}", 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)
|
||
|
||
def test_local_child_resume_rejects_missing_parent_sha(self):
|
||
"""Child: configure the task card as child with
|
||
immediate_target=feature/parent and parent_branch=feature/parent,
|
||
create that parent branch locally (no origin fetch), branch off
|
||
it into feature/parent-child, but DELIBERATELY OMIT
|
||
parent_head_sha from the child card so the recorded parent head
|
||
is empty. Make origin unreachable, capture a local receipt,
|
||
then assert `task resume` in local mode refuses to import the
|
||
receipt.
|
||
|
||
The registered task cannot be reconciled against any
|
||
authoritative parent head when parent_head_sha is missing
|
||
AND origin is unreachable: the local branch tip is the only
|
||
handle, but without a recorded parent_head_sha the production
|
||
code cannot verify the local tip is the tip the child branched
|
||
off of. The resume must refuse with REMOTE_UNVERIFIED, must
|
||
not emit `[resumed]`, must not import the receipt (the child
|
||
card stays byte-for-byte unchanged and the receipt_sha256 is
|
||
not embedded), and HEAD must remain untouched.
|
||
|
||
Currently RED on current code: cmd_resume only runs the
|
||
parent_head_sha ancestry check when `[ -n "$recorded" ]` is
|
||
true; an empty parent_head_sha short-circuits the check and
|
||
lets the resume fall through to IMPORT_RECEIPT and the
|
||
`[resumed]` line. This test records the expected red result so
|
||
the post-fix implementation enforces the parent_head_sha
|
||
presence contract under the local-outage fallback path.
|
||
|
||
Preserves test_local_child_resume_uses_local_parent_when_origin_is_unreachable
|
||
(the real-outage-with-valid-parent case) and
|
||
test_local_child_resume_rejects_reachable_missing_parent (the
|
||
reachable-known-missing-parent case)."""
|
||
f = self.fixture
|
||
r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
|
||
# (1) Create the parent initiative branch in the main repo so
|
||
# the parent has a deterministic tip commit, push it to the
|
||
# bare remote so the worktree can fetch it once at the start,
|
||
# then create the child branch on top of the parent inside the
|
||
# worktree and switch the worktree to it. We do NOT push or
|
||
# fetch parent in the worktree because the test explicitly
|
||
# makes origin unreachable before the resume step.
|
||
f.git(f.repo, "checkout", "-b", "feature/parent")
|
||
(f.repo / "parent-tip").write_text("authorized parent initiative tip\n")
|
||
f.git(f.repo, "add", "parent-tip")
|
||
f.git(f.repo, "commit", "-m", "Parent initiative tip commit")
|
||
parent_head = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
||
f.git(f.repo, "push", "origin", "feature/parent")
|
||
# The worktree shares the refs/heads namespace with f.repo,
|
||
# so `checkout -b feature/parent` in f.repo already materialises
|
||
# refs/heads/feature/parent inside the worktree. The test must
|
||
# refuse even when this local branch exists, so leave it intact.
|
||
f.git(self.wt, "branch", "feature/parent-child", parent_head)
|
||
f.git(self.wt, "checkout", "feature/parent-child")
|
||
# (2) Write the child card on the child branch so branch_card
|
||
# finds it (the file is named after the branch slug) and the
|
||
# resume's immediate_target + parent_branch fields resolve to
|
||
# feature/parent. The card DELIBERATELY OMITS parent_head_sha so
|
||
# the recorded parent head is empty.
|
||
child_card = self.wt / ".task" / "feature-parent-child.md"
|
||
self.assertFalse(child_card.exists())
|
||
child_card.parent.mkdir(exist_ok=True)
|
||
child_card.write_text(
|
||
"# feature/parent-child\n"
|
||
"任务主记录：fixture；状态：进行中；owner：Lifecycle Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {r0}\n"
|
||
f"- 工作区：{self.wt}（worktree）；分支：feature/parent-child\n"
|
||
"- 分支类型：child；initiative_id：feature-parent；root_target：main；"
|
||
"immediate_target：feature/parent；accepting_requirements：false；"
|
||
"parent_branch：feature/parent；"
|
||
"parent_pr：无；parent_source：local（非正式准入）\n"
|
||
)
|
||
f.git(self.wt, "add", str(child_card.relative_to(self.wt)))
|
||
f.git(self.wt, "commit", "-m", "Register child fixture task without parent_head_sha")
|
||
# (3) Capture a local receipt on the child branch BEFORE
|
||
# making origin unreachable.
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
receipt_sha = json.loads(original)["receipt_sha256"]
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
child_card_before = child_card.read_bytes()
|
||
# (4) Break the origin URL so the offline scenario is real.
|
||
f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
|
||
# Confirm the local parent branch still exists at the recorded
|
||
# tip (resolve_parent_ref would otherwise fall through to a
|
||
# bare error before the missing parent_head_sha check matters).
|
||
self.assertEqual(
|
||
f.git(self.wt, "rev-parse", "--verify",
|
||
"refs/heads/feature/parent").stdout.strip(), parent_head)
|
||
# (5) Local resume must refuse with REMOTE_UNVERIFIED because
|
||
# the recorded parent_head_sha is empty: the local parent tip
|
||
# alone is not a sufficient handle for authorized continuation
|
||
# when origin is unreachable. No [resumed] line, the receipt is
|
||
# not imported (card bytes unchanged and receipt_sha256 not
|
||
# embedded), the receipt file is byte-for-byte unchanged, and
|
||
# HEAD is preserved.
|
||
result = self.resume(receipt, mode="local")
|
||
self.assertNotEqual(result.returncode, 0,
|
||
f"local child resume must reject when origin is unreachable "
|
||
f"and the recorded parent_head_sha is empty; current code "
|
||
f"short-circuits the empty recorded check and lets IMPORT_RECEIPT "
|
||
f"run, returning 0 and printing [resumed]; "
|
||
f"stdout={result.stdout}")
|
||
self.assertNotIn("[resumed]", result.stdout,
|
||
"the resume must NOT import the receipt when the registered "
|
||
"parent_head_sha is empty and origin is unreachable — the local "
|
||
"parent tip alone is not a sufficient handle for authorized "
|
||
f"continuation under the offline fallback path; stdout={result.stdout}")
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", result.stdout,
|
||
"the resume must surface REMOTE_UNVERIFIED for the unreachable-"
|
||
"origin-and-missing-parent_head_sha scenario so the owner can "
|
||
"backfill the recorded parent head before retrying; "
|
||
f"stdout={result.stdout}")
|
||
self.assertEqual(child_card.read_bytes(), child_card_before,
|
||
"the child card must not be updated with receipt_sha256 when "
|
||
"the resume refuses to import the receipt")
|
||
self.assertEqual(receipt.read_bytes(), original,
|
||
"the receipt file must be byte-for-byte unchanged when the "
|
||
"resume refuses to import it")
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head,
|
||
"the child HEAD must not move when the resume refuses to "
|
||
"import the receipt")
|
||
self.assertNotIn(receipt_sha, child_card.read_text(),
|
||
"the child card must not embed the receipt sha256 when "
|
||
"the resume refuses to import the receipt")
|
||
|
||
def test_local_resume_rejects_reachable_auth_failure(self):
|
||
"""Standalone: capture a valid TASK_WORK_MODE=local precheck
|
||
receipt against the real bare remote (origin is reachable and
|
||
serves refs/heads/main normally), then point the worktree at a
|
||
disposable wrapper that makes `fetch` and
|
||
`ls-remote origin refs/heads/main` both fail with explicit
|
||
"Authentication failed (HTTP 401)" and run `task resume` in
|
||
local mode.
|
||
|
||
The wrapper proves the wrapper was actually used (configured
|
||
marker, fetch-auth-failure, ls-remote-auth-failure). The resume
|
||
must NOT yield the local-continuation positive signal: it
|
||
must refuse with REMOTE_UNVERIFIED, must not emit `[resumed]`,
|
||
must not import the receipt (the standalone card stays byte-
|
||
for-byte unchanged and the receipt_sha256 is not embedded),
|
||
the receipt file must be byte-for-byte unchanged, and HEAD
|
||
must remain untouched.
|
||
|
||
Currently RED on current code: when fetch and ls-remote both
|
||
fail (auth error), `target_history` sees an empty listing and
|
||
short-circuits to "ok", so the unreachable-fallback branch
|
||
falls through to IMPORT_RECEIPT and the `[resumed]` line even
|
||
though the authoritative target head cannot be confirmed.
|
||
This test records the expected red result so the post-fix
|
||
implementation refuses the auth-failure scenario instead of
|
||
silently promoting the registered task to local continuation.
|
||
|
||
The test owns its own disposable git wrapper so the auth
|
||
failure scenario does not leak into the wrapper shared with
|
||
the no-cache tests; the existing
|
||
test_local_resume_continues_when_origin_is_unreachable
|
||
(unreachable origin with the bare remote URL rewritten) and
|
||
test_local_child_resume_uses_local_parent_when_origin_is_unreachable
|
||
(unreachable origin with a child task) are preserved."""
|
||
f = self.fixture
|
||
# (1) Capture a valid local receipt while origin is reachable
|
||
# so the receipt reflects the registered task state under the
|
||
# disposable bare remote. The capture itself runs without the
|
||
# wrapper so the receipt is a real artifact, not a fake.
|
||
receipt = self.capture()
|
||
original = receipt.read_bytes()
|
||
receipt_sha = json.loads(original)["receipt_sha256"]
|
||
head = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
card_before = self.card.read_bytes()
|
||
# (2) Build the disposable auth-failure wrapper so the wrapper
|
||
# intercepts fetch (128, Authentication failed) and
|
||
# ls-remote origin refs/heads/main (128, Authentication failed,
|
||
# no value served) while everything else delegates to the real
|
||
# git. The wrapper's own marker line proves it was loaded.
|
||
bin_dir, wrapper_log = self.write_no_cache_wrapper(
|
||
f.root / "auth-failure-bin", "", log_name="auth-failure.log",
|
||
auth_fail=True)
|
||
path_env = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = (f.env | {"PATH": path_env}
|
||
| {"TASK_WORK_MODE": "local",
|
||
"PRECHECK_RECEIPT": str(receipt)})
|
||
# (3) Local resume under the auth-failure wrapper must refuse.
|
||
result = f.task(self.wt, "resume", check=False, env=wrapper_env)
|
||
# Proof the wrapper was actually used.
|
||
log = wrapper_log.read_text()
|
||
self.assertIn("configured auth-failure-mode", log)
|
||
self.assertIn("fetch-auth-failure", log)
|
||
self.assertIn("ls-remote-auth-failure", log)
|
||
# Auth-failure scenarios must assert the wrapper actually
|
||
# served the expected (auth) response, not only that the
|
||
# wrapper initialized.
|
||
self.assertNotEqual(result.returncode, 0,
|
||
f"local resume must reject when fetch and ls-remote both "
|
||
f"return Authentication failed (HTTP 401); current code "
|
||
f"falls through target_history's empty-listing short-"
|
||
f"circuit and reaches IMPORT_RECEIPT, returning 0 and "
|
||
f"printing [resumed]; stdout={result.stdout}")
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", result.stdout,
|
||
"the resume must surface REMOTE_UNVERIFIED when the reachable "
|
||
"origin returns Authentication failed (HTTP 401) instead of an "
|
||
"advertised target head; the post-fix implementation must not "
|
||
"promote the registered task to local continuation under an "
|
||
f"authoritative-but-unverifiable remote; stdout={result.stdout}")
|
||
self.assertNotIn("[resumed]", result.stdout,
|
||
"the resume must NOT import the receipt when the reachable "
|
||
"origin returns Authentication failed (HTTP 401); accepting "
|
||
"the registered task would advertise delivery readiness for a "
|
||
f"target head that cannot be confirmed against origin; "
|
||
f"stdout={result.stdout}")
|
||
self.assertEqual(self.card.read_bytes(), card_before,
|
||
"the registered task card must not be updated with "
|
||
"receipt_sha256 when the resume refuses to import the receipt")
|
||
self.assertEqual(receipt.read_bytes(), original,
|
||
"the receipt file must be byte-for-byte unchanged when the "
|
||
"resume refuses to import it")
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head,
|
||
"the task HEAD must not move when the resume refuses to "
|
||
"import the receipt")
|
||
self.assertNotIn(receipt_sha, self.card.read_text(),
|
||
"the registered task card must not embed the receipt sha256 "
|
||
"when the resume refuses to import the receipt")
|
||
|
||
def test_local_precheck_rejects_reachable_auth_failure(self):
|
||
"""Standalone: capture a valid TASK_WORK_MODE=local precheck
|
||
receipt against the real bare remote (origin is reachable and
|
||
serves refs/heads/main normally), then point the worktree at a
|
||
disposable wrapper that makes `fetch` and
|
||
`ls-remote origin refs/heads/main` both fail with explicit
|
||
"Authentication failed (HTTP 401)" and run `task precheck` in
|
||
local mode.
|
||
|
||
The wrapper proves the wrapper was actually used (configured
|
||
marker, fetch-auth-failure, ls-remote-auth-failure). The
|
||
precheck must NOT yield the local-continuation positive signal:
|
||
it must refuse with REMOTE_UNVERIFIED, must not advertise
|
||
LOCAL_READY, and HEAD must remain untouched.
|
||
|
||
Currently RED on current code: when fetch and ls-remote both
|
||
fail (auth error), `target_history` sees an empty listing and
|
||
short-circuits to "ok", so the unreachable-fallback branch falls
|
||
through to LOCAL_READY even though the authoritative target
|
||
head cannot be confirmed. This test records the expected red
|
||
result so the post-fix implementation refuses the auth-failure
|
||
scenario instead of silently promoting the registered task to
|
||
local continuation.
|
||
|
||
The test owns its own disposable git wrapper so the auth
|
||
failure scenario does not leak into the wrapper shared with
|
||
the no-cache tests; the existing
|
||
test_local_resume_rejects_reachable_auth_failure (unreachable
|
||
origin under resume) and test_local_child_resume_rejects_missing_parent_sha
|
||
(unreachable origin with a child task missing parent_head_sha)
|
||
are preserved."""
|
||
f = self.fixture
|
||
head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
|
||
# (1) Build the disposable auth-failure wrapper so the wrapper
|
||
# intercepts fetch (128, Authentication failed) and
|
||
# ls-remote origin refs/heads/main (128, Authentication failed,
|
||
# no value served) while everything else delegates to the real
|
||
# git. The wrapper's own marker line proves it was loaded.
|
||
bin_dir, wrapper_log = self.write_no_cache_wrapper(
|
||
f.root / "auth-failure-precheck-bin", "",
|
||
log_name="auth-failure-precheck.log",
|
||
auth_fail=True)
|
||
path_env = f"{bin_dir}{os.pathsep}{f.env.get('PATH', os.environ['PATH'])}"
|
||
wrapper_env = (f.env | {"PATH": path_env}
|
||
| {"TASK_WORK_MODE": "local"})
|
||
# (2) Local precheck under the auth-failure wrapper must refuse.
|
||
local = f.task(self.wt, "precheck", check=False, env=wrapper_env)
|
||
# Proof the wrapper was actually used.
|
||
log = wrapper_log.read_text()
|
||
self.assertIn("configured auth-failure-mode", log)
|
||
self.assertIn("fetch-auth-failure", log)
|
||
self.assertIn("ls-remote-auth-failure", log)
|
||
# Auth-failure scenarios must assert the wrapper actually
|
||
# served the expected (auth) response, not only that the
|
||
# wrapper initialized.
|
||
self.assertNotEqual(local.returncode, 0,
|
||
f"local precheck must reject when fetch and ls-remote both "
|
||
f"return Authentication failed (HTTP 401); current code "
|
||
f"falls through target_history's empty-listing short-"
|
||
f"circuit and reaches LOCAL_READY, returning 0; "
|
||
f"stdout={local.stdout}")
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", local.stdout,
|
||
"the precheck must surface REMOTE_UNVERIFIED when the reachable "
|
||
"origin returns Authentication failed (HTTP 401) instead of an "
|
||
"advertised target head; the post-fix implementation must not "
|
||
"promote the registered task to local continuation under an "
|
||
f"authoritative-but-unverifiable remote; stdout={local.stdout}")
|
||
self.assertNotIn("[verdict] LOCAL_READY", local.stdout,
|
||
"the precheck must NOT advertise LOCAL_READY when the "
|
||
"reachable origin returns Authentication failed (HTTP 401); "
|
||
"accepting the registered task would advertise local "
|
||
"continuation for a target head that cannot be confirmed "
|
||
f"against origin; stdout={local.stdout}")
|
||
self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before,
|
||
"the task HEAD must not move when the precheck refuses to "
|
||
"advertise LOCAL_READY")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|