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