import copy import csv import hashlib import io import json import os import shutil import subprocess import sys import tempfile import unittest from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[3] MODULE_ROOT = PROJECT_ROOT / "dev" / "ana-dev" PYTHON = sys.executable sys.path.insert(0, str(MODULE_ROOT)) from shared_content_publisher import read_direct_current_snapshot DIRECT_GUARD_PATH = PROJECT_ROOT / "dev-doc" / "ana-doc" / "开发方案" / "SHARED-CONTENT-PUBLISHER-DIRECT-STABLE-PATH-PRODUCTION-GUARD-V001.json" DIRECT_SCHEMA_PATH = PROJECT_ROOT / "dev" / "ana-dev" / "shared_content_publisher" / "batch_config.direct_stable_path_set.schema.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 sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest().upper() def ident(path: Path) -> tuple[int, str]: data = read(path) return len(data), sha(data) 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 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 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) class DirectFixture: def __init__(self, root: Path, fault="NONE", after=None): self.root = root self.industry = "DEMO-DIRECT" self.task = "TASK-DEMO-DIRECT" self.case = "CASE-DEMO-DIRECT" self.batch = "BATCH-DEMO-DIRECT" self.run_id = "RUN-DEMO-DIRECT" self.attempt = "ATTEMPT-DEMO-DIRECT-000002" self.audit = "AUDIT-DEMO-DIRECT-PASS" self.handoff = "HANDOFF-DEMO-DIRECT-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), ("accepted_audit_id", self.audit), )) self.release = hashlib.sha256(preimage.encode()).hexdigest() self.case_root = root / "ana-data" / "cases" / "Demo案例" self.core_root = self.case_root / "核心文档" self.manifest_root = self.case_root / "manifest" self.result_root = root / "ana-data" / "result" / "Demo案例" self.case_result_root = self.result_root / self.case self.candidate = root / "ana-data" / "tmp" / "Demo案例" / "candidate" self.staging_parent = root / "ana-data" / "tmp" / "Demo案例" / "staging" self.staging = self.staging_parent / self.attempt self.history = self.case_root / "审计包" / "TASK-DEMO" / "attempts" / "ATTEMPT-DEMO-DIRECT-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.core_existing = self.core_root / "overview.md" self.core_new = self.core_root / "产业链" / "new.md" self.current_manifest = self.manifest_root / "current_output_manifest.csv" self.legacy_map = self.manifest_root / "legacy_case_path_map.csv" self.case_result = self.case_result_root / "result_index.md" self.ledger = self.manifest_root / "promotion_ledger.csv" self.lock = self.manifest_root / ".promotion.lock" for directory in ( self.core_root, self.core_new.parent, self.manifest_root, self.result_root, self.case_result_root, self.candidate, self.staging_parent, self.history, ): os.makedirs(long(directory), exist_ok=True) self.prior_index_data = b"# Demo current\n\n- release: prior\n- core: [overview](core-old)\n" self.prior_core_data = b"# Prior overview\n\nprior-evidence\n" self.prior_result_data = "# Demo result\n\n- cases: ../../cases/Demo案例/当前成果索引.md\n".encode() write(self.case_index, self.prior_index_data) write(self.core_existing, self.prior_core_data) write(self.result_index, self.prior_result_data) prior_rows_data = [ ("CASES-CURRENT", self.rel(self.case_index), "CASES_CURRENT_INDEX", *ident(self.case_index)), ("CORE-EXISTING", self.rel(self.core_existing), "CORE_DOCUMENT", *ident(self.core_existing)), ("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, ) write(self.current_manifest, prior_manifest) self.prior_release_set = release_set(prior_manifest, self.rel(self.current_manifest)) 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): rel = "/".join( [f"long-{index:02d}"] + [(f"segment-{level:02d}-" + "x" * 22) for level in range(7)] + ["receipt.json"] ) write(self.history / rel, canonical({"long": index})) self.history_rows.append(self.history_row(rel)) assert len(self.history_rows) == 63 sources = { "core/existing.md": b"# Candidate overview\n\nsource-evidence-closed\n", "core/new.md": b"# Candidate new chain\n\nsource-evidence-closed\n", "manifest/legacy.csv": b"old,new\nlegacy,new\n", "result/current.md": self.prior_result_data, "result/case.md": b"# Case result\n\nsource-evidence-closed\n", "entry/current.md": ( "# Demo current\n\n- release: candidate\n" "- core: [overview](核心文档/overview.md)\n" "- evidence: source-evidence-closed\n" ).encode("utf-8"), } for rel, data in sources.items(): write(self.candidate / rel, data) preliminary = [ self.row("CORE-EXISTING", "core/existing.md", self.core_existing, "CORE_DOCUMENT", "RELEASE_MEMBER"), self.row("CORE-NEW", "core/new.md", self.core_new, "CORE_DOCUMENT", "RELEASE_MEMBER"), self.row("LEGACY-MAP", "manifest/legacy.csv", self.legacy_map, "LEGACY_PATH_MAP", "RELEASE_MEMBER"), self.row("RESULT-CURRENT", "result/current.md", self.result_index, "RESULT_THIN_CURRENT_INDEX", "RELEASE_MEMBER"), self.row("RESULT-CASE", "result/case.md", self.case_result, "CASE_RESULT_INDEX", "RELEASE_MEMBER"), self.row("CASES-CURRENT", "entry/current.md", self.case_index, "CASES_CURRENT_INDEX", "CASE_CURRENT_INDEX"), ] current_rows = [ (row["member_id"], row["formal_relative_path"], row["artifact_type"], row["bytes"], row["sha256"]) for row in preliminary if row["member_id"] in {"CORE-EXISTING", "CORE-NEW", "RESULT-CURRENT", "CASES-CURRENT"} ] current_manifest_candidate = csv_bytes( ["member_id", "formal_relative_path", "artifact_type", "bytes", "sha256"], current_rows, ) self.current_manifest_source = "manifest/current.csv" write(self.candidate / self.current_manifest_source, current_manifest_candidate) self.candidate_rows = preliminary[:2] + [ self.row("CURRENT-MANIFEST", self.current_manifest_source, self.current_manifest, "CURRENT_OUTPUT_MANIFEST", "RELEASE_MEMBER"), ] + preliminary[2:] self.candidate_manifest_rel = "candidate_output_manifest.csv" 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 ], ) write(self.candidate / self.candidate_manifest_rel, candidate_manifest) candidate_manifest_id = ident(self.candidate / self.candidate_manifest_rel) current_manifest_id = ident(self.candidate / self.current_manifest_source) self.candidate_set = release_set(current_manifest_candidate, self.rel(self.current_manifest)) prior_ledger_row = [ "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:00.000000Z", "receipts/old.json", "ATTEMPT-DEMO-DIRECT-000001", "1", ] ledger = csv_bytes(LEDGER_HEADER, [prior_ledger_row]) write(self.ledger, ledger) destination_prior = [] present = { "CORE-EXISTING": self.core_existing, "CURRENT-MANIFEST": self.current_manifest, "RESULT-CURRENT": self.result_index, "CASES-CURRENT": self.case_index, } for row in self.candidate_rows: path = self.root / Path(row["formal_relative_path"]) if row["member_id"] in present: size, digest = ident(path) destination_prior.append({ "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "present": True, "object_type": "FILE", "reparse": False, "bytes": size, "sha256": digest, }) else: destination_prior.append({ "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "present": False, "object_type": "ABSENT", "reparse": False, "bytes": None, "sha256": None, }) state_id = ident(self.history / self.state_rel) prior_manifest_id = ident(self.current_manifest) result_id = ident(self.result_index) self.config = { "schema_version": "SHARED_CONTENT_PUBLISHER_CONFIG_V3", "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), "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-DIRECT-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, "current_manifest_bytes": current_manifest_id[0], "current_manifest_sha256": current_manifest_id[1], "manifest_self_formal_relative_path": self.rel(self.current_manifest), "rows": self.candidate_rows, "link_checks": [{"relative_path": "entry/current.md", "required_utf8_substrings": ["核心文档/overview.md", "source-evidence-closed"]}], "evidence_checks": [{"relative_path": "core/new.md", "required_utf8_substrings": ["source-evidence-closed"]}], }, "prior": { "release_id": self.prior_release, "manifest_formal_relative_path": self.rel(self.current_manifest), "manifest_bytes": prior_manifest_id[0], "manifest_sha256": prior_manifest_id[1], "manifest_self_formal_relative_path": self.rel(self.current_manifest), "release_set_sha256": self.prior_release_set, "rows": self.prior_rows, }, "destination_prior": destination_prior, "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-old"], }, "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": "DIRECT_STABLE_PATH_SET_V1", "staging_relative_path": self.rel(self.staging)}, "test_control": { "environment": "ISOLATED_TEST", "fault": fault, "race_marker_relative_path": "", "fault_commit_after": after, }, } self.config_path = self.root / "batch-config.json" self.prior_file_state = self.formal_identities() 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 row(self, member, source, target, artifact_type, role): size, digest = ident(self.candidate / source) return { "member_id": member, "relative_path": source, "formal_relative_path": self.rel(target), "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): env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" command = [PYTHON, "-m", "shared_content_publisher", "--config", str(self.config_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=60, ) return proc, json.loads(proc.stdout.strip().splitlines()[-1]) def read_current(self): env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" proc = subprocess.run( [PYTHON, "-m", "shared_content_publisher", "--config", str(self.config_path), "--read-direct-current"], cwd=self.root, env=env, text=True, encoding="utf-8", capture_output=True, timeout=60, ) 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_identities(self): result = {} for row in self.config["expected"]["destination_prior"] if hasattr(self, "config") else []: path = self.root / Path(row["formal_relative_path"]) result[row["member_id"]] = ident(path) if path.exists() else None return result def asserted_prior(self): for state in self.config["expected"]["destination_prior"]: path = self.root / Path(state["formal_relative_path"]) if state["present"]: self_test = ident(path) if self_test != (state["bytes"], state["sha256"]): raise AssertionError((state["member_id"], self_test)) elif path.exists(): raise AssertionError((state["member_id"], "expected absent")) def accepted_topology_fixture(root: Path) -> DirectFixture: """Build an isolated transaction from the exact accepted 10-row package.""" fx = object.__new__(DirectFixture) fx.root = root fx.industry = "农业案例" fx.task = "TASK-AGRICULTURE-PESTICIDE-FERTILIZER-20260818-001" fx.case = "ANA-AGRICULTURE-PESTICIDE-FERTILIZER-20260818-001" fx.batch = "BATCH-004" fx.run_id = "RUN-ANA-AGRICULTURE-PESTICIDE-FERTILIZER-20260818-001-BATCH-004-001" fx.attempt = "ATTEMPT-B004-000003-ISOLATED" fx.audit = "AUDIT-ANA-AGRICULTURE-STABLE-CORE-CONSOLIDATION-BATCH004-EXECUTION-OUTPUT-REPAIR006-REREVIEW-20260820-001" fx.handoff = "HANDOFF-AGRICULTURE-ANALYST-LAOSHEN-STABLE-CORE-CONSOLIDATION-BATCH004-EXECUTION-OUTPUT-REPAIR006-REREVIEW-20260820-001" fx.release = "f39db200057f733df3d34ebc0387000d6af4a29d2a3e3ed787c20268b5fcace7" fx.prior_release = "b4f588b1b717fa58b41bf5667e1ce820bd7e99dfa467a39d9bd1e4ad978cde5d" production_candidate = PROJECT_ROOT / "ana-data" / "tmp" / "农业案例" / fx.task / fx.run_id / "candidate" fx.candidate = root / "ana-data" / "tmp" / "农业案例" / fx.task / fx.run_id / "candidate" shutil.copytree(long(production_candidate), long(fx.candidate)) fx.case_root = root / "ana-data" / "cases" / "农业案例" fx.core_root = fx.case_root / "核心文档" fx.manifest_root = fx.case_root / "manifest" fx.result_root = root / "ana-data" / "result" / "农业案例" fx.case_result_root = fx.result_root / fx.case fx.case_index = fx.case_root / "当前成果索引.md" fx.result_index = fx.result_root / "当前成果索引.md" fx.current_manifest = fx.manifest_root / "current_output_manifest.csv" fx.ledger = fx.manifest_root / "promotion_ledger.csv" fx.lock = fx.manifest_root / ".promotion.lock" fx.history = fx.case_root / "审计包" / fx.case / fx.batch / fx.run_id / "attempts" / "ATTEMPT-B004-000001" fx.receipt = fx.case_root / "审计包" / fx.case / fx.batch / fx.run_id / "attempts" / fx.attempt fx.staging_parent = root / "ana-data" / "tmp" / "农业案例" / "publisher-staging" fx.staging = fx.staging_parent / fx.attempt for directory in (fx.core_root / "产业链", fx.manifest_root, fx.case_result_root, fx.history, fx.staging_parent): os.makedirs(long(directory), exist_ok=True) candidate_manifest_path = fx.candidate / "manifest" / "candidate_output_manifest.csv" candidate_reader = csv.DictReader(io.StringIO(read(candidate_manifest_path).decode("utf-8"), newline="")) manifest_rows = list(candidate_reader) self_assert = [row["member_id"] for row in manifest_rows] if len(manifest_rows) != 10 or len(set(self_assert)) != 10: raise AssertionError("accepted candidate topology is not 10 unique rows") fx.candidate_rows = [] for item in manifest_rows: source = item["candidate_relative_path"].removeprefix("candidate/") fx.candidate_rows.append({ "member_id": item["member_id"], "relative_path": source, "formal_relative_path": item["formal_relative_path"], "artifact_type": item["artifact_type"], "commit_role": "CASE_CURRENT_INDEX" if item["member_id"] == "CASES-CURRENT-INDEX" else "RELEASE_MEMBER", "bytes": int(item["bytes"]), "sha256": item["sha256"], }) os.makedirs(long((root / Path(item["formal_relative_path"])).parent), exist_ok=True) # The accepted BATCH-004 candidate has since been published to production, # so the live industry root is no longer a valid source for its historical # prior state. Bind this fixture to the immutable prior snapshot captured # by the real ATTEMPT-B004-000003 instead of allowing later business # publication to make prior and candidate indistinguishable. production_snapshot = ( PROJECT_ROOT / "ana-data" / "cases" / "农业案例" / "审计包" / fx.case / fx.batch / fx.run_id / "attempts" / "ATTEMPT-B004-000003" / "prior_snapshot" / "files" ) production_present = { row["formal_relative_path"]: production_snapshot / Path(row["formal_relative_path"]) for row in fx.candidate_rows if os.path.isfile(long(production_snapshot / Path(row["formal_relative_path"]))) } for formal, source in production_present.items(): write(root / Path(formal), read(source)) prior_manifest_data = read( production_snapshot / "ana-data" / "cases" / "农业案例" / "manifest" / "current_output_manifest.csv" ) prior_reader = csv.DictReader(io.StringIO(prior_manifest_data.decode("utf-8"), newline="")) prior_rows_raw = list(prior_reader) fx.prior_rows = [{ "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "artifact_type": row["artifact_type"], "bytes": int(row["bytes"]), "sha256": row["sha256"], } for row in prior_rows_raw] fx.prior_release_set = release_set(prior_manifest_data, "ana-data/cases/农业案例/manifest/current_output_manifest.csv") fx.prior_index_data = read(fx.case_index) fx.state_rel = "rollback_current_validation.json" state = canonical({"semantic_state": "RECOVERY_REQUIRED", "status": "BLOCKED", "exit_code": 30}) write(fx.history / fx.state_rel, state) fx.history_rows = [fx.history_row(fx.state_rel)] ledger = csv_bytes(LEDGER_HEADER, [[ "old-release", "1", "ROLLED_BACK", fx.prior_release, fx.prior_release_set, "C" * 64, fx.task, fx.case, fx.batch, fx.run_id, "AUDIT-OLD", "tester", "PID-OLD-INSTANCE", "2026-01-01T00:00:00.000000Z", "receipts/old.json", "ATTEMPT-B004-000001", "1", ]]) write(fx.ledger, ledger) destination_prior = [] for row in fx.candidate_rows: target = root / Path(row["formal_relative_path"]) if target.is_file(): size, digest = ident(target) destination_prior.append({ "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "present": True, "object_type": "FILE", "reparse": False, "bytes": size, "sha256": digest, }) else: destination_prior.append({ "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "present": False, "object_type": "ABSENT", "reparse": False, "bytes": None, "sha256": None, }) current_candidate = fx.candidate / "manifest" / "current_output_manifest.csv" result_id = ident(fx.result_index) state_id = ident(fx.history / fx.state_rel) fx.config = { "schema_version": "SHARED_CONTENT_PUBLISHER_CONFIG_V3", "identity": { "industry": fx.industry, "task_id": fx.task, "case_id": fx.case, "batch_id": fx.batch, "run_id": fx.run_id, "attempt_id": fx.attempt, "canonical_audit_id": fx.audit, "review_handoff_id": fx.handoff, "release_id": fx.release, "prior_release_id": fx.prior_release, "candidate_release_set_sha256": "45ECBC310A2BC6E5B77F5F036641C2BAE7C9C97898EBB7206D9EEA1A9104F1C1", "operator": "test.operator", }, "roots": { "operation_root": str(root), "resolved_root": str(root), "volume_identity": (root.drive or str(os.stat(root).st_dev)).upper(), "candidate_root": fx.rel(fx.candidate), "case_current_index_path": fx.rel(fx.case_index), "result_current_index_path": fx.rel(fx.result_index), "ledger_path": fx.rel(fx.ledger), "lock_path": fx.rel(fx.lock), "attempt_receipt_root": fx.rel(fx.receipt), "history_root": fx.rel(fx.history), "current_state_relative_path": fx.state_rel, }, "expected": { "ledger": { "bytes": len(ledger), "sha256": sha(ledger), "next_event_seq": 2, "next_attempt_seq": 2, "prior_attempt_id": "ATTEMPT-B004-000001", "prior_terminal_state": "ROLLED_BACK", }, "candidate": { "manifest_relative_path": "manifest/candidate_output_manifest.csv", "manifest_bytes": ident(candidate_manifest_path)[0], "manifest_sha256": ident(candidate_manifest_path)[1], "current_manifest_relative_path": "manifest/current_output_manifest.csv", "current_manifest_bytes": ident(current_candidate)[0], "current_manifest_sha256": ident(current_candidate)[1], "manifest_self_formal_relative_path": "ana-data/cases/农业案例/manifest/current_output_manifest.csv", "rows": fx.candidate_rows, "link_checks": [{"relative_path": "cases/当前成果索引.md", "required_utf8_substrings": ["核心文档/农业行业视图.md", "accepted_audit_id="]}], "evidence_checks": [{"relative_path": "core/产业链/化肥产业链_重点研究.md", "required_utf8_substrings": ["证据状态", "UNKNOWN"]}], }, "prior": { "release_id": fx.prior_release, "manifest_formal_relative_path": "ana-data/cases/农业案例/manifest/current_output_manifest.csv", "manifest_bytes": len(prior_manifest_data), "manifest_sha256": sha(prior_manifest_data), "manifest_self_formal_relative_path": "ana-data/cases/农业案例/manifest/current_output_manifest.csv", "release_set_sha256": fx.prior_release_set, "rows": fx.prior_rows, }, "destination_prior": destination_prior, "history": {"rows": fx.history_rows, "minimum_long_paths": 0, "long_path_threshold": 260}, "case_current_index": { "bytes": len(fx.prior_index_data), "sha256": sha(fx.prior_index_data), "required_utf8_substrings": ["农业案例", "当前成果"], }, "result_current_index": { "member_id": "RESULT-CURRENT-INDEX", "formal_relative_path": fx.rel(fx.result_index), "artifact_type": "RESULT_THIN_CURRENT_INDEX", "bytes": result_id[0], "sha256": result_id[1], "required_utf8_substrings": ["cases/农业案例/当前成果索引.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": "DIRECT_STABLE_PATH_SET_V1", "staging_relative_path": fx.rel(fx.staging)}, "test_control": {"environment": "ISOLATED_TEST", "fault": "NONE", "race_marker_relative_path": "", "fault_commit_after": None}, } fx.config_path = root / "accepted-topology-config.json" fx.save() return fx class DirectStablePathPublisherTests(unittest.TestCase): def fixture(self, fault="NONE", after=None): root = Path(tempfile.mkdtemp(prefix="shared-direct-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda: shutil.rmtree(long(root), ignore_errors=True)) return DirectFixture(root, fault=fault, after=after) def genesis_fixture(self, fault="NONE", after=None): """Project-shaped first publication with every stable destination absent.""" fx = self.fixture(fault=fault, after=after) for path in (fx.case_index, fx.core_existing, fx.result_index, fx.current_manifest): os.unlink(long(path)) state_path = fx.history / fx.state_rel os.unlink(long(state_path)) fx.history_rows = [row for row in fx.history_rows if row["relative_path"] != fx.state_rel] fx.config["expected"]["history"]["rows"] = fx.history_rows fx.config["expected"]["prior"] = { "mode": "ABSENT_GENESIS_V1", "release_id": fx.prior_release, "release_set_sha256": fx.prior_release_set, "rows": [], } fx.config["expected"]["destination_prior"] = [ { "member_id": row["member_id"], "formal_relative_path": row["formal_relative_path"], "present": False, "object_type": "ABSENT", "reparse": False, "bytes": None, "sha256": None, } for row in fx.candidate_rows ] fx.config["expected"]["case_current_index"] = {"present": False} candidate_result = next( row for row in fx.candidate_rows if row["member_id"] == "RESULT-CURRENT" ) fx.config["expected"]["result_current_index"] = { "member_id": candidate_result["member_id"], "formal_relative_path": candidate_result["formal_relative_path"], "artifact_type": candidate_result["artifact_type"], "present": False, "required_utf8_substrings": ["../../cases/Demo案例/当前成果索引.md"], } fx.config["expected"]["current_state"] = {"present": False} fx.prior_file_state = fx.formal_identities() fx.save() return fx def test_00_production_guard_precedes_direct_isolated_mutation(self): guard = json.loads(read(DIRECT_GUARD_PATH)) self.assertEqual(guard["schema_version"], "SHARED_CONTENT_PUBLISHER_DIRECT_STABLE_PATH_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 item in guard["formal_files"]: live = PROJECT_ROOT / Path(item["relative_path"]) snapshot = prior_files / Path(item["relative_path"]) if os.path.isfile(long(snapshot)): self.assertEqual(ident(snapshot), (item["bytes"], item["sha256"])) elif item["relative_path"].endswith("/promotion_ledger.csv"): data = read(live)[:item["bytes"]] self.assertEqual((len(data), sha(data)), (item["bytes"], item["sha256"])) else: self.assertEqual(ident(live), (item["bytes"], item["sha256"])) for item in guard["accepted_candidate_files"]: self.assertEqual(ident(PROJECT_ROOT / Path(item["relative_path"])), (item["bytes"], item["sha256"])) for relative in guard["absent_paths"]: path = PROJECT_ROOT / Path(relative) if path == committed_attempt: terminal = json.loads(read(path / "terminal.json")) self.assertEqual((terminal["status"], terminal["exit_code"]), ("COMMITTED", 0)) else: self.assertFalse(path.exists(), relative) self.assertEqual( ( guard["production_config_count"], guard["production_publisher_invocation_count"], guard["production_formal_mutation_count"], ), (0, 0, 0), ) schema = json.loads(read(DIRECT_SCHEMA_PATH), object_pairs_hook=lambda pairs: self._unique_pairs(pairs)) self.assertEqual(schema["$id"], "SHARED_CONTENT_PUBLISHER_CONFIG_V3") definitions = schema["$defs"] refs = [] def collect(value): if isinstance(value, dict): if "$ref" in value: refs.append(value["$ref"]) for child in value.values(): collect(child) elif isinstance(value, list): for child in value: collect(child) collect(schema) self.assertTrue(refs) for reference in refs: self.assertTrue(reference.startswith("#/$defs/"), reference) self.assertIn(reference.rsplit("/", 1)[-1], definitions) @staticmethod def _unique_pairs(pairs): value = {} for key, item in pairs: if key in value: raise AssertionError(f"duplicate JSON key: {key}") value[key] = item return value def test_01_mixed_present_absent_commit_and_exact_idempotent_replay(self): fx = self.fixture() proc, validation = fx.run(validate=True) self.assertEqual((proc.returncode, validation["status"], validation["target_count"]), (0, "VALIDATION_PASS", 7)) proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"], terminal["commit_strategy"]), (0, "COMMITTED", "DIRECT_STABLE_PATH_SET_V1")) self.assertEqual(terminal["activation_formal_relative_path"], fx.rel(fx.case_index)) self.assertEqual(terminal["destination_count"], 7) commit_plan = json.loads(read(fx.receipt / "commit_plan.json")) self.assertEqual(commit_plan["activation_position"], 7) self.assertEqual(commit_plan["members"][-1]["member_id"], "CASES-CURRENT") proc, replay = fx.run() self.assertEqual((proc.returncode, replay["status"]), (0, "IDEMPOTENT_COMMITTED")) def test_01b_absent_genesis_commit_replay_and_prewrite_contract(self): fx = self.genesis_fixture() proc, validation = fx.run(validate=True) self.assertEqual((proc.returncode, validation["status"]), (0, "VALIDATION_PASS")) proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"]), (0, "COMMITTED")) self.assertTrue(all((fx.root / Path(row["formal_relative_path"])).exists() for row in fx.candidate_rows)) proc, replay = fx.run() self.assertEqual((proc.returncode, replay["status"]), (0, "IDEMPOTENT_COMMITTED")) appeared = self.genesis_fixture() write(appeared.case_index, b"# undeclared preexisting current\n") appeared.save() proc, blocked = appeared.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertEqual(blocked["error_code"], "GENESIS_TARGET_PRESENT") self.assertFalse(appeared.lock.exists()) def test_02a_absent_genesis_interruptions_restore_exact_absent_prior(self): for fault, after in ( ("DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION", None), ("DIRECT_FAIL_AFTER_COMMIT", 1), ("DIRECT_FAIL_AFTER_COMMIT", 6), ): with self.subTest(fault=fault, after=after): fx = self.genesis_fixture(fault=fault, after=after) first_proc, first_terminal = fx.run() self.assertNotEqual(first_proc.returncode, 0) if first_terminal["status"] != "ROLLED_BACK": second_proc, terminal = fx.run() self.assertEqual((second_proc.returncode, terminal["status"]), (20, "ROLLED_BACK")) self.assertTrue(all(not (fx.root / Path(row["formal_relative_path"])).exists() for row in fx.candidate_rows)) replay_proc, replay = fx.run() self.assertEqual((replay_proc.returncode, replay["status"]), (20, "IDEMPOTENT_ROLLED_BACK")) def test_01a_exact_accepted_ten_row_topology_commits_in_isolation(self): root = Path(tempfile.mkdtemp(prefix="shared-direct-accepted-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda: shutil.rmtree(long(root), ignore_errors=True)) fx = accepted_topology_fixture(root) proc, validation = fx.run(validate=True) self.assertEqual((proc.returncode, validation["status"], validation["target_count"]), (0, "VALIDATION_PASS", 10)) proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"], terminal["destination_count"]), (0, "COMMITTED", 10)) plan = json.loads(read(fx.receipt / "commit_plan.json")) self.assertEqual((plan["activation_position"], plan["members"][-1]["member_id"]), (10, "CASES-CURRENT-INDEX")) self.assertEqual( terminal["candidate_release_set_sha256"], "45ECBC310A2BC6E5B77F5F036641C2BAE7C9C97898EBB7206D9EEA1A9104F1C1", ) proc, replay = fx.run() self.assertEqual((proc.returncode, replay["status"], replay["destination_count"]), (0, "IDEMPOTENT_COMMITTED", 10)) for boundary in range(1, 11): with self.subTest(accepted_boundary=boundary): boundary_root = Path(tempfile.mkdtemp(prefix="shared-direct-accepted-boundary-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda path=boundary_root: shutil.rmtree(long(path), ignore_errors=True)) boundary_fx = accepted_topology_fixture(boundary_root) boundary_fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": "DIRECT_FAIL_AFTER_COMMIT", "race_marker_relative_path": "", "fault_commit_after": boundary, } boundary_fx.save() proc, rolled_back = boundary_fx.run() self.assertEqual((proc.returncode, rolled_back["status"], rolled_back["rolled_back"]), (20, "ROLLED_BACK", True)) boundary_fx.asserted_prior() def test_02_failure_at_every_commit_boundary_rolls_back_exact_prior(self): for boundary in range(1, 8): with self.subTest(boundary=boundary): fx = self.fixture("DIRECT_FAIL_AFTER_COMMIT", boundary) proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"], terminal["exit_code"]), (20, "ROLLED_BACK", 20)) self.assertIn(f"INJECTED_COMMIT_BOUNDARY:{boundary}", terminal["failure_detail"]) fx.asserted_prior() self.assertFalse(fx.lock.exists()) self.assertFalse(fx.staging.exists()) self.assertEqual(read(fx.case_index), fx.prior_index_data) states = [row["state"] for row in csv.DictReader(io.StringIO(read(fx.ledger).decode(), newline=""))] self.assertEqual(states[-5:], ["PREPARED", "VERIFIED", "COMMITTING", "ROLLING_BACK", "ROLLED_BACK"]) def test_03_extra_target_after_commit_is_removed_by_owned_rollback(self): fx = self.fixture("DIRECT_EXTRA_TARGET_AFTER_COMMIT", 4) proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"]), (20, "ROLLED_BACK")) fx.asserted_prior() self.assertFalse(any(path.name.startswith(".publisher-attack-") for path in fx.manifest_root.iterdir())) def test_04_candidate_target_and_extra_replay_tamper_never_false_idempotent(self): for mode in ("target", "extra"): with self.subTest(mode=mode): fx = self.fixture() proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"]), (0, "COMMITTED")) if mode == "target": write(fx.core_existing, b"tampered\n") else: write(fx.manifest_root / "undeclared-after-commit.txt", b"extra\n") proc, result = fx.run() self.assertNotEqual(result.get("status"), "IDEMPOTENT_COMMITTED") self.assertNotEqual(proc.returncode, 0) def test_04a_invalid_terminal_replay_uses_complete_chain_then_rolls_back(self): fx = self.fixture() proc, terminal = fx.run() self.assertEqual((proc.returncode, terminal["status"]), (0, "COMMITTED")) terminal_path = fx.receipt / "terminal.json" tampered = json.loads(read(terminal_path)) tampered["review_handoff_id"] = "HANDOFF-TAMPERED" write(terminal_path, canonical(tampered)) proc, recovered = fx.run() self.assertEqual((proc.returncode, recovered["status"], recovered["rolled_back"]), (20, "ROLLED_BACK", True)) fx.asserted_prior() self.assertTrue((fx.receipt / "recovery_terminal.json").exists()) def test_05_config_path_and_destination_contract_negatives_are_prewrite(self): mutations = [] mutations.append(("missing", lambda c: c["commit"].pop("staging_relative_path"))) mutations.append(("extra", lambda c: c["commit"].update({"extra": 1}))) mutations.append(("type", lambda c: c["expected"].update({"destination_prior": "bad"}))) mutations.append(("traversal", lambda c: c["expected"]["candidate"]["rows"][0].update({"formal_relative_path": "../escape.md"}))) mutations.append(("duplicate", lambda c: c["expected"]["candidate"]["rows"][1].update({"formal_relative_path": c["expected"]["candidate"]["rows"][0]["formal_relative_path"]}))) mutations.append(("undeclared-root", lambda c: c["expected"]["candidate"]["rows"][0].update({"formal_relative_path": "ana-data/tmp/outside.md"}))) mutations.append(("volume-drift", lambda c: c["roots"].update({"volume_identity": "Z:"}))) mutations.append(("protected-overlap", lambda c: c["commit"].update({"staging_relative_path": c["roots"]["candidate_root"]}))) for name, mutate in mutations: with self.subTest(name=name): fx = self.fixture() before = ident(fx.ledger) mutate(fx.config) fx.save() proc, result = fx.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertEqual(ident(fx.ledger), before) self.assertFalse(fx.receipt.exists()) def test_06_target_preexists_candidate_and_staging_extra_fail_closed(self): fx = self.fixture() write(fx.core_new, b"preexisting\n") proc, result = fx.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertFalse(fx.receipt.exists()) def test_06a_candidate_and_existing_target_drift_fail_before_lock(self): for mode in ("candidate", "target"): with self.subTest(mode=mode): fx = self.fixture() before = ident(fx.ledger) if mode == "candidate": write(fx.candidate / "core/new.md", b"candidate drift\n") else: write(fx.core_existing, b"target drift\n") proc, result = fx.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertEqual(ident(fx.ledger), before) self.assertFalse(fx.lock.exists()) self.assertFalse(fx.receipt.exists()) def test_06b_locked_ledger_race_is_rejected_before_first_event_append(self): fx = self.fixture() marker = fx.root / "race-marker" fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": "PAUSE_AFTER_LOCK", "race_marker_relative_path": fx.rel(marker), "fault_commit_after": None, } fx.save() process = fx.popen() deadline = __import__("time").monotonic() + 10 while __import__("time").monotonic() < deadline and not fx.lock.exists(): __import__("time").sleep(0.01) self.assertTrue(fx.lock.exists()) original = read(fx.ledger) write(fx.ledger, original + b"race\n") write(marker, b"go\n") stdout, stderr = process.communicate(timeout=20) result = json.loads(stdout.strip().splitlines()[-1]) self.assertNotEqual(process.returncode, 0, stderr) self.assertEqual(result["error_code"], "LEDGER_IDENTITY_LOCKED") self.assertFalse(fx.receipt.exists()) self.assertEqual(read(fx.ledger), original + b"race\n") fx = self.fixture() write(fx.staging / "UNDECLARED.txt", b"extra\n") proc, result = fx.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertEqual(result["status"], "FAIL_CLOSED") self.assertFalse(fx.receipt.exists()) @unittest.skipUnless(os.name == "nt", "Windows junction coverage") def test_07_reparse_ancestor_is_rejected_before_first_formal_write(self): fx = self.fixture() real = fx.root / "real-core" os.makedirs(real) os.rmdir(fx.core_new.parent) proc = subprocess.run( ["cmd", "/c", "mklink", "/J", str(fx.core_new.parent), str(real)], text=True, encoding="utf-8", errors="replace", capture_output=True, ) if proc.returncode != 0: self.skipTest(proc.stderr or proc.stdout) before = ident(fx.ledger) proc, result = fx.run(validate=True) self.assertNotEqual(proc.returncode, 0) self.assertIn(result["error_code"], {"PATH_REPARSE", "PATH_ALIAS"}) self.assertEqual(ident(fx.ledger), before) self.assertFalse(fx.receipt.exists()) def test_08_authoritative_reader_never_announces_mixed_generation_at_any_boundary(self): for boundary in range(1, 11): with self.subTest(boundary=boundary): root = Path(tempfile.mkdtemp(prefix="shared-direct-observer-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda path=root: shutil.rmtree(long(path), ignore_errors=True)) fx = accepted_topology_fixture(root) proc, prior = fx.read_current() self.assertEqual((proc.returncode, prior["status"], prior["payload_returned"]), (20, "STOP_NOT_CURRENT", False)) marker = fx.root / f"observer-{boundary}" fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": "DIRECT_PAUSE_AFTER_COMMIT", "race_marker_relative_path": fx.rel(marker), "fault_commit_after": boundary, } fx.save() process = fx.popen() ready = Path(str(marker) + ".ready") deadline = __import__("time").monotonic() + 20 while __import__("time").monotonic() < deadline and not ready.exists(): __import__("time").sleep(0.01) self.assertTrue(ready.exists(), f"boundary {boundary} did not pause") proc, observed = fx.read_current() self.assertEqual((proc.returncode, observed["status"], observed["payload_returned"]), (75, "RETRY_PUBLISH_IN_PROGRESS", False)) write(Path(str(marker) + ".continue"), b"continue\n") stdout, stderr = process.communicate(timeout=30) terminal = json.loads(stdout.strip().splitlines()[-1]) self.assertEqual((process.returncode, terminal["status"]), (0, "COMMITTED"), stderr) proc, current = fx.read_current() self.assertEqual((proc.returncode, current["status"], current["payload_returned"], current["current_announced"]), (0, "CURRENT_READ_COMPLETE", False, True)) self.assertEqual(current["destination_count"], 10) snapshot = read_direct_current_snapshot(fx.config_path) self.assertEqual((snapshot["status"], snapshot["payload_returned"], len(snapshot["payloads"])), ("CURRENT_READ_COMPLETE", True, 10)) def test_09_every_durable_prefix_and_commit_boundary_recovers_on_same_config_rerun(self): faults = [ ("DIRECT_INTERRUPT_AFTER_SNAPSHOT", None), ("DIRECT_INTERRUPT_AFTER_PREPARED", None), ("DIRECT_INTERRUPT_AFTER_VERIFIED", None), ("DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION", None), ] + [("DIRECT_INTERRUPT_AFTER_COMMIT", boundary) for boundary in range(1, 11)] for fault, boundary in faults: with self.subTest(fault=fault, boundary=boundary): root = Path(tempfile.mkdtemp(prefix="shared-direct-restart-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda path=root: shutil.rmtree(long(path), ignore_errors=True)) fx = accepted_topology_fixture(root) fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": fault, "race_marker_relative_path": "", "fault_commit_after": boundary, } fx.save() proc, interrupted = fx.run() self.assertEqual((proc.returncode, interrupted["status"], interrupted["error_code"]), (20, "FAIL_CLOSED", "INJECTED_PROCESS_RESTART")) self.assertFalse(fx.lock.exists()) self.assertTrue(fx.receipt.exists()) proc, recovered = fx.run() self.assertEqual((proc.returncode, recovered["status"], recovered["rolled_back"]), (20, "ROLLED_BACK", True)) fx.asserted_prior() self.assertFalse(fx.staging.exists()) proc, replay = fx.run() self.assertEqual((proc.returncode, replay["status"], replay["rolled_back"]), (20, "IDEMPOTENT_ROLLED_BACK", True)) def test_10_recovery_prefix_tamper_is_fail_closed_not_repeat_blocked_or_idempotent(self): attacks = ("ledger-operator", "snapshot", "receipt", "staging") for attack in attacks: with self.subTest(attack=attack): fx = self.fixture("DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION") proc, interrupted = fx.run() self.assertEqual((proc.returncode, interrupted["error_code"]), (20, "INJECTED_PROCESS_RESTART")) if attack == "ledger-operator": rows = list(csv.reader(io.StringIO(read(fx.ledger).decode("utf-8"), newline=""))) rows[-1][11] = "ATTACKED-OPERATOR" write(fx.ledger, csv_bytes(rows[0], rows[1:])) elif attack == "snapshot": target = next(path for path in (fx.receipt / "prior_snapshot" / "files").rglob("*") if path.is_file()) write(target, read(target) + b"tamper") elif attack == "receipt": target = fx.receipt / "prior_snapshot.json" value = json.loads(read(target)) value["destination_count"] += 1 write(target, canonical(value)) else: target = next(path for path in (fx.staging / "files").rglob("*") if path.is_file()) write(target, read(target) + b"tamper") proc, stopped = fx.run() self.assertNotEqual(proc.returncode, 0) self.assertNotIn(stopped.get("status"), {"IDEMPOTENT_COMMITTED", "IDEMPOTENT_ROLLED_BACK"}) self.assertNotEqual(stopped.get("error_code"), "TERMINAL_REPEAT_BLOCKED") def test_10a_each_restart_boundary_tamper_is_fail_closed(self): cases = [ ("DIRECT_INTERRUPT_AFTER_SNAPSHOT", None, "snapshot"), ("DIRECT_INTERRUPT_AFTER_PREPARED", None, "ledger"), ("DIRECT_INTERRUPT_AFTER_VERIFIED", None, "staging"), ("DIRECT_INTERRUPT_BEFORE_FIRST_DESTINATION", None, "receipt"), ] + [("DIRECT_INTERRUPT_AFTER_COMMIT", boundary, "destination") for boundary in range(1, 11)] for fault, boundary, attack in cases: with self.subTest(fault=fault, boundary=boundary): root = Path(tempfile.mkdtemp(prefix="shared-direct-restart-tamper-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda path=root: shutil.rmtree(long(path), ignore_errors=True)) fx = accepted_topology_fixture(root) fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": fault, "race_marker_relative_path": "", "fault_commit_after": boundary, } fx.save() proc, interrupted = fx.run() self.assertEqual((proc.returncode, interrupted["error_code"]), (20, "INJECTED_PROCESS_RESTART")) if attack == "snapshot": target = fx.receipt / "prior_snapshot" / "direct_prior_state.json" write(target, read(target) + b"tamper") elif attack == "ledger": rows = list(csv.reader(io.StringIO(read(fx.ledger).decode("utf-8"), newline=""))) rows[-1][11] = "ATTACKED-OPERATOR" write(fx.ledger, csv_bytes(rows[0], rows[1:])) elif attack == "staging": target = next(path for path in (fx.staging / "files").rglob("*") if path.is_file()) write(target, read(target) + b"tamper") elif attack == "receipt": target = fx.receipt / "prior_snapshot.json" value = json.loads(read(target)) value["config_sha256"] = "A" * 64 write(target, canonical(value)) else: commit_plan = json.loads(read(fx.receipt / "commit_plan.json")) member = commit_plan["members"][boundary - 1]["member_id"] row = next(item for item in fx.config["expected"]["candidate"]["rows"] if item["member_id"] == member) target = fx.root / Path(row["formal_relative_path"]) write(target, read(target) + b"tamper") proc, stopped = fx.run() self.assertNotEqual(proc.returncode, 0) self.assertNotIn(stopped.get("status"), {"IDEMPOTENT_COMMITTED", "IDEMPOTENT_ROLLED_BACK"}) def test_11_every_preanchor_write_flush_readback_and_atomic_publish_boundary_reruns(self): for fault_kind in ("DIRECT_INTERRUPT_PREANCHOR", "DIRECT_IO_ERROR_PREANCHOR"): probe = self.fixture() present = sum(1 for row in probe.config["expected"]["destination_prior"] if row["present"]) boundary_count = (present + 5) * 5 + 3 for boundary in range(1, boundary_count + 1): with self.subTest(fault=fault_kind, boundary=boundary): root = Path(tempfile.mkdtemp(prefix="shared-direct-preanchor-", dir=os.environ.get("TEMP"))) self.addCleanup(lambda path=root: shutil.rmtree(long(path), ignore_errors=True)) fx = DirectFixture(root) fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": fault_kind, "race_marker_relative_path": "", "fault_commit_after": boundary, } fx.save() proc, stopped = fx.run() fx.asserted_prior() self.assertFalse(fx.lock.exists()) preparations = list(fx.receipt.parent.glob(".recovery-anchor-*.preparing")) if boundary < boundary_count: self.assertNotEqual(proc.returncode, 0) self.assertEqual(stopped["status"], "FAIL_CLOSED") self.assertFalse(fx.receipt.exists()) self.assertEqual(len(preparations), 1) proc, resumed = fx.run() self.assertEqual((proc.returncode, resumed["status"]), (0, "COMMITTED")) elif fault_kind == "DIRECT_INTERRUPT_PREANCHOR": self.assertEqual((proc.returncode, stopped["error_code"]), (20, "INJECTED_PREANCHOR_RESTART")) self.assertTrue(fx.receipt.exists()) self.assertFalse(preparations) proc, resumed = fx.run() self.assertEqual((proc.returncode, resumed["status"], resumed["rolled_back"]), (20, "ROLLED_BACK", True)) fx.asserted_prior() else: self.assertEqual((proc.returncode, stopped["status"], stopped["rolled_back"]), (20, "ROLLED_BACK", True)) self.assertTrue(fx.receipt.exists()) self.assertFalse(preparations) proc, resumed = fx.run() self.assertEqual((proc.returncode, resumed["status"]), (20, "IDEMPOTENT_ROLLED_BACK")) fx.asserted_prior() self.assertFalse(list(fx.receipt.parent.glob(".recovery-anchor-*.preparing"))) def test_12_preanchor_tamper_ambiguity_and_official_anchor_tamper_are_recovery_required(self): for attack in ("final-member", "unexpected-member"): with self.subTest(attack=attack): fx = self.fixture() fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": "DIRECT_INTERRUPT_PREANCHOR", "race_marker_relative_path": "", "fault_commit_after": 5, } fx.save() proc, stopped = fx.run() self.assertEqual((proc.returncode, stopped["error_code"]), (20, "INJECTED_PREANCHOR_RESTART")) preparation = next(fx.receipt.parent.glob(".recovery-anchor-*.preparing")) if attack == "final-member": target = next(path for path in preparation.rglob("*") if path.is_file()) write(target, read(target) + b"tamper") else: write(preparation / "UNDECLARED.txt", b"ambiguous\n") proc, result = fx.run() self.assertEqual((proc.returncode, result["status"], result["exit_code"]), (30, "RECOVERY_REQUIRED", 30)) self.assertTrue(result["error_code"].startswith("RECOVERY_REQUIRED_PREPARATION")) fx.asserted_prior() self.assertFalse(fx.lock.exists()) self.assertFalse(fx.receipt.exists()) fx = self.fixture() present = sum(1 for row in fx.config["expected"]["destination_prior"] if row["present"]) boundary_count = (present + 5) * 5 + 3 fx.config["test_control"] = { "environment": "ISOLATED_TEST", "fault": "DIRECT_INTERRUPT_PREANCHOR", "race_marker_relative_path": "", "fault_commit_after": boundary_count, } fx.save() proc, stopped = fx.run() self.assertEqual((proc.returncode, stopped["error_code"]), (20, "INJECTED_PREANCHOR_RESTART")) manifest = fx.receipt / "recovery_anchor_manifest.json" write(manifest, read(manifest) + b"tamper") proc, result = fx.run() self.assertEqual((proc.returncode, result["status"], result["error_code"]), (30, "RECOVERY_REQUIRED", "RECOVERY_REQUIRED_OFFICIAL_ANCHOR")) fx.asserted_prior() self.assertFalse(fx.lock.exists()) if __name__ == "__main__": unittest.main()