635 lines
32 KiB
Python
635 lines
32 KiB
Python
"""Exercise the evidence recorder against disposable real Git repositories."""
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
|
||
|
||
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
||
|
||
|
||
# Literal delimiters exercise header, JSON, repr and assignment boundaries.
|
||
AUTHORIZATION_FORMS = (
|
||
'Authorization: {scheme} {secret}',
|
||
'{{"Authorization":"{scheme} {secret}"}}',
|
||
"{{'Authorization': '{scheme} {secret}'}}",
|
||
'Authorization: "{scheme} {secret}"',
|
||
"Authorization: '{scheme} {secret}'",
|
||
'AUTHORIZATION = {scheme} {secret}',
|
||
'"authorization" = "{scheme} {secret}"',
|
||
"'aUtHoRiZaTiOn' = '{scheme} {secret}'",
|
||
'authorization={scheme} {secret}',
|
||
'{{ "Authorization"\t:\t"{scheme}\t{secret}", "ok": true }}',
|
||
)
|
||
|
||
|
||
def authorization_cases():
|
||
for scheme in ("Bearer", "tOkEn", "bAsIc"):
|
||
for index, form in enumerate(AUTHORIZATION_FORMS):
|
||
secret = f"syntheticAuthValue{scheme}{index}"
|
||
yield (form.format(scheme=scheme, secret=secret), secret,
|
||
form.format(scheme=scheme, secret="<redacted>"))
|
||
|
||
|
||
def authorization_command(headers):
|
||
# Use a shell quote that leaves each credential spelling intact in CMD.
|
||
args = [('"' + header + '"') if "'" in header else ("'" + header + "'")
|
||
for header in headers]
|
||
return "printf '%s\\n' " + " ".join(args)
|
||
|
||
|
||
class EvidenceTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.temp = tempfile.TemporaryDirectory()
|
||
self.addCleanup(self.temp.cleanup)
|
||
self.base = Path(self.temp.name)
|
||
self.repo = self.base / "repo"
|
||
self.repo.mkdir()
|
||
self.git("init", "-b", "main")
|
||
self.git("config", "user.name", "Evidence Test")
|
||
self.git("config", "user.email", "evidence@example.invalid")
|
||
(self.repo / "tracked").write_text("initial\n")
|
||
self.git("add", "tracked")
|
||
self.git("commit", "-qm", "Initial")
|
||
self.head = self.git("rev-parse", "HEAD")
|
||
self.git("update-ref", "refs/remotes/origin/main", self.head)
|
||
self.receipt = self.base / "receipt.json"
|
||
self.evidence_dir = self.base / "evidence"
|
||
self.card = self.repo / ".task/main.md"
|
||
self.card.parent.mkdir()
|
||
self.card.write_text("# Test task\n")
|
||
|
||
def git(self, *args, cwd=None):
|
||
return subprocess.check_output(
|
||
["git", *args], cwd=cwd or self.repo, text=True,
|
||
stderr=subprocess.DEVNULL).strip()
|
||
|
||
def run_evidence(self, command=None, cwd=None, **values):
|
||
env = os.environ.copy()
|
||
for key in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT",
|
||
"VALIDATE_RECEIPT_PURPOSE", "TASK_WORK_MODE"):
|
||
env.pop(key, None)
|
||
env["AGENT_EVIDENCE_DIR"] = str(self.evidence_dir)
|
||
if command is not None:
|
||
env["CMD"] = command
|
||
env.update({key: str(value) for key, value in values.items()})
|
||
return subprocess.run(
|
||
["bash", str(SCRIPTS / "evidence.sh")], cwd=cwd or self.repo,
|
||
env=env, capture_output=True, text=True)
|
||
|
||
def capture(self, command="printf 'hello\\n'", **kwargs):
|
||
result = self.run_evidence(command, RECEIPT=self.receipt, **kwargs)
|
||
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
||
self.assertTrue(self.receipt.exists(), "capture must write a receipt")
|
||
return json.loads(self.receipt.read_text())
|
||
|
||
def import_receipt(self, **kwargs):
|
||
return self.run_evidence(IMPORT_RECEIPT=self.receipt, **kwargs)
|
||
|
||
def resign(self, data):
|
||
payload = {key: value for key, value in data.items() if key != "receipt_sha256"}
|
||
data["receipt_sha256"] = hashlib.sha256(json.dumps(
|
||
payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")
|
||
).encode()).hexdigest()
|
||
self.receipt.write_text(json.dumps(data))
|
||
|
||
def test_pipeline_failure_is_recorded_and_returned(self):
|
||
result = self.run_evidence("false | tail -n 1")
|
||
self.assertEqual(result.returncode, 1)
|
||
self.assertIn('"exit_code": 1', self.card.read_text())
|
||
|
||
def test_child_failure_code_is_preserved(self):
|
||
result = self.run_evidence("exit 7")
|
||
self.assertEqual(result.returncode, 7)
|
||
self.assertIn('"exit_code": 7', self.card.read_text())
|
||
|
||
def test_validation_purpose_is_not_injected_into_measured_command(self):
|
||
data = self.capture("printf '%s' \"${VALIDATE_RECEIPT_PURPOSE-unset}\"",
|
||
VALIDATE_RECEIPT_PURPOSE="resume-local")
|
||
self.assertEqual(data["output"], "unset")
|
||
|
||
def test_fixed_candidate_capture_preserves_checkout(self):
|
||
self.git("add", ".task/main.md")
|
||
self.git("commit", "-qm", "Save fixture card before validation")
|
||
head = self.git("rev-parse", "HEAD")
|
||
card = self.card.read_bytes()
|
||
index = self.git("ls-files", "--stage")
|
||
data = self.capture("printf 'fixed candidate check\\n'")
|
||
self.assertEqual(data["before"]["head"], head)
|
||
self.assertEqual(data["after"]["head"], head)
|
||
self.assertEqual(data["output"], "fixed candidate check\n")
|
||
self.assertEqual(self.git("rev-parse", "HEAD"), head)
|
||
self.assertEqual(self.git("ls-files", "--stage"), index)
|
||
self.assertEqual(self.card.read_bytes(), card)
|
||
self.assertEqual(self.git("status", "--porcelain"), "")
|
||
self.assertEqual(self.receipt.stat().st_mode & 0o777, 0o600)
|
||
|
||
def test_capture_without_card_and_import_without_rerun(self):
|
||
self.card.unlink()
|
||
data = self.capture("printf 'once\\n' >> invocation; cat invocation")
|
||
self.assertEqual(data["before"]["head"], self.head)
|
||
self.assertEqual(data["output"], "once\n")
|
||
self.assertEqual(data["source"], "SELF")
|
||
self.assertFalse(self.card.exists())
|
||
self.card.write_text("# Test task\n")
|
||
result = self.import_receipt()
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertEqual((self.repo / "invocation").read_text(), "once\n")
|
||
self.assertIn(data["receipt_sha256"], self.card.read_text())
|
||
|
||
def test_recorder_environment_does_not_leak_into_nested_commands(self):
|
||
data = self.capture("printf '%s,%s,%s' \"${CMD-unset}\" \"${RECEIPT-unset}\" \"${SRC-unset}\"", SRC="SELF")
|
||
self.assertEqual(data["output"], "unset,unset,unset")
|
||
|
||
def test_missing_card_does_not_execute(self):
|
||
self.card.unlink()
|
||
result = self.run_evidence("touch should-not-exist")
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertFalse((self.repo / "should-not-exist").exists())
|
||
|
||
def test_import_missing_card_fails_without_losing_receipt(self):
|
||
self.capture()
|
||
self.card.unlink()
|
||
result = self.import_receipt()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertTrue(self.receipt.exists())
|
||
|
||
def test_linked_worktree_import_preserves_source_identity(self):
|
||
data = self.capture()
|
||
other = self.base / "linked"
|
||
self.git("worktree", "add", "-qb", "feature/linked", str(other))
|
||
card = other / ".task/feature-linked.md"
|
||
card.parent.mkdir()
|
||
card.write_text("# Linked task\n")
|
||
result = self.import_receipt(cwd=other)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn(str(self.repo), card.read_text())
|
||
self.assertIn(data["before"]["head"], card.read_text())
|
||
|
||
def test_unrelated_repo_cannot_import(self):
|
||
self.capture()
|
||
other = self.base / "unrelated"
|
||
subprocess.check_call(["git", "clone", "-q", str(self.repo), str(other)])
|
||
card = other / ".task/main.md"
|
||
card.parent.mkdir()
|
||
card.write_text("# Other task\n")
|
||
result = self.import_receipt(cwd=other)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(card.read_text(), "# Other task\n")
|
||
|
||
def test_all_persisted_and_displayed_text_is_sanitized(self):
|
||
secrets = ["syntheticBearerValue", "syntheticEnvValue", "syntheticUrlPassword",
|
||
"syntheticQueryValue", "syntheticHeaderValue", "syntheticPlainValue"]
|
||
command = "printf '%s\\n' " + " ".join([
|
||
"'Authorization: Bearer syntheticBearerValue'",
|
||
"'GITEA_TOKEN=syntheticEnvValue'",
|
||
"'https://user:syntheticUrlPassword@example.invalid'",
|
||
"'https://example.invalid?access_token=syntheticQueryValue&ok=1'",
|
||
"'Authorization: token syntheticHeaderValue'",
|
||
"'password: syntheticPlainValue'",
|
||
])
|
||
data = self.capture(command)
|
||
imported = self.import_receipt()
|
||
all_text = self.receipt.read_text() + self.card.read_text() + imported.stdout + imported.stderr
|
||
for secret in secrets:
|
||
self.assertNotIn(secret, all_text)
|
||
self.assertIn("<redacted>", data["command"])
|
||
self.assertEqual(data["output_sha256"], hashlib.sha256(data["output"].encode()).hexdigest())
|
||
self.assertEqual(data["command_sha256"], hashlib.sha256(command.encode()).hexdigest())
|
||
|
||
def test_authorization_forms_are_redacted_in_capture_console_and_import(self):
|
||
cases = list(authorization_cases())
|
||
command = authorization_command([header for header, _, _ in cases])
|
||
result = self.run_evidence(command, RECEIPT=self.receipt)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
data = json.loads(self.receipt.read_text())
|
||
imported = self.import_receipt()
|
||
self.assertEqual(imported.returncode, 0, imported.stderr)
|
||
persisted = self.receipt.read_text() + self.card.read_text()
|
||
displayed = result.stdout + result.stderr + imported.stdout + imported.stderr
|
||
for header, secret, expected in cases:
|
||
with self.subTest(header=header):
|
||
self.assertNotIn(secret, persisted + displayed)
|
||
self.assertIn(expected, data["output"].splitlines())
|
||
self.assertEqual(data["command_sha256"], hashlib.sha256(command.encode()).hexdigest())
|
||
self.assertEqual(data["output_sha256"], hashlib.sha256(data["output"].encode()).hexdigest())
|
||
self.assertEqual(data["source"], "SELF")
|
||
self.assertEqual(data["schema"], 1)
|
||
|
||
def test_authorization_forms_are_redacted_on_direct_card_capture(self):
|
||
cases = list(authorization_cases())
|
||
command = authorization_command([header for header, _, _ in cases])
|
||
result = self.run_evidence(command)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
visible = self.card.read_text() + result.stdout + result.stderr
|
||
for header, secret, _ in cases:
|
||
with self.subTest(header=header):
|
||
self.assertNotIn(secret, visible)
|
||
self.assertIn("<redacted>", visible)
|
||
|
||
def test_import_rejects_unredacted_authorization_with_valid_checksums(self):
|
||
original = self.capture()
|
||
for header, _, _ in authorization_cases():
|
||
with self.subTest(header=header):
|
||
self.card.write_text("# Test task\n")
|
||
data = dict(original, output=header)
|
||
data["output_sha256"] = hashlib.sha256(header.encode()).hexdigest()
|
||
self.resign(data)
|
||
result = self.import_receipt()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("unredacted credential syntax", result.stderr)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def test_known_secret_environment_value_is_redacted_without_label(self):
|
||
data = self.capture("printf %s \"$GITEA_TOKEN\"", GITEA_TOKEN="unlabelledSyntheticCredential")
|
||
self.assertNotIn("unlabelledSyntheticCredential", self.receipt.read_text())
|
||
self.assertIn("<redacted>", data["output"])
|
||
|
||
def test_quoted_credentials_with_spaces_are_fully_redacted(self):
|
||
secret = "synthetic phrase with spaces"
|
||
data = self.capture('printf \'password="synthetic phrase with spaces"\\n\'')
|
||
self.assertNotIn(secret, self.receipt.read_text())
|
||
self.assertNotIn("phrase with spaces", data["command"] + data["output"])
|
||
|
||
def test_command_line_token_arguments_are_redacted(self):
|
||
data = self.capture("printf '%s\\n' 'tool --token syntheticArgumentValue'")
|
||
self.assertNotIn("syntheticArgumentValue", self.receipt.read_text())
|
||
self.assertIn("<redacted>", data["command"])
|
||
|
||
def test_source_cannot_be_promoted_by_environment(self):
|
||
for source in ("GATE", "REPLAY"):
|
||
with self.subTest(source=source):
|
||
result = self.run_evidence("touch should-not-exist", SRC=source)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertFalse((self.repo / "should-not-exist").exists())
|
||
|
||
def test_tamper_is_rejected_before_card_append(self):
|
||
data = self.capture()
|
||
data["output"] = "changed output"
|
||
self.receipt.write_text(json.dumps(data))
|
||
result = self.import_receipt()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def test_invalid_field_types_and_forged_source_are_rejected(self):
|
||
original = self.capture()
|
||
for field, value in (("exit_code", True), ("source", "GATE"), ("before", [])):
|
||
with self.subTest(field=field):
|
||
changed = dict(original, **{field: value})
|
||
self.resign(changed)
|
||
self.assertNotEqual(self.import_receipt().returncode, 0)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def test_output_digest_checked_even_if_receipt_digest_recomputed(self):
|
||
data = self.capture()
|
||
data["output_sha256"] = "0" * 64
|
||
self.resign(data)
|
||
self.assertNotEqual(self.import_receipt().returncode, 0)
|
||
|
||
def test_missing_required_snapshot_fields_are_rejected(self):
|
||
original = self.capture()
|
||
for field in ("target_head", "status"):
|
||
with self.subTest(field=field):
|
||
self.card.write_text("# Test task\n")
|
||
data = json.loads(json.dumps(original))
|
||
del data["before"][field]
|
||
self.resign(data)
|
||
result = self.import_receipt()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertNotIn("Traceback", result.stderr)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def test_before_after_heads_do_not_relabel_changed_command(self):
|
||
data = self.capture("printf changed > tracked; git add tracked; git commit -qm Changed")
|
||
self.assertEqual(data["before"]["head"], self.head)
|
||
self.assertEqual(data["after"]["head"], self.git("rev-parse", "HEAD"))
|
||
self.assertNotEqual(data["before"]["head"], data["after"]["head"])
|
||
|
||
def test_dirty_and_untracked_contents_have_object_digests(self):
|
||
(self.repo / "tracked").write_text("modified\n")
|
||
(self.repo / "extra").write_text("untracked\n")
|
||
data = self.capture()
|
||
self.assertIn("tracked", data["before"]["status"])
|
||
self.assertEqual(data["before"]["untracked"]["extra"], hashlib.sha256(b"untracked\n").hexdigest())
|
||
self.assertNotEqual(data["before"]["tracked_diff_sha256"], hashlib.sha256(b"").hexdigest())
|
||
|
||
def test_nested_cwd_records_untracked_files_from_repository_root(self):
|
||
nested = self.repo / "nested"
|
||
nested.mkdir()
|
||
(nested / "extra").write_text("nested source\n")
|
||
data = self.capture(cwd=nested)
|
||
self.assertEqual(data["before"]["untracked"]["nested/extra"],
|
||
hashlib.sha256(b"nested source\n").hexdigest())
|
||
|
||
def test_after_snapshot_failure_keeps_command_exit_and_output(self):
|
||
result = self.run_evidence("mv .git unavailable-git; printf captured; exit 7", RECEIPT=self.receipt)
|
||
self.assertEqual(result.returncode, 7)
|
||
self.assertTrue(self.receipt.exists())
|
||
data = json.loads(self.receipt.read_text())
|
||
self.assertIsNone(data["after"])
|
||
self.assertTrue(data["after_error"])
|
||
self.assertEqual(data["output"], "captured")
|
||
|
||
def test_import_retry_is_idempotent(self):
|
||
self.capture()
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
recorded = self.card.read_text()
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
self.assertEqual(self.card.read_text(), recorded)
|
||
self.assertEqual(recorded.count("```json"), 1)
|
||
|
||
def card_summaries(self):
|
||
blocks = self.card.read_text().split("```json\n")[1:]
|
||
return [json.loads(block.split("\n```", 1)[0]) for block in blocks]
|
||
|
||
def test_card_holds_summary_and_full_receipt_is_stored_outside_repo(self):
|
||
result = self.run_evidence("seq 1 30")
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
text = self.card.read_text()
|
||
self.assertIn("<!-- Local receipt: SELF.", text)
|
||
self.assertNotIn('"submodules"', text)
|
||
[summary] = self.card_summaries()
|
||
self.assertEqual(summary["summary_schema"], 1)
|
||
self.assertEqual(summary["source"], "SELF")
|
||
self.assertEqual(summary["output_tail"], [str(n) for n in range(16, 31)])
|
||
path = Path(summary["receipt_path"])
|
||
self.assertEqual(path, self.evidence_dir.resolve() / (summary["receipt_sha256"] + ".json"))
|
||
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
||
self.assertEqual(self.evidence_dir.stat().st_mode & 0o777, 0o700)
|
||
full = json.loads(path.read_text())
|
||
payload = {key: value for key, value in full.items() if key != "receipt_sha256"}
|
||
self.assertEqual(full["receipt_sha256"], hashlib.sha256(json.dumps(
|
||
payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest())
|
||
self.assertEqual(full["output"], "".join(f"{n}\n" for n in range(1, 31)))
|
||
self.assertEqual((summary["before_head"], summary["after_head"]), (self.head, self.head))
|
||
self.assertEqual(summary["output_sha256"], full["output_sha256"])
|
||
self.assertIn("submodules", full["before"])
|
||
|
||
def test_card_summary_honors_tail(self):
|
||
self.assertEqual(self.run_evidence("seq 1 30", TAIL=2).returncode, 0)
|
||
self.assertEqual(self.card_summaries()[0]["output_tail"], ["29", "30"])
|
||
|
||
def test_imported_receipt_is_stored_verbatim_and_deduplicated(self):
|
||
data = self.capture()
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
[summary] = self.card_summaries()
|
||
self.assertEqual(json.loads(Path(summary["receipt_path"]).read_text()), data)
|
||
self.assertEqual(summary["receipt_sha256"], data["receipt_sha256"])
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
self.assertEqual(len(self.card_summaries()), 1)
|
||
|
||
def test_full_receipt_store_refuses_different_content(self):
|
||
data = self.capture()
|
||
self.evidence_dir.mkdir()
|
||
occupied = self.evidence_dir / (data["receipt_sha256"] + ".json")
|
||
occupied.write_text('{"other": true}\n')
|
||
result = self.import_receipt()
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("different content", result.stderr)
|
||
self.assertEqual(occupied.read_text(), '{"other": true}\n')
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def test_full_output_is_preserved_not_only_tail(self):
|
||
data = self.capture("seq 1 30", TAIL=2)
|
||
self.assertTrue(data["output"].startswith("1\n2\n"))
|
||
self.assertTrue(data["output"].endswith("29\n30\n"))
|
||
|
||
def test_receipt_write_failure_cannot_claim_success(self):
|
||
bad = self.base / "missing" / "receipt.json"
|
||
result = self.run_evidence("printf hello", RECEIPT=bad)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
|
||
def test_failed_child_exit_survives_receipt_write_failure(self):
|
||
bad = self.base / "missing" / "receipt.json"
|
||
result = self.run_evidence("exit 7", RECEIPT=bad)
|
||
self.assertEqual(result.returncode, 7)
|
||
|
||
def test_capture_does_not_overwrite_an_existing_receipt(self):
|
||
self.receipt.write_text("preserve me")
|
||
result = self.run_evidence("touch should-not-exist", RECEIPT=self.receipt)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(self.receipt.read_text(), "preserve me")
|
||
self.assertFalse((self.repo / "should-not-exist").exists())
|
||
|
||
def test_validate_precheck_checks_result_and_current_base(self):
|
||
(self.repo / "Makefile").write_text("precheck:\n\t@echo '[verdict] READY — fixture'\n")
|
||
self.capture("make precheck")
|
||
valid = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
||
self.assertEqual(valid.returncode, 0, valid.stderr)
|
||
self.git("commit", "--allow-empty", "-qm", "Advance base")
|
||
stale = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
||
self.assertNotEqual(stale.returncode, 0)
|
||
|
||
def test_validate_precheck_rejects_successful_arbitrary_command(self):
|
||
self.capture("printf '[verdict] READY\\n'")
|
||
result = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
|
||
def test_child_receipts_record_root_and_immediate_targets(self):
|
||
self.card.write_text("# Test task\n- 分支类型：child；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：feature/alpha；accepting_requirements：false\n")
|
||
data = self.capture()
|
||
self.assertEqual((data["root_target"], data["immediate_target"]), ("main", "feature/alpha"))
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
self.assertIn(data["receipt_sha256"], self.card.read_text())
|
||
checked = subprocess.run(["python3", str(SCRIPTS / "check-evidence.py"), str(self.card)],
|
||
cwd=self.repo, capture_output=True, text=True)
|
||
self.assertNotIn("WARN", checked.stdout + checked.stderr)
|
||
|
||
def archive_card(self, status):
|
||
self.card.unlink()
|
||
archived = self.repo / ".task/archive/feature-done.md"
|
||
archived.parent.mkdir()
|
||
archived.write_text(f"# Done task\n任务主记录：PR #3 fixture；状态：{status}；owner：test\n"
|
||
"- 工作区：fixture（worktree）；分支：main\n"
|
||
"- 分支类型：child；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：feature/alpha；accepting_requirements：false\n")
|
||
return archived
|
||
|
||
def test_evidence_appends_to_pre_merge_closed_archived_card(self):
|
||
archived = self.archive_card("已完成（合入前闭合）")
|
||
result = self.run_evidence("printf 'archived\\n'")
|
||
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
||
text = archived.read_text()
|
||
self.assertIn("## 证据（Evidence）", text)
|
||
self.assertIn('"immediate_target": "feature/alpha"', text)
|
||
|
||
def test_evidence_ignores_archived_card_with_other_status(self):
|
||
archived = self.archive_card("已完成（回填）")
|
||
original = archived.read_text()
|
||
result = self.run_evidence("touch should-not-exist")
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertFalse((self.repo / "should-not-exist").exists())
|
||
self.assertEqual(archived.read_text(), original)
|
||
|
||
def test_receipts_without_target_metadata_default_to_main(self):
|
||
data = self.capture()
|
||
self.assertEqual((data["root_target"], data["immediate_target"]), ("main", "main"))
|
||
|
||
def test_receipt_target_fields_must_come_in_pairs(self):
|
||
original = self.capture()
|
||
partial = dict(original)
|
||
del partial["immediate_target"]
|
||
self.resign(partial)
|
||
self.assertNotEqual(self.import_receipt().returncode, 0)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
def local_task(self, status="进行中"):
|
||
"""Register a clean linked task worktree whose precheck prints the real local task context."""
|
||
wt = self.base.resolve() / "task-wt"
|
||
self.git("worktree", "add", "-qb", "feature/local", str(wt), self.head)
|
||
(wt / ".task").mkdir()
|
||
(wt / ".task/feature-local.md").write_text(
|
||
f"# Local task\n任务主记录：fixture；状态：{status}；owner：Evidence Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {self.head}\n"
|
||
f"- 工作区：{wt}（worktree）；分支：feature/local\n")
|
||
(wt / "Makefile").write_text(
|
||
f"VERDICT ?= LOCAL_READY\nprecheck:\n\t@python3 {SCRIPTS / 'evidence.py'} --local-context\n"
|
||
"\t@echo '[verdict] $(VERDICT) — fixture'\n")
|
||
self.git("add", ".task", "Makefile", cwd=wt)
|
||
self.git("commit", "-qm", "Register local task", cwd=wt)
|
||
return wt
|
||
|
||
def validate(self, wt, purpose="resume-local", **values):
|
||
return self.run_evidence(cwd=wt, VALIDATE_RECEIPT=self.receipt,
|
||
VALIDATE_RECEIPT_PURPOSE=purpose, **values)
|
||
|
||
def advance_target(self):
|
||
head = self.git("commit-tree", "-p", self.head, "-m", "Advance target", self.head + "^{tree}")
|
||
self.git("update-ref", "refs/remotes/origin/main", head)
|
||
|
||
def test_local_continuation_accepts_registered_task_with_known_lag(self):
|
||
wt = self.local_task()
|
||
self.capture("make precheck", cwd=wt)
|
||
result = self.validate(wt)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn("local continuation validated", result.stdout)
|
||
self.assertIn("src=SELF", result.stdout)
|
||
self.advance_target()
|
||
lagging = self.validate(wt)
|
||
self.assertEqual(lagging.returncode, 0, lagging.stderr)
|
||
|
||
def test_local_continuation_accepts_registered_task_without_remote_tracking_ref(self):
|
||
wt = self.local_task()
|
||
self.capture("make precheck", cwd=wt)
|
||
task_head = self.git("rev-parse", "HEAD", cwd=wt)
|
||
self.git("update-ref", "-d", "refs/remotes/origin/main")
|
||
# The R0 commit recorded in the task card must remain reachable locally
|
||
# even after the remote-tracking ref is removed.
|
||
self.git("cat-file", "-e", self.head)
|
||
result = self.validate(wt)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn("local continuation validated", result.stdout)
|
||
self.assertEqual(self.git("rev-parse", "HEAD", cwd=wt), task_head)
|
||
|
||
def test_local_precheck_captures_without_remote_tracking_ref(self):
|
||
wt = self.local_task()
|
||
head_before = self.git("rev-parse", "HEAD", cwd=wt)
|
||
# Remove the remote-tracking ref BEFORE capturing, so the precheck fixture
|
||
# itself must tolerate the missing refs/remotes/origin/main. The existing
|
||
# --local-context rejects this and the capture fails; recording that
|
||
# expected red result here.
|
||
self.git("update-ref", "-d", "refs/remotes/origin/main")
|
||
# The R0 commit recorded on the task card must remain reachable locally
|
||
# even after the remote-tracking ref is removed.
|
||
self.git("cat-file", "-e", self.head)
|
||
result = self.run_evidence("make precheck", cwd=wt, RECEIPT=self.receipt)
|
||
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
||
self.assertTrue(self.receipt.exists(), "capture must write a receipt")
|
||
data = json.loads(self.receipt.read_text())
|
||
self.assertEqual(data["exit_code"], 0, data["output"])
|
||
self.assertIn("[verdict] LOCAL_READY", data["output"])
|
||
states = re.findall(r"^\[task-state\] registered=(.+) initial_base=([0-9a-f]+)$",
|
||
data["output"], re.M)
|
||
self.assertEqual(len(states), 1, data["output"])
|
||
registered, initial_base = states[0]
|
||
self.assertEqual(Path(registered).resolve(), wt.resolve())
|
||
self.assertEqual(initial_base, self.head)
|
||
self.assertEqual(self.git("rev-parse", "HEAD", cwd=wt), head_before)
|
||
|
||
def test_local_ready_never_authorizes_start_delivery_or_promoted_source(self):
|
||
wt = self.local_task()
|
||
self.capture("make precheck", cwd=wt)
|
||
for purpose in ("start", "resume-delivery"):
|
||
with self.subTest(purpose=purpose):
|
||
result = self.validate(wt, purpose)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("Precheck receipt is not READY", result.stderr)
|
||
for source in ("GATE", "REPLAY"):
|
||
with self.subTest(source=source):
|
||
result = self.validate(wt, SRC=source)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("Local evidence is SELF", result.stderr)
|
||
|
||
def test_delivery_resume_still_requires_ready_and_current_target(self):
|
||
wt = self.local_task()
|
||
self.capture("make precheck", cwd=wt, VERDICT="READY")
|
||
local = self.validate(wt)
|
||
self.assertNotEqual(local.returncode, 0)
|
||
self.assertIn("Precheck receipt is not LOCAL_READY", local.stderr)
|
||
current = self.validate(wt, "resume-delivery")
|
||
self.assertEqual(current.returncode, 0, current.stderr)
|
||
self.assertIn("precheck validated", current.stdout)
|
||
self.advance_target()
|
||
stale = self.validate(wt, "resume-delivery")
|
||
self.assertNotEqual(stale.returncode, 0)
|
||
self.assertIn("authoritative target has changed", stale.stderr)
|
||
|
||
def test_local_continuation_rejects_closed_task(self):
|
||
wt = self.local_task(status="已完成（合入前闭合）")
|
||
self.capture("make precheck", cwd=wt)
|
||
result = self.validate(wt)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("active registered task", result.stderr)
|
||
|
||
def test_local_continuation_rejects_dirty_workspace_or_changed_head(self):
|
||
wt = self.local_task()
|
||
self.capture("make precheck", cwd=wt)
|
||
(wt / "extra").write_text("unsaved\n")
|
||
dirty = self.validate(wt)
|
||
self.assertNotEqual(dirty.returncode, 0)
|
||
self.assertIn("Task workspace is not ready", dirty.stderr)
|
||
(wt / "extra").unlink()
|
||
self.git("commit", "--allow-empty", "-qm", "Advance task", cwd=wt)
|
||
moved = self.validate(wt)
|
||
self.assertNotEqual(moved.returncode, 0)
|
||
self.assertIn("source HEAD has changed", moved.stderr)
|
||
|
||
def test_local_continuation_rejects_mismatched_identity_baseline_workspace_or_scope(self):
|
||
wt = self.local_task()
|
||
original = self.capture("make precheck", cwd=wt)
|
||
identity = "different task identity or initial baseline"
|
||
cases = (
|
||
("task identity", identity,
|
||
lambda d: d.update(output=d["output"].replace(str(wt), str(self.base / "other")))),
|
||
("initial baseline", identity,
|
||
lambda d: d.update(output=d["output"].replace(self.head, "0" * 40))),
|
||
("missing identity", identity,
|
||
lambda d: d.update(output="[verdict] LOCAL_READY — fixture\n")),
|
||
("dirty receipt", "workspace was not clean",
|
||
lambda d: d["after"].update(status="?? extra\n", untracked={"extra": None})),
|
||
("tracked diff", "workspace was not clean",
|
||
lambda d: d["before"].update(tracked_diff_sha256="1" * 64)),
|
||
("target scope", "authorization scope",
|
||
lambda d: d.update(immediate_target="feature/other")),
|
||
("missing scope", "authorization scope",
|
||
lambda d: [d.pop(key) for key in ("root_target", "immediate_target")]),
|
||
)
|
||
for name, message, change in cases:
|
||
with self.subTest(case=name):
|
||
data = json.loads(json.dumps(original))
|
||
change(data)
|
||
data["output_sha256"] = hashlib.sha256(data["output"].encode()).hexdigest()
|
||
self.resign(data)
|
||
result = self.validate(wt)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn(message, result.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|