weichen
eb46bbb296
Why: an existing authorized task should not merge a moving target merely to resume local work, while new task startup and delivery remain strict. Evidence: SELF red-green recovery and environment regressions; task, recovery, evidence and local-mode write-guard suites passed through make evidence. Impact: add explicit LOCAL_READY and task-resume, validate worktree/card/baseline and receipt purpose, preserve schema 1 and prevent local receipts from starting or releasing tasks. Refs: U2; policy/merge-queue.md; .task/feature-merge-queue-policy-draft.md
196 lines
9.1 KiB
Python
196 lines
9.1 KiB
Python
"""Local recovery is task-scoped and cannot grant delivery readiness."""
|
||
import json
|
||
import unittest
|
||
|
||
import test_task as lifecycle
|
||
|
||
|
||
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 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_rejects_dirty_state_and_unknown_remote(self):
|
||
f = self.fixture
|
||
(self.wt / "tracked").write_text("preserve current edit\n")
|
||
result = self.precheck()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("SHARED_WORKSPACE_BUSY", result.stdout)
|
||
f.git(self.wt, "commit", "-am", "Save owner edit")
|
||
f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
|
||
result = self.precheck()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("REMOTE_UNVERIFIED", result.stdout)
|
||
|
||
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 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"})
|
||
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)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|