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()
|
