from __future__ import annotations 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" RUN_ROOT = PROJECT_ROOT / "ana-data" / "tmp" / "半导体案例" / "TASK-ANA-SEMI-CANONICAL-MIGRATION-20260829-001" / "RUN-ANA-SEMI-CANONICAL-MIGRATION-20260829-001" sys.path.insert(0, str(RUN_ROOT)) from test_execution_binding_v002 import materialize_fixture, run_publisher def long(path: Path) -> str: value = os.path.abspath(path) if os.name != "nt" or value.startswith("\\\\?\\"): return value if value.startswith("\\\\"): return "\\\\?\\UNC\\" + value[2:] return "\\\\?\\" + value def read(path: Path) -> bytes: with open(long(path), "rb") as handle: return handle.read() def write(path: Path, data: bytes) -> None: os.makedirs(long(path.parent), exist_ok=True) with open(long(path), "xb") as handle: handle.write(data) def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" 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) class Fixture: def __init__(self, fault: str = "NONE", template: Path | None = None, legacy_backup: Path | None = None): self._owns_root = template is None self.root = Path(tempfile.mkdtemp(prefix="ana-semi-postcommit-archive-")) if template is None else template if template is None: self.publisher, self.publisher_config, self.bootstrap = materialize_fixture(self.root) code, terminal, stderr = run_publisher(self.root, self.publisher_config) if (code, terminal.get("status")) != (0, "COMMITTED"): raise AssertionError((code, terminal, stderr)) else: self.publisher_config = self.root / "batch_config.json" self.publisher = json.loads(read(self.publisher_config).decode("utf-8")) self.bootstrap = json.loads(read(RUN_ROOT / "successor_binding_v002/evidence/genesis_bootstrap_plan.json").decode("utf-8")) self.source = self.root / Path(self.publisher["roots"]["history_root"]) self.target = self.root / Path(self.bootstrap["archive_target"]) self.map_path = self.root / "ana-data/cases/半导体案例/manifest/legacy_case_path_map.csv" self.receipt = self.target.parent / "postcommit_archive" / "ARCHIVE-ANA-SEMI-CANONICAL-000001" if template is not None: if legacy_backup is None: raise AssertionError("legacy backup required") shutil.rmtree(long(self.source), ignore_errors=True) shutil.rmtree(long(self.target), ignore_errors=True) shutil.rmtree(long(self.receipt), ignore_errors=True) shutil.rmtree(long(self.receipt.parent / "rollback" / self.receipt.name), ignore_errors=True) marker_prefix = self.receipt.name + ".test-fault-" if self.receipt.parent.exists(): for name in os.listdir(long(self.receipt.parent)): if name.startswith(marker_prefix): os.unlink(long(self.receipt.parent / name)) shutil.copytree(long(legacy_backup), long(self.source), copy_function=shutil.copy2) os.makedirs(long(self.receipt.parent), exist_ok=True) with io.StringIO(read(self.map_path).decode("utf-8"), newline="") as stream: rows = list(csv.DictReader(stream)) rows.sort(key=lambda row: row["legacy_path"].casefold()) source_descriptor = [{"bytes": int(row["legacy_bytes"]), "path": row["legacy_path"], "sha256": row["legacy_sha256"]} for row in rows] canonical_descriptor = [{"bytes": int(row["canonical_bytes"]), "path": row["canonical_target"], "sha256": row["canonical_sha256"]} for row in rows] publisher_id = ident(self.publisher_config) map_id = ident(self.map_path) self.config = { "schema_version": "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_CONFIG_V1", "identity": { "task_id": self.publisher["identity"]["task_id"], "case_id": self.publisher["identity"]["case_id"], "batch_id": self.publisher["identity"]["batch_id"], "run_id": self.publisher["identity"]["run_id"], "attempt_id": self.publisher["identity"]["attempt_id"], "archive_id": "ARCHIVE-ANA-SEMI-CANONICAL-000001", "operator": "test.operator", "archive_gate_audit_id": "AUDIT-ANA-SEMI-CANONICAL-MIGRATION-POSTCOMMIT-ARCHIVE-FOCUSED-REREVIEW-20260829-001", }, "roots": { "operation_root": str(self.root), "resolved_root": str(self.root.resolve()), "publisher_config_path": os.path.relpath(self.publisher_config, self.root).replace("\\", "/"), "source_root": os.path.relpath(self.source, self.root).replace("\\", "/"), "target_root": os.path.relpath(self.target, self.root).replace("\\", "/"), "receipt_root": os.path.relpath(self.receipt, self.root).replace("\\", "/"), "legacy_map_path": os.path.relpath(self.map_path, self.root).replace("\\", "/"), }, "expected": { "publisher_config_bytes": publisher_id[0], "publisher_config_sha256": publisher_id[1], "legacy_map_bytes": map_id[0], "legacy_map_sha256": map_id[1], "row_count": len(rows), "source_set_sha256": sha(canonical(source_descriptor)), "canonical_set_sha256": sha(canonical(canonical_descriptor)), }, "test_control": {"environment": "ISOLATED_TEST", "fault": fault}, } self.config_path = self.root / "archive_config.json" if self.config_path.exists(): os.unlink(long(self.config_path)) write(self.config_path, canonical(self.config)) def cleanup(self) -> None: if self._owns_root: shutil.rmtree(long(self.root), ignore_errors=True) def run(self) -> tuple[int, dict[str, object], str]: env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" completed = subprocess.run( [sys.executable, "-m", "shared_content_publisher.postcommit_archive", "--config", str(self.config_path)], cwd=self.root, env=env, text=True, encoding="utf-8", capture_output=True, timeout=180, ) line = completed.stdout.strip().splitlines()[-1] if completed.stdout.strip() else "{}" return completed.returncode, json.loads(line), completed.stderr def rollback_config(self, fault: str = "NONE") -> Path: archive_terminal = self.receipt / "terminal.json" archive_config_id = ident(self.config_path) archive_terminal_id = ident(archive_terminal) value = { "schema_version": "SHARED_CONTENT_POSTCOMMIT_ARCHIVE_ROLLBACK_CONFIG_V1", "identity": { "task_id": self.publisher["identity"]["task_id"], "case_id": self.publisher["identity"]["case_id"], "batch_id": self.publisher["identity"]["batch_id"], "run_id": self.publisher["identity"]["run_id"], "attempt_id": self.publisher["identity"]["attempt_id"], "archive_id": "ARCHIVE-ANA-SEMI-CANONICAL-000001", "rollback_id": "ROLLBACK-ANA-SEMI-CANONICAL-000001", "operator": "test.rollback.operator", "rollback_authorization_id": "AUTH-TEST-ANA-SEMI-CANONICAL-ROLLBACK-000001", "rollback_audit_id": "AUDIT-TEST-ANA-SEMI-CANONICAL-ROLLBACK-000001", }, "roots": { "operation_root": str(self.root), "resolved_root": str(self.root.resolve()), "archive_config_path": os.path.relpath(self.config_path, self.root).replace("\\", "/"), }, "binding": { "archive_config_bytes": archive_config_id[0], "archive_config_sha256": archive_config_id[1], "archive_terminal_bytes": archive_terminal_id[0], "archive_terminal_sha256": archive_terminal_id[1], }, "test_control": {"environment": "ISOLATED_TEST", "fault": fault}, } rollback_parent = self.receipt.parent / "rollback" os.makedirs(long(rollback_parent), exist_ok=True) rollback_receipt = rollback_parent / self.receipt.name if rollback_receipt.exists(): shutil.rmtree(long(rollback_receipt)) for name in os.listdir(long(rollback_parent)): if name.startswith(self.receipt.name + ".test-fault-"): os.unlink(long(rollback_parent / name)) path = self.root / "rollback_config.json" if path.exists(): os.unlink(long(path)) write(path, canonical(value)) return path def run_rollback(self, config: Path) -> tuple[int, dict[str, object], str]: env = os.environ.copy() env["PYTHONPATH"] = str(MODULE_ROOT) env["PYTHONUTF8"] = "1" env["PYTHONIOENCODING"] = "utf-8" completed = subprocess.run( [sys.executable, "-m", "shared_content_publisher.postcommit_archive", "--rollback-config", str(config)], cwd=self.root, env=env, text=True, encoding="utf-8", capture_output=True, timeout=180, ) line = completed.stdout.strip().splitlines()[-1] if completed.stdout.strip() else "{}" return completed.returncode, json.loads(line), completed.stderr class PostcommitArchiveTests(unittest.TestCase): @classmethod def setUpClass(cls): cls._template_fixture = Fixture() cls._legacy_backup_root = Path(tempfile.mkdtemp(prefix="ana-semi-postcommit-legacy-backup-")) cls._legacy_backup = cls._legacy_backup_root / "legacy" shutil.copytree(long(cls._template_fixture.source), long(cls._legacy_backup), copy_function=shutil.copy2) @classmethod def tearDownClass(cls): cls._template_fixture.cleanup() shutil.rmtree(long(cls._legacy_backup_root), ignore_errors=True) def fixture(self, fault: str = "NONE") -> Fixture: fixture = Fixture(fault, self._template_fixture.root, self._legacy_backup) self.addCleanup(fixture.cleanup) return fixture def test_01_success_and_idempotent_replay(self): fx = self.fixture() code, terminal, stderr = fx.run() self.assertEqual((code, terminal.get("status"), terminal.get("row_count")), (0, "ARCHIVED", 645), (terminal, stderr)) self.assertFalse(fx.source.exists()) self.assertTrue(fx.target.is_dir()) code, replay, stderr = fx.run() self.assertEqual((code, replay["status"]), (0, "IDEMPOTENT_ARCHIVED"), stderr) def test_02_every_durable_archive_boundary_recovers_same_config(self): faults = ( "INTERRUPT_AFTER_RECEIPT_ROOT", "INTERRUPT_AFTER_CONTRACT", "INTERRUPT_AFTER_MAP", "INTERRUPT_AFTER_ANCHOR", "INTERRUPT_BEFORE_RENAME", "INTERRUPT_AFTER_RENAME", "INTERRUPT_AFTER_TERMINAL_TEMP", ) for fault in faults: with self.subTest(fault=fault): fx = self.fixture(fault) code, stopped, _ = fx.run() self.assertEqual((code, stopped["status"], stopped["error_code"]), (20, "FAIL_CLOSED", "INJECTED_PROCESS_RESTART")) code, recovered, stderr = fx.run() self.assertEqual(code, 0, stderr) self.assertIn(recovered["status"], {"ARCHIVED", "IDEMPOTENT_ARCHIVED"}) self.assertFalse(fx.source.exists()) self.assertTrue(fx.target.is_dir()) def test_03_partial_receipt_tamper_is_recovery_required(self): fx = self.fixture("INTERRUPT_AFTER_CONTRACT") self.assertEqual(fx.run()[0], 20) contract = fx.receipt / "archive_contract.json" with open(long(contract), "ab") as handle: handle.write(b"tamper") code, terminal, _ = fx.run() self.assertEqual((code, terminal["status"], terminal["error_code"]), (30, "RECOVERY_REQUIRED", "RECOVERY_REQUIRED_RECEIPT_PARTIAL")) self.assertTrue(fx.source.is_dir()) self.assertFalse(fx.target.exists()) def test_04_source_target_truth_matrix_fails_closed(self): both = self.fixture() shutil.copytree(long(both.source), long(both.target), copy_function=shutil.copy2) code, terminal, _ = both.run() self.assertEqual((code, terminal["error_code"]), (30, "RECOVERY_REQUIRED_SOURCE_AND_TARGET")) neither = self.fixture() hidden = neither.root / "hidden-legacy" os.rename(long(neither.source), long(hidden)) code, terminal, _ = neither.run() self.assertEqual((code, terminal["error_code"]), (30, "RECOVERY_REQUIRED_SOURCE_AND_TARGET")) def test_05_postrename_archive_and_canonical_drift_are_rejected(self): archive = self.fixture("INTERRUPT_AFTER_RENAME") self.assertEqual(archive.run()[0], 20) target = next(path for path in archive.target.rglob("*") if path.is_file()) with open(long(target), "ab") as handle: handle.write(b"tamper") code, terminal, _ = archive.run() self.assertEqual((code, terminal["status"], terminal["error_code"]), (30, "RECOVERY_REQUIRED", "ARCHIVE_IDENTITY")) canonical = self.fixture() target = canonical.root / "ana-data/cases/半导体案例/核心文档/半导体的产业链总览.md" original = read(target) try: with open(long(target), "ab") as handle: handle.write(b"tamper") code, terminal, _ = canonical.run() self.assertEqual((code, terminal["status"], terminal["error_code"]), (30, "RECOVERY_REQUIRED", "PUBLISHER_NOT_COMMITTED")) finally: with open(long(target), "wb") as handle: handle.write(original) def test_06_complete_648_formal_set_drift_is_rejected_after_archive(self): paths = ( "ana-data/result/半导体案例/当前成果索引.md", "ana-data/cases/半导体案例/manifest/current_output_manifest.csv", "ana-data/cases/半导体案例/当前成果索引.md", ) for relative in paths: with self.subTest(relative=relative): fx = self.fixture() self.assertEqual(fx.run()[0], 0) path = fx.root / relative original = read(path) try: with open(long(path), "ab") as handle: handle.write(b"tamper") code, terminal, _ = fx.run() self.assertEqual( (code, terminal["status"], terminal["error_code"]), (30, "RECOVERY_REQUIRED", "PUBLISHER_FORMAL_IDENTITY"), ) finally: with open(long(path), "wb") as handle: handle.write(original) def test_07_independently_authorized_rollback_restores_legacy_and_replays(self): fx = self.fixture() self.assertEqual(fx.run()[0], 0) rollback_config = fx.rollback_config() code, terminal, stderr = fx.run_rollback(rollback_config) self.assertEqual((code, terminal["status"]), (0, "LEGACY_RESTORED"), stderr) self.assertTrue(fx.source.is_dir()) self.assertFalse(fx.target.exists()) code, terminal, stderr = fx.run_rollback(rollback_config) self.assertEqual((code, terminal["status"]), (0, "IDEMPOTENT_LEGACY_RESTORED"), stderr) code, terminal, _ = fx.run() self.assertEqual((code, terminal["error_code"]), (30, "RECOVERY_REQUIRED_ARCHIVE_ROLLED_BACK")) def test_08_authorized_rollback_boundary_restarts_are_deterministic(self): faults = ( "INTERRUPT_AFTER_ROLLBACK_ANCHOR", "INTERRUPT_BEFORE_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_RENAME", "INTERRUPT_AFTER_ROLLBACK_TERMINAL_TEMP", ) for fault in faults: with self.subTest(fault=fault): fx = self.fixture() self.assertEqual(fx.run()[0], 0) rollback_config = fx.rollback_config(fault) code, stopped, _ = fx.run_rollback(rollback_config) self.assertEqual((code, stopped["error_code"]), (20, "INJECTED_ROLLBACK_PROCESS_RESTART")) code, recovered, stderr = fx.run_rollback(rollback_config) self.assertEqual(code, 0, stderr) self.assertIn(recovered["status"], {"LEGACY_RESTORED", "IDEMPOTENT_LEGACY_RESTORED"}) def test_09_postrollback_rename_restart_revalidates_complete_formal_set(self): fx = self.fixture() self.assertEqual(fx.run()[0], 0) rollback_config = fx.rollback_config("INTERRUPT_AFTER_ROLLBACK_RENAME") code, stopped, _ = fx.run_rollback(rollback_config) self.assertEqual((code, stopped["error_code"]), (20, "INJECTED_ROLLBACK_PROCESS_RESTART")) self.assertTrue(fx.source.is_dir()) self.assertFalse(fx.target.exists()) result_index = fx.root / "ana-data/result/半导体案例/当前成果索引.md" original = read(result_index) try: with open(long(result_index), "ab") as handle: handle.write(b"tamper-after-rollback-rename") code, terminal, _ = fx.run_rollback(rollback_config) self.assertEqual( (code, terminal["status"], terminal["error_code"]), (30, "RECOVERY_REQUIRED", "PUBLISHER_FORMAL_IDENTITY"), ) finally: with open(long(result_index), "wb") as handle: handle.write(original) if __name__ == "__main__": unittest.main()