246 lines
12 KiB
Python
246 lines
12 KiB
Python
"""Local recovery is task-scoped and cannot grant delivery readiness."""
|
||
import json
|
||
import os
|
||
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 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 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)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|