import csv import hashlib import io import json import os import shutil import subprocess import sys import tempfile import time import unittest from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[3] MODULE_ROOT = PROJECT_ROOT / "dev" / "ana-dev" PYTHON = sys.executable GUARD_PATH = PROJECT_ROOT / "dev-doc" / "ana-doc" / "开发方案" / "SHARED-CONTENT-PUBLISHER-PRODUCTION-GUARD-V001.json" LEDGER_HEADER = [ "release_id", "event_seq", "state", "prior_release_id", "prior_release_set_sha256", "candidate_release_set_sha256", "task_id", "case_id", "batch_id", "run_id", "accepted_audit_id", "operator", "process_identity", "event_time", "recovery_or_rollback_receipt", "attempt_id", "attempt_seq", ] def long(path: Path) -> str: text = os.path.abspath(path) if os.name != "nt" or text.startswith("\\\\?\\"): return text if text.startswith("\\\\"): return "\\\\?\\UNC\\" + text[2:] return "\\\\?\\" + text def write(path: Path, data: bytes) -> None: os.makedirs(long(path.parent), exist_ok=True) with open(long(path), "wb") as handle: handle.write(data) def read(path: Path) -> bytes: with open(long(path), "rb") as handle: return handle.read() def ident(path: Path) -> tuple[int, str]: data = read(path) return len(data), hashlib.sha256(data).hexdigest().upper() def sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest().upper() def canonical(value, newline=True) -> bytes: data = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") return data + (b"\n" if newline else b"") def release_set(data: bytes, formal: str) -> str: descriptor = canonical({"bytes": len(data), "formal_relative_path": formal, "sha256": sha(data)}, newline=False) return sha(data + descriptor) def csv_bytes(header, rows) -> bytes: stream = io.StringIO(newline="") writer = csv.writer(stream, lineterminator="\n") writer.writerow(header) writer.writerows(rows) return stream.getvalue().encode("utf-8") def ledger_rows(data: bytes) -> list[dict[str, str]]: reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig"), newline="")) if reader.fieldnames != LEDGER_HEADER: raise AssertionError(reader.fieldnames) return list(reader) def event_chain_digest(rows: list[dict[str, str]]) -> str: payload = { "columns": LEDGER_HEADER, "rows": [[row[column] for column in LEDGER_HEADER] for row in rows], "schema_version": "SHARED_CONTENT_PUBLISHER_ATTEMPT_EVENT_CHAIN_V1", } return sha(canonical(payload, newline=False)) class Fixture: def __init__(self, root: Path, fault="NONE"): self.root = root self.industry = "DEMO-INDUSTRY" self.task = "TASK-DEMO-PUBLISH" self.case = "CASE-DEMO-PUBLISH" self.batch = "BATCH-DEMO-PUBLISH" self.run_id = "RUN-DEMO-PUBLISH" self.attempt = "ATTEMPT-DEMO-000002" self.audit = "AUDIT-DEMO-PUBLISH-PASS" self.handoff = "HANDOFF-DEMO-PUBLISH-REVIEW" self.prior_release = "b" * 64 preimage = "".join(f"{key}={value}\n" for key, value in ( ("industry", self.industry), ("task_id", self.task), ("case_id", self.case), ("batch_id", self.batch), ("run_id", self.run_id), ("canonical_audit_id", self.audit), )) self.release = hashlib.sha256(preimage.encode()).hexdigest() self.case_root = root / "ana-data" / "cases" / "Demo案例" self.result_root = root / "ana-data" / "result" / "Demo案例" self.manifest_root = self.case_root / "manifest" self.release_store = self.case_root / "核心文档" / ".releases" self.target = self.release_store / self.release self.staging = self.release_store / (".staging-" + self.attempt) self.candidate = root / "ana-data" / "tmp" / "Demo案例" / "candidate" self.history = self.case_root / "审计包" / "TASK-DEMO" / "attempts" / "ATTEMPT-DEMO-000001" self.receipt = self.case_root / "审计包" / "TASK-DEMO" / "attempts" / self.attempt self.case_index = self.case_root / "当前成果索引.md" self.result_index = self.result_root / "当前成果索引.md" self.ledger = self.manifest_root / "promotion_ledger.csv" self.lock = self.manifest_root / ".promotion.lock" for directory in (self.case_root, self.result_root, self.manifest_root, self.release_store, self.candidate, self.history): os.makedirs(long(directory), exist_ok=True) self.prior_index_data = b"# Demo current\n\n- release: prior\n- core: [core](core/current.md)\n" self.result_data = b"# Demo result current\n\n- cases: ../../cases/Demo\xe6\xa1\x88\xe4\xbe\x8b/\xe5\xbd\x93\xe5\x89\x8d\xe6\x88\x90\xe6\x9e\x9c\xe7\xb4\xa2\xe5\xbc\x95.md\n" self.prior_core_data = b"# Prior core\n\nprior-evidence\n" write(self.case_index, self.prior_index_data) write(self.result_index, self.result_data) write(self.case_root / "core" / "current.md", self.prior_core_data) prior_rows_data = [ ("CASES-CURRENT", self.rel(self.case_index), "CASES_CURRENT_INDEX", *ident(self.case_index)), ("CORE-PRIOR", self.rel(self.case_root / "core" / "current.md"), "CORE_DOCUMENT", *ident(self.case_root / "core" / "current.md")), ("RESULT-CURRENT", self.rel(self.result_index), "RESULT_THIN_CURRENT_INDEX", *ident(self.result_index)), ] prior_manifest = csv_bytes( ["member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"], prior_rows_data, ) self.prior_manifest_path = self.manifest_root / "current_output_manifest.csv" write(self.prior_manifest_path, prior_manifest) self.prior_manifest_formal = self.rel(self.prior_manifest_path) self.prior_release_set = release_set(prior_manifest, self.prior_manifest_formal) self.prior_rows = [ {"member_id": row[0], "formal_relative_path": row[1], "artifact_type": row[2], "bytes": row[3], "sha256": row[4]} for row in prior_rows_data ] state = canonical({"semantic_state": "RECOVERY_REQUIRED", "status": "BLOCKED", "exit_code": 30}) self.state_rel = "rollback_current_validation.json" write(self.history / self.state_rel, state) self.history_rows = [self.history_row(self.state_rel)] for index in range(50): rel = f"short/item-{index:02d}.json" write(self.history / rel, canonical({"index": index})) self.history_rows.append(self.history_row(rel)) for index in range(12): parts = [f"long-{index:02d}"] + [(f"segment-{level:02d}-" + "x" * 22) for level in range(7)] + ["receipt.json"] rel = "/".join(parts) write(self.history / rel, canonical({"long": index})) self.history_rows.append(self.history_row(rel)) assert len(self.history_rows) == 63 assert sum(len(os.path.abspath(self.history / row["relative_path"])) >= 260 for row in self.history_rows) >= 12 self.target_rel = self.rel(self.target) self.core_source_rel = "release/core/research.md" self.current_manifest_source_rel = "release/manifest/current_output_manifest.csv" self.index_source_rel = "entry/当前成果索引.md" candidate_core = b"# Candidate core\n\nsource-evidence-closed\n" candidate_index = (f"# Demo current\n\n- release: {self.release}\n- core: [core](核心文档/.releases/{self.release}/core/research.md)\n- evidence: source-evidence-closed\n").encode("utf-8") write(self.candidate / self.core_source_rel, candidate_core) write(self.candidate / self.index_source_rel, candidate_index) core_id = ident(self.candidate / self.core_source_rel) index_id = ident(self.candidate / self.index_source_rel) result_id = ident(self.result_index) self.current_manifest_rows = [ ("CORE-CANDIDATE", f"{self.target_rel}/core/research.md", "CORE_DOCUMENT", *core_id), ("CASES-CURRENT", self.rel(self.case_index), "CASES_CURRENT_INDEX", *index_id), ("RESULT-CURRENT", self.rel(self.result_index), "RESULT_THIN_CURRENT_INDEX", *result_id), ] current_manifest = csv_bytes( ["member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"], self.current_manifest_rows, ) write(self.candidate / self.current_manifest_source_rel, current_manifest) current_manifest_id = ident(self.candidate / self.current_manifest_source_rel) self.current_manifest_formal = f"{self.target_rel}/manifest/current_output_manifest.csv" self.candidate_rows = [ self.artifact_row("CORE-CANDIDATE", self.core_source_rel, f"{self.target_rel}/core/research.md", "CORE_DOCUMENT", "RELEASE_MEMBER"), self.artifact_row("CURRENT-MANIFEST", self.current_manifest_source_rel, self.current_manifest_formal, "CURRENT_OUTPUT_MANIFEST", "RELEASE_MEMBER"), self.artifact_row("CASES-CURRENT", self.index_source_rel, self.rel(self.case_index), "CASES_CURRENT_INDEX", "CASE_CURRENT_INDEX"), ] candidate_manifest = csv_bytes( ["member_id", "candidate_relative_path", "formal_relative_path", "artifact_type", "bytes", "sha256"], [(row["member_id"], "candidate/" + row["relative_path"], row["formal_relative_path"], row["artifact_type"], row["bytes"], row["sha256"]) for row in self.candidate_rows], ) self.candidate_manifest_rel = "candidate_output_manifest.csv" write(self.candidate / self.candidate_manifest_rel, candidate_manifest) candidate_manifest_id = ident(self.candidate / self.candidate_manifest_rel) self.candidate_set = release_set(current_manifest, self.current_manifest_formal) ledger = csv_bytes(LEDGER_HEADER, [[ "old-release", "1", "ROLLED_BACK", self.prior_release, self.prior_release_set, "C" * 64, self.task, self.case, self.batch, self.run_id, "AUDIT-OLD", "tester", "PID-OLD-INSTANCE", "2026-01-01T00:00:00Z", "receipts/old.json", "ATTEMPT-DEMO-000001", "1", ]]) write(self.ledger, ledger) state_id = ident(self.history / self.state_rel) prior_manifest_id = ident(self.prior_manifest_path) self.config = { "schema_version": "SHARED_CONTENT_PUBLISHER_CONFIG_V2", "identity": { "industry": self.industry, "task_id": self.task, "case_id": self.case, "batch_id": self.batch, "run_id": self.run_id, "attempt_id": self.attempt, "canonical_audit_id": self.audit, "review_handoff_id": self.handoff, "release_id": self.release, "prior_release_id": self.prior_release, "candidate_release_set_sha256": self.candidate_set, "operator": "test.operator", }, "roots": { "operation_root": str(self.root), "resolved_root": str(self.root), "volume_identity": (self.root.drive or str(os.stat(self.root).st_dev)).upper(), "candidate_root": self.rel(self.candidate), "release_store_root": self.rel(self.release_store), "case_current_index_path": self.rel(self.case_index), "result_current_index_path": self.rel(self.result_index), "ledger_path": self.rel(self.ledger), "lock_path": self.rel(self.lock), "attempt_receipt_root": self.rel(self.receipt), "history_root": self.rel(self.history), "current_state_relative_path": self.state_rel, }, "expected": { "ledger": {"bytes": len(ledger), "sha256": sha(ledger), "next_event_seq": 2, "next_attempt_seq": 2, "prior_attempt_id": "ATTEMPT-DEMO-000001", "prior_terminal_state": "ROLLED_BACK"}, "candidate": { "manifest_relative_path": self.candidate_manifest_rel, "manifest_bytes": candidate_manifest_id[0], "manifest_sha256": candidate_manifest_id[1], "current_manifest_relative_path": self.current_manifest_source_rel, "current_manifest_bytes": current_manifest_id[0], "current_manifest_sha256": current_manifest_id[1], "manifest_self_formal_relative_path": self.current_manifest_formal, "rows": self.candidate_rows, "link_checks": [{"relative_path": self.index_source_rel, "required_utf8_substrings": ["核心文档/.releases/", "source-evidence-closed"]}], "evidence_checks": [{"relative_path": self.core_source_rel, "required_utf8_substrings": ["source-evidence-closed"]}], }, "prior": { "release_id": self.prior_release, "manifest_formal_relative_path": self.prior_manifest_formal, "manifest_bytes": prior_manifest_id[0], "manifest_sha256": prior_manifest_id[1], "manifest_self_formal_relative_path": self.prior_manifest_formal, "release_set_sha256": self.prior_release_set, "rows": self.prior_rows, }, "history": {"rows": self.history_rows, "minimum_long_paths": 12, "long_path_threshold": 260}, "case_current_index": {"bytes": len(self.prior_index_data), "sha256": sha(self.prior_index_data), "required_utf8_substrings": ["release: prior", "core/current.md"]}, "result_current_index": { "member_id": "RESULT-CURRENT", "formal_relative_path": self.rel(self.result_index), "artifact_type": "RESULT_THIN_CURRENT_INDEX", "bytes": result_id[0], "sha256": result_id[1], "required_utf8_substrings": ["../../cases/Demo案例/当前成果索引.md"], }, "current_state": { "bytes": state_id[0], "sha256": state_id[1], "semantic_field": "semantic_state", "semantic_value": "RECOVERY_REQUIRED", "status_field": "status", "status_value": "BLOCKED", "exit_code_field": "exit_code", "exit_code_value": 30, }, }, "commit": {"strategy": "MARKDOWN_CURRENT_INDEX_REPLACE_V1", "target_release_relative_path": self.target_rel, "staging_relative_path": self.rel(self.staging)}, "test_control": {"environment": "ISOLATED_TEST", "fault": fault, "race_marker_relative_path": ""}, } self.config_path = self.root / "batch-config.json" self.save() def rel(self, path: Path) -> str: return os.path.relpath(path, self.root).replace("\\", "/") def history_row(self, rel: str): size, digest = ident(self.history / rel) return {"relative_path": rel, "bytes": size, "sha256": digest} def artifact_row(self, member, source, formal, artifact_type, role): size, digest = ident(self.candidate / source) return {"member_id": member, "relative_path": source, "formal_relative_path": formal, "artifact_type": artifact_type, "commit_role": role, "bytes": size, "sha256": digest} def save(self): write(self.config_path, canonical(self.config)) def run(self, validate=False, config_path=None): path = config_path or self.config_path env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" command = [PYTHON, "-m", "shared_content_publisher", "--config", str(path)] if validate: command.append("--validate-only") proc = subprocess.run(command, cwd=self.root, env=env, text=True, encoding="utf-8", capture_output=True, timeout=40) return proc, json.loads(proc.stdout.strip().splitlines()[-1]) def popen(self): env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" return subprocess.Popen([PYTHON, "-m", "shared_content_publisher", "--config", str(self.config_path)], cwd=self.root, env=env, text=True, encoding="utf-8", stdout=subprocess.PIPE, stderr=subprocess.PIPE) def formal_state(self): return { "ledger": ident(self.ledger), "case": ident(self.case_index), "result": ident(self.result_index), "target": self.target.exists(), "receipt": self.receipt.exists(), "lock": self.lock.exists(), } def rebuild_candidate_manifests(self, current_rows=None): rows = current_rows if current_rows is not None else self.current_manifest_rows current_data = csv_bytes(["member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"], rows) write(self.candidate / self.current_manifest_source_rel, current_data) for row in self.candidate_rows: if row["relative_path"] == self.current_manifest_source_rel: row["bytes"], row["sha256"] = ident(self.candidate / self.current_manifest_source_rel) candidate_data = csv_bytes( ["member_id", "candidate_relative_path", "formal_relative_path", "artifact_type", "bytes", "sha256"], [(row["member_id"], "candidate/" + row["relative_path"], row["formal_relative_path"], row["artifact_type"], row["bytes"], row["sha256"]) for row in self.candidate_rows], ) write(self.candidate / self.candidate_manifest_rel, candidate_data) current_id = ident(self.candidate / self.current_manifest_source_rel) manifest_id = ident(self.candidate / self.candidate_manifest_rel) section = self.config["expected"]["candidate"] section["rows"] = self.candidate_rows section["current_manifest_bytes"], section["current_manifest_sha256"] = current_id section["manifest_bytes"], section["manifest_sha256"] = manifest_id self.config["identity"]["candidate_release_set_sha256"] = release_set(current_data, self.current_manifest_formal) self.save() class SharedPublisherTests(unittest.TestCase): def fixture(self, fault="NONE"): root = Path(tempfile.mkdtemp(prefix="shared-publisher-v2-")) self.addCleanup(lambda: shutil.rmtree(long(root), ignore_errors=True)) return Fixture(root, fault=fault) def assert_prewrite_failure(self, fx, code): before = fx.formal_state() proc, result = fx.run() self.assertNotEqual(proc.returncode, 0, proc.stdout + proc.stderr) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertEqual(result["error_code"], code) self.assertEqual(fx.formal_state(), before) def test_00_production_guard_precedes_isolated_mutation(self): guard = json.loads(read(GUARD_PATH).decode("utf-8")) self.assertEqual(guard["schema_version"], "SHARED_CONTENT_PUBLISHER_PRODUCTION_GUARD_V1") committed_attempt = ( PROJECT_ROOT / "ana-data" / "cases" / "农业案例" / "审计包" / "ANA-AGRICULTURE-PESTICIDE-FERTILIZER-20260818-001" / "BATCH-004" / "RUN-ANA-AGRICULTURE-PESTICIDE-FERTILIZER-20260818-001-BATCH-004-001" / "attempts" / "ATTEMPT-B004-000003" ) prior_files = committed_attempt / "prior_snapshot" / "files" for row in guard["files"]: live = PROJECT_ROOT / row["relative_path"] snapshot = prior_files / row["relative_path"] if os.path.isfile(long(snapshot)): self.assertEqual(ident(snapshot), (row["bytes"], row["sha256"])) elif row["relative_path"].endswith("/promotion_ledger.csv"): data = read(live)[:row["bytes"]] self.assertEqual((len(data), sha(data)), (row["bytes"], row["sha256"])) else: self.assertEqual(ident(live), (row["bytes"], row["sha256"])) for rel in guard["absent_paths"]: path = PROJECT_ROOT / rel if path == committed_attempt: terminal = json.loads(read(path / "terminal.json").decode("utf-8")) self.assertEqual((terminal["status"], terminal["exit_code"]), ("COMMITTED", 0)) else: self.assertFalse(path.exists(), rel) def test_01_markdown_commit_and_strict_idempotent_replay(self): fx = self.fixture() check, valid = fx.run(validate=True) self.assertEqual(check.returncode, 0, check.stdout + check.stderr) self.assertEqual(valid["lifecycle"], "FRESH") proc, result = fx.run() self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) self.assertEqual(result["status"], "COMMITTED") committed_rows = [row for row in ledger_rows(read(fx.ledger)) if row["attempt_id"] == fx.attempt] self.assertEqual(result["attempt_event_chain_row_count"], 4) self.assertEqual(result["attempt_event_chain_sha256"], event_chain_digest(committed_rows)) self.assertEqual(result["event_time_start_utc"], committed_rows[0]["event_time"]) self.assertEqual(result["event_time_end_utc"], committed_rows[-1]["event_time"]) self.assertEqual(result["event_process_identities"], [result["process_instance_identity"]] * 4) self.assertEqual(read(fx.case_index), read(fx.candidate / fx.index_source_rel)) self.assertEqual(ident(fx.result_index), (result["result_current_bytes"], result["result_current_sha256"])) ledger_after = read(fx.ledger) again, replay = fx.run() self.assertEqual(again.returncode, 0, again.stdout + again.stderr) self.assertEqual(replay["status"], "IDEMPOTENT_COMMITTED") self.assertEqual(read(fx.ledger), ledger_after) self.assertFalse(fx.lock.exists()) def test_02_postcommit_failure_rolls_back_markdown_once(self): fx = self.fixture(fault="POSTCOMMIT_READBACK_FAIL") proc, result = fx.run() self.assertEqual(proc.returncode, 20, proc.stdout + proc.stderr) self.assertEqual(result["status"], "ROLLED_BACK") self.assertEqual(read(fx.case_index), fx.prior_index_data) rollback_rows = [row for row in ledger_rows(read(fx.ledger)) if row["attempt_id"] == fx.attempt] self.assertEqual(result["attempt_event_chain_row_count"], 5) self.assertEqual(result["attempt_event_chain_sha256"], event_chain_digest(rollback_rows)) self.assertEqual(result["event_process_identities"], [result["process_instance_identity"]] * 5) ledger_after = read(fx.ledger) again, blocked = fx.run() self.assertEqual(again.returncode, 12, again.stdout + again.stderr) self.assertEqual(blocked["error_code"], "TERMINAL_REPEAT_BLOCKED") self.assertEqual(read(fx.ledger), ledger_after) def test_03_locked_ledger_race_is_detected_before_append(self): fx = self.fixture(fault="PAUSE_AFTER_LOCK") marker = fx.root / "race.ready" fx.config["test_control"]["race_marker_relative_path"] = fx.rel(marker) fx.save() process = fx.popen() deadline = time.monotonic() + 10 while time.monotonic() < deadline and not fx.lock.exists(): time.sleep(0.01) self.assertTrue(fx.lock.exists()) race = csv_bytes(LEDGER_HEADER, [[ "race", "2", "PREPARED", fx.prior_release, fx.prior_release_set, "D" * 64, fx.task, fx.case, fx.batch, fx.run_id, "AUDIT-RACE", "racer", "PID-RACE-INSTANCE", "2026-01-01T00:00:01Z", "race.json", "ATTEMPT-RACE", "2", ]]) with open(long(fx.ledger), "ab") as handle: handle.write(race.split(b"\n", 1)[1]) write(marker, b"ready\n") stdout, stderr = process.communicate(timeout=20) result = json.loads(stdout.strip().splitlines()[-1]) self.assertEqual(process.returncode, 12, stdout + stderr) self.assertEqual(result["error_code"], "LEDGER_IDENTITY_LOCKED") self.assertEqual(read(fx.case_index), fx.prior_index_data) self.assertFalse(fx.target.exists()) self.assertFalse(fx.receipt.exists()) self.assertFalse(fx.lock.exists()) def test_04_duplicate_sequence_and_declared_sequence_drift(self): fx = self.fixture() base = read(fx.ledger) duplicate = csv_bytes(LEDGER_HEADER, [[ "dup", "1", "ROLLED_BACK", fx.prior_release, fx.prior_release_set, "E" * 64, fx.task, fx.case, fx.batch, fx.run_id, "AUDIT-DUP", "tester", "PID-DUP", "2026-01-01T00:00:02Z", "dup.json", "ATTEMPT-DUP", "2", ]]).split(b"\n", 1)[1] write(fx.ledger, base + duplicate) fx.config["expected"]["ledger"]["bytes"] = len(base + duplicate) fx.config["expected"]["ledger"]["sha256"] = sha(base + duplicate) fx.save() self.assert_prewrite_failure(fx, "LEDGER_SEQUENCE_UNIQUE") fx2 = self.fixture() fx2.config["expected"]["ledger"]["next_event_seq"] = 99 fx2.save() self.assert_prewrite_failure(fx2, "LEDGER_NEXT_EVENT") def test_05_path_overlap_and_junction_ancestor_fail_closed(self): fx = self.fixture() fx.config["roots"]["attempt_receipt_root"] = fx.target_rel fx.save() self.assert_prewrite_failure(fx, "PATH_OVERLAP") fx2 = self.fixture() alias = fx2.root.parent / (fx2.root.name + "-junction") created = False if os.name == "nt": made = subprocess.run(["cmd", "/c", "mklink", "/J", str(alias), str(fx2.root)], capture_output=True) created = made.returncode == 0 else: os.symlink(fx2.root, alias, target_is_directory=True) created = True if not created: self.skipTest("junction creation unavailable") self.addCleanup(lambda: os.rmdir(long(alias)) if os.path.lexists(long(alias)) else None) fx2.config["roots"]["operation_root"] = str(alias) fx2.config["roots"]["resolved_root"] = str(alias) alias_config = alias / fx2.config_path.name fx2.save() proc, result = fx2.run(config_path=alias_config) self.assertEqual(proc.returncode, 12, proc.stdout + proc.stderr) self.assertIn(result["error_code"], ("PATH_REPARSE", "PATH_ALIAS")) def test_06_manifest_omission_and_empty_checks_are_rejected(self): fx = self.fixture() fx.rebuild_candidate_manifests([row for row in fx.current_manifest_rows if row[0] != "RESULT-CURRENT"]) self.assert_prewrite_failure(fx, "CURRENT_MANIFEST_COVERAGE") fx2 = self.fixture() fx2.config["expected"]["candidate"]["link_checks"] = [] fx2.save() self.assert_prewrite_failure(fx2, "CONFIG_TYPE") fx3 = self.fixture() fx3.config["expected"]["candidate"]["evidence_checks"] = [] fx3.save() self.assert_prewrite_failure(fx3, "CONFIG_TYPE") def test_07_extra_target_file_forces_rollback(self): fx = self.fixture(fault="EXTRA_TARGET_AFTER_RENAME") proc, result = fx.run() self.assertEqual(proc.returncode, 20, proc.stdout + proc.stderr) self.assertEqual(result["status"], "ROLLED_BACK") self.assertEqual(read(fx.case_index), fx.prior_index_data) self.assertTrue((fx.target / "UNDECLARED.txt").exists()) def test_08_replay_target_attack_rolls_back_before_false_idempotence(self): fx = self.fixture() proc, _ = fx.run() self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) write(fx.target / "UNDECLARED.txt", b"attack\n") again, result = fx.run() self.assertEqual(again.returncode, 20, again.stdout + again.stderr) self.assertEqual(result["status"], "ROLLED_BACK") self.assertEqual(read(fx.case_index), fx.prior_index_data) self.assertFalse(fx.lock.exists()) def test_09_invalid_config_cannot_use_replay_bypass(self): fx = self.fixture() proc, _ = fx.run() self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) fx.config["identity"]["unexpected"] = "attack" fx.save() again, result = fx.run() self.assertEqual(again.returncode, 12, again.stdout + again.stderr) self.assertEqual(result["error_code"], "CONFIG_KEYS") self.assertNotEqual(result["status"], "IDEMPOTENT_COMMITTED") def test_10_terminal_and_event_chain_tamper_force_rollback(self): fx = self.fixture() proc, _ = fx.run() self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) terminal_path = fx.receipt / "terminal.json" terminal = json.loads(read(terminal_path).decode("utf-8")) terminal["volume_identity"] = "DRIFT" write(terminal_path, canonical(terminal)) again, result = fx.run() self.assertEqual(again.returncode, 20, again.stdout + again.stderr) self.assertEqual(result["status"], "ROLLED_BACK") self.assertEqual(read(fx.case_index), fx.prior_index_data) self.assertTrue((fx.receipt / "terminal.json").exists()) self.assertTrue((fx.receipt / "recovery_terminal.json").exists()) recovery_rows = [row for row in ledger_rows(read(fx.ledger)) if row["attempt_id"] == fx.attempt] self.assertEqual(result["attempt_event_chain_row_count"], 6) self.assertEqual(result["attempt_event_chain_sha256"], event_chain_digest(recovery_rows)) self.assertEqual(result["event_process_identities"], [row["process_identity"] for row in recovery_rows]) fx2 = self.fixture() proc2, _ = fx2.run() self.assertEqual(proc2.returncode, 0, proc2.stdout + proc2.stderr) write(fx2.receipt / "terminal.json", b"{invalid terminal\n") again2, result2 = fx2.run() self.assertEqual(again2.returncode, 20, again2.stdout + again2.stderr) self.assertEqual(result2["status"], "ROLLED_BACK") self.assertEqual(read(fx2.case_index), fx2.prior_index_data) def test_11_prior_release_set_and_result_thin_entry_are_bound(self): fx = self.fixture() fx.config["expected"]["prior"]["release_set_sha256"] = "0" * 64 fx.save() self.assert_prewrite_failure(fx, "PRIOR_RELEASE_SET") fx2 = self.fixture() write(fx2.result_index, b"# not a thin entry\n") self.assert_prewrite_failure(fx2, "RESULT_INDEX_IDENTITY") def test_11a_every_ledger_column_tamper_is_fail_closed(self): replacements = { "release_id": "attacked-release", "event_seq": "999", "state": "ABORTED", "prior_release_id": "a" * 64, "prior_release_set_sha256": "A" * 64, "candidate_release_set_sha256": "B" * 64, "task_id": "TASK-ATTACKED", "case_id": "CASE-ATTACKED", "batch_id": "BATCH-ATTACKED", "run_id": "RUN-ATTACKED", "accepted_audit_id": "AUDIT-ATTACKED", "operator": "ATTACKED-OPERATOR", "process_identity": "PID-ATTACKED-INSTANCE", "event_time": "2099-01-01T00:00:00.000000Z", "recovery_or_rollback_receipt": "receipts/attacked.json", "attempt_id": "ATTEMPT-ATTACKED-999999", "attempt_seq": "999", } self.assertEqual(set(replacements), set(LEDGER_HEADER)) for column in LEDGER_HEADER: with self.subTest(column=column): fx = self.fixture() first, committed = fx.run() self.assertEqual(first.returncode, 0, first.stdout + first.stderr) self.assertEqual(committed["status"], "COMMITTED") rows = ledger_rows(read(fx.ledger)) rows[-1][column] = replacements[column] write(fx.ledger, csv_bytes(LEDGER_HEADER, [[row[key] for key in LEDGER_HEADER] for row in rows])) replay, result = fx.run() self.assertNotEqual(replay.returncode, 0, replay.stdout + replay.stderr) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertNotEqual(result["status"], "IDEMPOTENT_COMMITTED") self.assertFalse(fx.lock.exists()) def test_11b_every_state_receipt_and_time_are_bound(self): for column in ("recovery_or_rollback_receipt", "event_time"): for state_index in range(4): with self.subTest(column=column, state_index=state_index): fx = self.fixture() first, committed = fx.run() self.assertEqual(first.returncode, 0, first.stdout + first.stderr) self.assertEqual(committed["status"], "COMMITTED") rows = ledger_rows(read(fx.ledger)) attempt_indexes = [index for index, row in enumerate(rows) if row["attempt_id"] == fx.attempt] target = rows[attempt_indexes[state_index]] if column == "recovery_or_rollback_receipt": target[column] = f"receipts/attacked-{state_index}.json" else: target[column] = f"2099-01-01T00:00:0{state_index}.000000Z" write(fx.ledger, csv_bytes(LEDGER_HEADER, [[row[key] for key in LEDGER_HEADER] for row in rows])) replay, result = fx.run() self.assertNotEqual(replay.returncode, 0, replay.stdout + replay.stderr) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertNotEqual(result["status"], "IDEMPOTENT_COMMITTED") self.assertFalse(fx.lock.exists()) def test_12_candidate_history_root_and_schema_negatives(self): fx = self.fixture() write(fx.candidate / fx.core_source_rel, b"drift\n") self.assert_prewrite_failure(fx, "CANDIDATE_IDENTITY") fx2 = self.fixture() long_row = next(row for row in fx2.history_rows if len(os.path.abspath(fx2.history / row["relative_path"])) >= 260) write(fx2.history / long_row["relative_path"], b"drift\n") self.assert_prewrite_failure(fx2, "HISTORY_IDENTITY") fx3 = self.fixture() fx3.config["roots"]["candidate_root"] = "../escape" fx3.save() self.assert_prewrite_failure(fx3, "PATH_RELATIVE") fx4 = self.fixture() del fx4.config["identity"]["operator"] fx4.save() self.assert_prewrite_failure(fx4, "CONFIG_KEYS") def test_13_audit_handoff_domains_remain_closed(self): fx = self.fixture() fx.config["identity"]["canonical_audit_id"], fx.config["identity"]["review_handoff_id"] = fx.config["identity"]["review_handoff_id"], fx.config["identity"]["canonical_audit_id"] fx.save() self.assert_prewrite_failure(fx, "AUDIT_ID_DOMAIN") if __name__ == "__main__": unittest.main(verbosity=2)