547 lines
22 KiB
Python
547 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ops import build_v3_genesis_ledger_bundle as builder
|
|
from ops import run_local_genesis_ledger_rebuild as genesis_rebuild
|
|
from scripts import compile_kb_source_packet as compiler
|
|
|
|
REVIEWER_ID = "11111111-1111-4111-8111-111111111111"
|
|
SERVICE_ID = "44444444-4444-4444-8444-444444444444"
|
|
PRIVATE_MARKER = "PRIVATE-V3-MATERIAL-MUST-NOT-BE-PRINTED"
|
|
|
|
|
|
def _sha256(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def _write_manifest(path: Path) -> Path:
|
|
singleton_rows = [
|
|
{
|
|
"kind": "identity",
|
|
"database": "teleo_fixture",
|
|
"server_version_num": 160010,
|
|
"transaction_read_only": "on",
|
|
"server_address": None,
|
|
"server_port": None,
|
|
"ssl": False,
|
|
"ssl_version": None,
|
|
"database_collation": "C",
|
|
"database_ctype": "C",
|
|
},
|
|
{"kind": "schemas", "items": ["kb_stage", "public"]},
|
|
]
|
|
singleton_rows.extend(
|
|
{"kind": kind, "items": []}
|
|
for kind in sorted(genesis_rebuild.load_manifest.__globals__["SINGLETON_KINDS"] - {"identity", "schemas"})
|
|
)
|
|
rows = [
|
|
*singleton_rows,
|
|
{
|
|
"kind": "table",
|
|
"schema": "public",
|
|
"table": "claims",
|
|
"row_count": 1,
|
|
"rowset_md5": "f" * 32,
|
|
},
|
|
]
|
|
path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _compiled_child(tmp_path: Path, *, salt: str) -> dict:
|
|
artifact_bytes = f"retained artifact {salt}\n".encode()
|
|
quote = f"The exact retained source {salt} requires review before canonical apply."
|
|
extracted_text = f"Header {salt}\n{quote}\n"
|
|
text_bytes = extracted_text.encode()
|
|
quote_bytes = quote.encode()
|
|
start = text_bytes.index(quote_bytes)
|
|
extractor_spec = f"fixture-extractor={salt}\n".encode()
|
|
artifact_sha256 = _sha256(artifact_bytes)
|
|
|
|
manifest = {
|
|
"schema": compiler.MANIFEST_SCHEMA_V3,
|
|
"artifact_sha256": artifact_sha256,
|
|
"extracted_text_sha256": _sha256(text_bytes),
|
|
"extractor": {
|
|
"name": "fixture-semantic-extractor",
|
|
"version": "3.1.0",
|
|
"spec_sha256": _sha256(extractor_spec),
|
|
},
|
|
"agent_id": REVIEWER_ID,
|
|
"source": {
|
|
"identity": f"document:sha256:{artifact_sha256}",
|
|
"source_key": f"source_{salt}",
|
|
"source_type": "article",
|
|
"title": f"Source {salt}",
|
|
"locator": f"artifact://sha256/{artifact_sha256}",
|
|
"access_scope": "collective_internal",
|
|
"ingestion_origin": "operator_upload",
|
|
"provenance_status": "verified",
|
|
"source_class": "primary_record",
|
|
"auto_promotion_policy": "human_required",
|
|
"captured_at": "2026-07-18T10:00:00+00:00",
|
|
},
|
|
"claims": [
|
|
{
|
|
"claim_key": f"review_gate_{salt}",
|
|
"type": "empirical",
|
|
"proposition": f"Source {salt} remains non-canonical until guarded review and apply succeed.",
|
|
"body_markdown": f"The retained source {salt} explicitly preserves the review boundary.",
|
|
"scope": f"The exact retained source fixture {salt}.",
|
|
"access_scope": "collective_internal",
|
|
"owner_agent_id": None,
|
|
"status": "active",
|
|
"revision_criteria": "Revise if the exact source bytes or review contract changes.",
|
|
"revision_kind": "original",
|
|
"supersedes_id": None,
|
|
"tags": ["genesis", "provenance", salt],
|
|
"evidence": [
|
|
{
|
|
"evidence_key": f"exact_quote_{salt}",
|
|
"quote": quote,
|
|
"polarity": "supports",
|
|
"locator": {
|
|
"unit": "utf8_bytes",
|
|
"start": start,
|
|
"end": start + len(quote_bytes),
|
|
},
|
|
"rationale": "The exact byte-bound sentence directly supports the bounded proposition.",
|
|
}
|
|
],
|
|
"assessment": {
|
|
"support_tier": "strong",
|
|
"challenge_tier": "none",
|
|
"overall_state": "support_dominant",
|
|
"rationale": "One exact verified primary-record sentence supports the claim.",
|
|
"supersedes_assessment_id": None,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
input_dir = tmp_path / f"compiler-{salt}"
|
|
input_dir.mkdir()
|
|
artifact = input_dir / "artifact.bin"
|
|
text = input_dir / "extracted.txt"
|
|
manifest_path = input_dir / "manifest.json"
|
|
extractor_path = input_dir / "extractor.spec"
|
|
artifact.write_bytes(artifact_bytes)
|
|
text.write_text(extracted_text, encoding="utf-8")
|
|
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
extractor_path.write_bytes(extractor_spec)
|
|
return compiler.compile_source_packet(artifact, text, manifest_path, extractor_path)["strict_child_proposal"]
|
|
|
|
|
|
def _applied_envelope(proposal: dict) -> dict:
|
|
fields = (
|
|
"id",
|
|
"proposal_type",
|
|
"status",
|
|
"payload",
|
|
"reviewed_by_handle",
|
|
"reviewed_by_agent_id",
|
|
"reviewed_at",
|
|
"review_note",
|
|
"applied_by_handle",
|
|
"applied_by_agent_id",
|
|
"applied_at",
|
|
)
|
|
return {field: proposal[field] for field in fields}
|
|
|
|
|
|
def _v3_material(tmp_path: Path, *, sequence: int, salt: str) -> dict:
|
|
child = _compiled_child(tmp_path, salt=salt)
|
|
approved = {
|
|
"id": child["id"],
|
|
"proposal_type": child["proposal_type"],
|
|
"status": "approved",
|
|
"proposed_by_handle": child["proposed_by_handle"],
|
|
"proposed_by_agent_id": child["proposed_by_agent_id"],
|
|
"channel": child["channel"],
|
|
"source_ref": child["source_ref"],
|
|
"rationale": PRIVATE_MARKER,
|
|
"payload": child["payload"],
|
|
"reviewed_by_handle": "m3taversal",
|
|
"reviewed_by_agent_id": REVIEWER_ID,
|
|
"reviewed_at": "2026-07-18T10:01:00.000000Z",
|
|
"review_note": "Reviewed the exact source-normalized V3 payload.",
|
|
"applied_by_handle": None,
|
|
"applied_by_agent_id": None,
|
|
"applied_at": None,
|
|
"created_at": "2026-07-18T10:00:30.000000Z",
|
|
"updated_at": "2026-07-18T10:01:00.000000Z",
|
|
}
|
|
applied = {
|
|
**approved,
|
|
"status": "applied",
|
|
"applied_by_handle": "kb-apply",
|
|
"applied_by_agent_id": SERVICE_ID,
|
|
"applied_at": "2026-07-18T10:02:00.000000Z",
|
|
"updated_at": "2026-07-18T10:02:00.000001Z",
|
|
}
|
|
approval = {
|
|
"proposal_id": approved["id"],
|
|
"proposal_type": approved["proposal_type"],
|
|
"payload": approved["payload"],
|
|
"reviewed_by_handle": approved["reviewed_by_handle"],
|
|
"reviewed_by_agent_id": approved["reviewed_by_agent_id"],
|
|
"reviewed_by_db_role": "kb_review",
|
|
"reviewed_at": approved["reviewed_at"],
|
|
"review_note": approved["review_note"],
|
|
}
|
|
|
|
proposal = _applied_envelope(applied)
|
|
apply_payload = proposal["payload"]["apply_payload"]
|
|
rows = genesis_rebuild.replay_receipt._expected_v3_semantic_rows(proposal, apply_payload)
|
|
created_at = "2026-07-18T10:01:30.000000Z"
|
|
for table, table_rows in rows.items():
|
|
for row in table_rows:
|
|
if table == "claims":
|
|
row.update(created_at=created_at, updated_at=created_at)
|
|
elif table in {
|
|
"sources",
|
|
"claim_evidence",
|
|
"claim_edges",
|
|
"claim_evidence_assessments",
|
|
}:
|
|
row["created_at"] = created_at
|
|
apply_sql = genesis_rebuild.apply_engine.build_apply_sql(proposal, applied["applied_by_handle"])
|
|
receipt = genesis_rebuild.replay_receipt.build_replay_receipt(
|
|
proposal,
|
|
rows,
|
|
apply_sql_sha256=genesis_rebuild.replay_receipt.sha256_text(apply_sql),
|
|
apply_sql_source="exact_executed_sql",
|
|
exported_at_utc="2026-07-18T10:03:00.000000Z",
|
|
)
|
|
return {
|
|
"artifact": genesis_rebuild.MATERIAL_ARTIFACT,
|
|
"contract_version": genesis_rebuild.MATERIAL_CONTRACT_VERSION,
|
|
"sequence": sequence,
|
|
"approved_proposal": approved,
|
|
"approval_snapshot": approval,
|
|
"applied_proposal": applied,
|
|
"replay_receipt": receipt,
|
|
}
|
|
|
|
|
|
def _v2_material() -> dict:
|
|
proposal_id = "70000000-0000-4000-8000-000000000001"
|
|
claim_a = "70000000-0000-4000-8000-000000000002"
|
|
claim_b = "70000000-0000-4000-8000-000000000003"
|
|
payload = {"apply_payload": {"from_claim": claim_a, "to_claim": claim_b, "edge_type": "supports", "weight": 0.8}}
|
|
approved = {
|
|
"id": proposal_id,
|
|
"proposal_type": "add_edge",
|
|
"status": "approved",
|
|
"proposed_by_handle": "fixture",
|
|
"proposed_by_agent_id": REVIEWER_ID,
|
|
"channel": "fixture",
|
|
"source_ref": "private://fixture",
|
|
"rationale": PRIVATE_MARKER,
|
|
"payload": payload,
|
|
"reviewed_by_handle": "m3taversal",
|
|
"reviewed_by_agent_id": REVIEWER_ID,
|
|
"reviewed_at": "2026-07-18T10:01:00.000000Z",
|
|
"review_note": "Reviewed exact V2 fixture.",
|
|
"applied_by_handle": None,
|
|
"applied_by_agent_id": None,
|
|
"applied_at": None,
|
|
"created_at": "2026-07-18T10:00:30.000000Z",
|
|
"updated_at": "2026-07-18T10:01:00.000000Z",
|
|
}
|
|
applied = {
|
|
**approved,
|
|
"status": "applied",
|
|
"applied_by_handle": "kb-apply",
|
|
"applied_by_agent_id": SERVICE_ID,
|
|
"applied_at": "2026-07-18T10:02:00.000000Z",
|
|
"updated_at": "2026-07-18T10:02:00.000001Z",
|
|
}
|
|
approval = {
|
|
"proposal_id": proposal_id,
|
|
"proposal_type": "add_edge",
|
|
"payload": payload,
|
|
"reviewed_by_handle": "m3taversal",
|
|
"reviewed_by_agent_id": REVIEWER_ID,
|
|
"reviewed_by_db_role": "kb_review",
|
|
"reviewed_at": approved["reviewed_at"],
|
|
"review_note": approved["review_note"],
|
|
}
|
|
receipt = genesis_rebuild.replay_receipt.build_replay_receipt(
|
|
_applied_envelope(applied),
|
|
{
|
|
"claim_edges": [
|
|
{
|
|
"id": "70000000-0000-4000-8000-000000000004",
|
|
"from_claim": claim_a,
|
|
"to_claim": claim_b,
|
|
"edge_type": "supports",
|
|
"weight": 0.8,
|
|
"created_by": None,
|
|
"created_at": "2026-07-18T10:01:30.000000Z",
|
|
}
|
|
]
|
|
},
|
|
apply_sql_sha256=genesis_rebuild.replay_receipt.sha256_text("exact V2 fixture SQL"),
|
|
apply_sql_source="exact_executed_sql",
|
|
exported_at_utc="2026-07-18T10:03:00.000000Z",
|
|
)
|
|
return {
|
|
"artifact": genesis_rebuild.MATERIAL_ARTIFACT,
|
|
"contract_version": genesis_rebuild.MATERIAL_CONTRACT_VERSION,
|
|
"sequence": 1,
|
|
"approved_proposal": approved,
|
|
"approval_snapshot": approval,
|
|
"applied_proposal": applied,
|
|
"replay_receipt": receipt,
|
|
}
|
|
|
|
|
|
def _write_json(path: Path, value: object, *, private: bool = False) -> Path:
|
|
path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
if private:
|
|
path.chmod(0o600)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def bundle_inputs(tmp_path: Path) -> dict[str, object]:
|
|
dump = tmp_path / "genesis.dump"
|
|
dump.write_bytes(b"PGDMPfixture")
|
|
genesis_manifest = _write_manifest(tmp_path / "genesis-manifest.jsonl")
|
|
final_manifest = _write_manifest(tmp_path / "final-manifest.jsonl")
|
|
materials = [
|
|
_write_json(tmp_path / "material-0001.json", _v3_material(tmp_path, sequence=1, salt="alpha"), private=True),
|
|
_write_json(tmp_path / "material-0002.json", _v3_material(tmp_path, sequence=2, salt="beta"), private=True),
|
|
]
|
|
note = tmp_path / "review-note.txt"
|
|
note.write_text(f"Explicitly reviewed all source-backed entries. {PRIVATE_MARKER}\n", encoding="utf-8")
|
|
return {
|
|
"dump": dump,
|
|
"genesis_manifest": genesis_manifest,
|
|
"final_manifest": final_manifest,
|
|
"materials": materials,
|
|
"note": note,
|
|
}
|
|
|
|
|
|
def _build_args(inputs: dict[str, object], ledger: Path, *, admission: Path | None = None) -> list[str]:
|
|
args = [
|
|
"--genesis-dump",
|
|
str(inputs["dump"]),
|
|
"--genesis-manifest",
|
|
str(inputs["genesis_manifest"]),
|
|
"--final-manifest",
|
|
str(inputs["final_manifest"]),
|
|
]
|
|
for material in inputs["materials"]:
|
|
args.extend(("--material", str(material)))
|
|
args.extend(("--output-ledger", str(ledger)))
|
|
if admission is not None:
|
|
args.extend(
|
|
(
|
|
"--emit-admission-manifest",
|
|
"--admission-output",
|
|
str(admission),
|
|
"--reviewed-by",
|
|
"m3taversal",
|
|
"--reviewed-by-agent-id",
|
|
REVIEWER_ID,
|
|
"--reviewed-at",
|
|
"2026-07-18T14:00:00+02:00",
|
|
"--review-note-file",
|
|
str(inputs["note"]),
|
|
"--admission-scope",
|
|
"collective_core",
|
|
"--admission-basis",
|
|
"forgejo_primary_source",
|
|
)
|
|
)
|
|
return args
|
|
|
|
|
|
def _verify_args(inputs: dict[str, object], ledger: Path, admission: Path | None = None) -> list[str]:
|
|
args = [
|
|
"--genesis-dump",
|
|
str(inputs["dump"]),
|
|
"--genesis-manifest",
|
|
str(inputs["genesis_manifest"]),
|
|
"--verify-ledger",
|
|
str(ledger),
|
|
"--ledger-sha256",
|
|
genesis_rebuild.canonical_rebuild.sha256_file(ledger),
|
|
]
|
|
if admission is not None:
|
|
args.extend(
|
|
(
|
|
"--admission-manifest",
|
|
str(admission),
|
|
"--admission-manifest-sha256",
|
|
genesis_rebuild.canonical_rebuild.sha256_file(admission),
|
|
)
|
|
)
|
|
return args
|
|
|
|
|
|
def test_deterministic_build_and_verify_every_pin_with_no_writes(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
first_ledger = tmp_path / "ledger-first.json"
|
|
second_ledger = tmp_path / "ledger-second.json"
|
|
first_admission = tmp_path / "admission-first.json"
|
|
second_admission = tmp_path / "admission-second.json"
|
|
|
|
assert builder.main(_build_args(bundle_inputs, first_ledger, admission=first_admission)) == 0
|
|
first_output = capsys.readouterr()
|
|
assert builder.main(_build_args(bundle_inputs, second_ledger, admission=second_admission)) == 0
|
|
second_output = capsys.readouterr()
|
|
|
|
assert first_ledger.read_bytes() == second_ledger.read_bytes()
|
|
assert first_admission.read_bytes() == second_admission.read_bytes()
|
|
assert PRIVATE_MARKER not in first_output.out + first_output.err + second_output.out + second_output.err
|
|
before = {path: (path.read_bytes(), path.stat().st_mtime_ns) for path in (first_ledger, first_admission)}
|
|
|
|
assert builder.main(_verify_args(bundle_inputs, first_ledger, first_admission)) == 0
|
|
verification = json.loads(capsys.readouterr().out)
|
|
|
|
assert verification["action"] == "verify"
|
|
assert verification["entry_count"] == 2
|
|
assert verification["admission"]["review_metadata_provenance"] == builder.REVIEW_METADATA_PROVENANCE
|
|
assert verification["admission"]["cryptographically_authenticated"] is False
|
|
assert before == {path: (path.read_bytes(), path.stat().st_mtime_ns) for path in (first_ledger, first_admission)}
|
|
|
|
|
|
def test_reordered_materials_and_v2_material_are_refused(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
reversed_inputs = {**bundle_inputs, "materials": list(reversed(bundle_inputs["materials"]))}
|
|
assert builder.main(_build_args(reversed_inputs, tmp_path / "reordered.json")) == 1
|
|
assert not (tmp_path / "reordered.json").exists()
|
|
assert json.loads(capsys.readouterr().err)["code"] == "invalid_exported_material"
|
|
|
|
v2_path = _write_json(tmp_path / "v2-material.json", _v2_material(), private=True)
|
|
v2_inputs = {**bundle_inputs, "materials": [v2_path]}
|
|
assert builder.main(_build_args(v2_inputs, tmp_path / "v2-ledger.json")) == 1
|
|
assert not (tmp_path / "v2-ledger.json").exists()
|
|
assert json.loads(capsys.readouterr().err)["code"] == "legacy_material_forbidden"
|
|
|
|
|
|
def test_hash_drift_engine_tamper_and_symlink_material_fail_closed(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
ledger = tmp_path / "ledger.json"
|
|
assert builder.main(_build_args(bundle_inputs, ledger)) == 0
|
|
capsys.readouterr()
|
|
|
|
first_material = bundle_inputs["materials"][0]
|
|
first_material.write_bytes(first_material.read_bytes() + b" ")
|
|
assert builder.main(_verify_args(bundle_inputs, ledger)) == 1
|
|
assert json.loads(capsys.readouterr().err)["code"] == "genesis_preflight_refused"
|
|
|
|
first_material.write_bytes(first_material.read_bytes()[:-1])
|
|
tampered = json.loads(ledger.read_text(encoding="utf-8"))
|
|
tampered["engine"]["apply_engine_sha256"] = "0" * 64
|
|
_write_json(ledger, tampered, private=True)
|
|
assert builder.main(_verify_args(bundle_inputs, ledger)) == 1
|
|
assert json.loads(capsys.readouterr().err)["code"] == "genesis_preflight_refused"
|
|
|
|
symlink = tmp_path / "material-symlink.json"
|
|
symlink.symlink_to(bundle_inputs["materials"][0])
|
|
symlink_inputs = {**bundle_inputs, "materials": [symlink]}
|
|
assert builder.main(_build_args(symlink_inputs, tmp_path / "symlink-ledger.json")) == 1
|
|
assert json.loads(capsys.readouterr().err)["code"] == "unsafe_input"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "expected_code"),
|
|
[
|
|
("missing_explicit_flag", "implicit_admission_forbidden"),
|
|
("missing_reviewer_id", "incomplete_admission_review"),
|
|
("alternate_reviewer", "invalid_reviewer_handle"),
|
|
("noncanonical_reviewer_id", "invalid_reviewer_identity"),
|
|
],
|
|
)
|
|
def test_admission_requires_complete_exact_operator_review(
|
|
bundle_inputs: dict[str, object],
|
|
tmp_path: Path,
|
|
capsys: pytest.CaptureFixture[str],
|
|
mutation: str,
|
|
expected_code: str,
|
|
) -> None:
|
|
ledger = tmp_path / f"{mutation}-ledger.json"
|
|
admission = tmp_path / f"{mutation}-admission.json"
|
|
args = _build_args(bundle_inputs, ledger, admission=admission)
|
|
if mutation == "missing_explicit_flag":
|
|
args.remove("--emit-admission-manifest")
|
|
elif mutation == "missing_reviewer_id":
|
|
index = args.index("--reviewed-by-agent-id")
|
|
del args[index : index + 2]
|
|
elif mutation == "alternate_reviewer":
|
|
args[args.index("--reviewed-by") + 1] = "m3ta"
|
|
else:
|
|
args[args.index("--reviewed-by-agent-id") + 1] = "AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA"
|
|
|
|
assert builder.main(args) == 1
|
|
output = capsys.readouterr()
|
|
assert json.loads(output.err)["code"] == expected_code
|
|
assert PRIVATE_MARKER not in output.out + output.err
|
|
assert not ledger.exists()
|
|
assert not admission.exists()
|
|
|
|
|
|
def test_admission_always_selects_all_entries_and_partial_manifest_is_rejected(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
ledger = tmp_path / "ledger.json"
|
|
admission = tmp_path / "admission.json"
|
|
assert builder.main(_build_args(bundle_inputs, ledger, admission=admission)) == 0
|
|
capsys.readouterr()
|
|
manifest = json.loads(admission.read_text(encoding="utf-8"))
|
|
assert [entry["sequence"] for entry in manifest["entries"]] == [1, 2]
|
|
|
|
manifest["entries"].pop()
|
|
_write_json(admission, manifest, private=True)
|
|
assert builder.main(_verify_args(bundle_inputs, ledger, admission)) == 1
|
|
assert json.loads(capsys.readouterr().err)["code"] == "genesis_preflight_refused"
|
|
|
|
|
|
def test_existing_outputs_are_never_overwritten_and_outputs_are_private(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
existing = tmp_path / "existing-ledger.json"
|
|
existing.write_text("operator-owned\n", encoding="utf-8")
|
|
assert builder.main(_build_args(bundle_inputs, existing)) == 1
|
|
assert existing.read_text(encoding="utf-8") == "operator-owned\n"
|
|
assert json.loads(capsys.readouterr().err)["code"] == "existing_output"
|
|
|
|
ledger = tmp_path / "private-ledger.json"
|
|
admission = tmp_path / "private-admission.json"
|
|
assert builder.main(_build_args(bundle_inputs, ledger, admission=admission)) == 0
|
|
output = capsys.readouterr()
|
|
assert stat.S_IMODE(ledger.stat().st_mode) == 0o600
|
|
assert stat.S_IMODE(admission.stat().st_mode) == 0o600
|
|
assert ledger.stat().st_nlink == admission.stat().st_nlink == 1
|
|
assert PRIVATE_MARKER not in output.out + output.err
|
|
|
|
|
|
def test_material_tamper_is_redacted_and_publishes_nothing(
|
|
bundle_inputs: dict[str, object], tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
material_path = bundle_inputs["materials"][0]
|
|
material = json.loads(material_path.read_text(encoding="utf-8"))
|
|
material["approved_proposal"]["payload"]["private_secret"] = PRIVATE_MARKER
|
|
_write_json(material_path, material, private=True)
|
|
tampered_inputs = {**bundle_inputs, "materials": [material_path]}
|
|
ledger = tmp_path / "tampered-ledger.json"
|
|
|
|
assert builder.main(_build_args(tampered_inputs, ledger)) == 1
|
|
output = capsys.readouterr()
|
|
assert not ledger.exists()
|
|
assert PRIVATE_MARKER not in output.out + output.err
|
|
assert json.loads(output.err)["status"] == "refused"
|