328 lines
12 KiB
Python
328 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import compile_kb_source_packet as compiler # noqa: E402, I001
|
|
|
|
|
|
ARTIFACT_BYTES = b"%PDF-1.7\nsource compiler fixture\x00\xff\n%%EOF\n"
|
|
EXTRACTED_TEXT = (
|
|
"Source packet compiler note.\n\n"
|
|
"A staged proposal remains non-canonical until a human review and guarded apply both succeed.\n\n"
|
|
"Every proposed claim must retain an exact quote from the extracted source text."
|
|
)
|
|
CLAIM_QUOTE = "A staged proposal remains non-canonical until a human review and guarded apply both succeed."
|
|
EVIDENCE_QUOTE = "Every proposed claim must retain an exact quote from the extracted source text."
|
|
|
|
|
|
def _sha256(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def _manifest() -> dict:
|
|
artifact_sha256 = _sha256(ARTIFACT_BYTES)
|
|
text_bytes = EXTRACTED_TEXT.encode("utf-8")
|
|
return {
|
|
"schema": compiler.MANIFEST_SCHEMA,
|
|
"artifact_sha256": artifact_sha256,
|
|
"extracted_text_sha256": _sha256(text_bytes),
|
|
"extractor": {"name": "fixture-text-extractor", "version": "1"},
|
|
"source": {
|
|
"identity": f"document:sha256:{artifact_sha256}",
|
|
"source_key": "source_compiler_note",
|
|
"source_type": "article",
|
|
"title": "Source packet compiler note",
|
|
"locator": f"artifact://sha256/{artifact_sha256}",
|
|
},
|
|
"claims": [
|
|
{
|
|
"claim_key": "review_gate_precedes_canonical_apply",
|
|
"type": "structural",
|
|
"text": "A staged proposal is not canonical knowledge until review and guarded apply succeed.",
|
|
"quote": CLAIM_QUOTE,
|
|
"confidence": 0.99,
|
|
"tags": ["review", "provenance"],
|
|
"evidence": [{"quote": EVIDENCE_QUOTE, "role": "grounds"}],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _write_inputs(tmp_path: Path, manifest: dict | None = None, prefix: str = "input") -> tuple[Path, Path, Path]:
|
|
tmp_path.mkdir(parents=True, exist_ok=True)
|
|
artifact = tmp_path / f"{prefix}.bin"
|
|
text = tmp_path / f"{prefix}.txt"
|
|
manifest_path = tmp_path / f"{prefix}.json"
|
|
artifact.write_bytes(ARTIFACT_BYTES)
|
|
text.write_text(EXTRACTED_TEXT, encoding="utf-8")
|
|
manifest_path.write_text(json.dumps(manifest or _manifest(), indent=2), encoding="utf-8")
|
|
return artifact, text, manifest_path
|
|
|
|
|
|
def _compile(tmp_path: Path, manifest: dict | None = None) -> dict:
|
|
paths = _write_inputs(tmp_path, manifest)
|
|
return compiler.compile_source_packet(*paths)
|
|
|
|
|
|
def test_compiles_deterministic_hash_bound_pending_review_bundle_without_db_write(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
artifact, text, manifest_path = _write_inputs(tmp_path, prefix="first")
|
|
|
|
def unexpected_db_call(*_args, **_kwargs):
|
|
raise AssertionError("compiler attempted a database write")
|
|
|
|
monkeypatch.setattr(compiler.stage, "stage_normalized_child", unexpected_db_call)
|
|
monkeypatch.setattr(compiler.stage.bound, "_psql_json", unexpected_db_call)
|
|
first = compiler.compile_source_packet(artifact, text, manifest_path)
|
|
|
|
copied = tmp_path / "copied"
|
|
copied.mkdir()
|
|
copied_paths = _write_inputs(copied, prefix="different-path")
|
|
second = compiler.compile_source_packet(*copied_paths)
|
|
|
|
assert first == second
|
|
assert first["status"] == "pending_review"
|
|
assert first["build_only"] is True
|
|
assert first["database_write_performed"] is False
|
|
assert first["canonical_apply_performed"] is False
|
|
assert first["stage_preview"]["executed"] is False
|
|
assert len(first["stage_preview"]["stage_sql_sha256"]) == 64
|
|
|
|
parent = first["parent_proposal"]
|
|
child = first["strict_child_proposal"]
|
|
apply_payload = child["payload"]["apply_payload"]
|
|
assert parent["status"] == child["status"] == "pending_review"
|
|
assert len(parent["payload"]["source_candidates"]) == 1
|
|
assert parent["payload"]["source_candidates"][0]["source_key"] == "source_compiler_note"
|
|
assert child["proposal_type"] == "approve_claim"
|
|
assert apply_payload["contract_version"] == 2
|
|
assert apply_payload["source_proposal_id"] == parent["id"]
|
|
assert len(apply_payload["claims"]) == 1
|
|
assert len(apply_payload["sources"]) == 2
|
|
assert len(apply_payload["evidence"]) == 2
|
|
assert first["hashes"]["artifact_sha256"] in {row["hash"] for row in apply_payload["sources"]}
|
|
assert first["compiler_contract"]["source_from_chat_labels"] is False
|
|
|
|
without_bundle_hash = copy.deepcopy(first)
|
|
bundle_hash = without_bundle_hash["hashes"].pop("bundle_content_sha256")
|
|
assert bundle_hash == compiler.canonical_sha256(without_bundle_hash)
|
|
|
|
|
|
def test_cli_emits_structured_json_to_stdout_and_optional_output(tmp_path: Path, capsys: pytest.CaptureFixture) -> None:
|
|
artifact, text, manifest_path = _write_inputs(tmp_path)
|
|
output = tmp_path / "compiled" / "packet.json"
|
|
|
|
status = compiler.main(
|
|
[
|
|
"--artifact",
|
|
str(artifact),
|
|
"--text",
|
|
str(text),
|
|
"--manifest",
|
|
str(manifest_path),
|
|
"--output",
|
|
str(output),
|
|
]
|
|
)
|
|
|
|
stdout = capsys.readouterr().out
|
|
assert status == 0
|
|
assert output.read_text(encoding="utf-8") == stdout
|
|
assert json.loads(stdout)["status"] == "pending_review"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "expected"),
|
|
[
|
|
("artifact_sha256", "artifact SHA-256 does not match"),
|
|
("extracted_text_sha256", "extracted text SHA-256 does not match"),
|
|
],
|
|
)
|
|
def test_rejects_hash_mismatches(tmp_path: Path, field: str, expected: str) -> None:
|
|
manifest = _manifest()
|
|
manifest[field] = "0" * 64
|
|
|
|
with pytest.raises(compiler.CompilerError, match=expected):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("target", "expected"),
|
|
[
|
|
("claim", "claims\\[0\\]\\.quote is not an exact substring"),
|
|
("evidence", "evidence\\[0\\]\\.quote is not an exact substring"),
|
|
],
|
|
)
|
|
def test_rejects_hallucinated_claim_or_evidence_quote(tmp_path: Path, target: str, expected: str) -> None:
|
|
manifest = _manifest()
|
|
if target == "claim":
|
|
manifest["claims"][0]["quote"] = "This quote was invented by the extractor."
|
|
else:
|
|
manifest["claims"][0]["evidence"][0]["quote"] = "This evidence was invented by the extractor."
|
|
|
|
with pytest.raises(compiler.CompilerError, match=expected):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
def test_rejects_duplicate_claim_and_derived_source_keys(tmp_path: Path) -> None:
|
|
duplicate_claim = _manifest()
|
|
duplicate_claim["claims"].append(copy.deepcopy(duplicate_claim["claims"][0]))
|
|
with pytest.raises(compiler.CompilerError, match="duplicate claim_key"):
|
|
_compile(tmp_path / "claim", duplicate_claim)
|
|
|
|
duplicate_source = _manifest()
|
|
duplicate_source["source"]["source_key"] = "review_gate_precedes_canonical_apply__proposal_body"
|
|
with pytest.raises(compiler.CompilerError, match="duplicate source_key after normalization"):
|
|
_compile(tmp_path / "source", duplicate_source)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("target", "expected"),
|
|
[
|
|
("claim", "claims\\[0\\]\\.type must be one of"),
|
|
("source", "source\\.source_type must be one of"),
|
|
("evidence", "evidence\\[0\\]\\.role must be one of"),
|
|
],
|
|
)
|
|
def test_rejects_types_outside_current_taxonomies(tmp_path: Path, target: str, expected: str) -> None:
|
|
manifest = _manifest()
|
|
if target == "claim":
|
|
manifest["claims"][0]["type"] = "fact"
|
|
elif target == "source":
|
|
manifest["source"]["source_type"] = "document"
|
|
else:
|
|
manifest["claims"][0]["evidence"][0]["role"] = "proves"
|
|
|
|
with pytest.raises(compiler.CompilerError, match=expected):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["identity", "locator"])
|
|
def test_rejects_missing_or_unstable_source_provenance(tmp_path: Path, field: str) -> None:
|
|
manifest = _manifest()
|
|
manifest["source"][field] = "message from Alice"
|
|
|
|
with pytest.raises(compiler.CompilerError, match="stable URI-like reference"):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("reference", "expected"),
|
|
[
|
|
("javascript:alert(1)", "unsafe URI scheme"),
|
|
("https:/missing-host", "must include a host"),
|
|
("https://user:password@example.com/source", "must not embed credentials"),
|
|
("telegram:Alice", "stable chat or message identifier"),
|
|
],
|
|
)
|
|
def test_rejects_unsafe_or_non_specific_source_references(tmp_path: Path, reference: str, expected: str) -> None:
|
|
manifest = _manifest()
|
|
manifest["source"]["locator"] = reference
|
|
|
|
with pytest.raises(compiler.CompilerError, match=expected):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
def test_rejects_tags_with_ambiguous_whitespace(tmp_path: Path) -> None:
|
|
manifest = _manifest()
|
|
manifest["claims"][0]["tags"] = [" review"]
|
|
|
|
with pytest.raises(compiler.CompilerError, match="tags must be a list of non-empty strings"):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
def test_rejects_path_gaps_invalid_utf8_and_output_overwrite(tmp_path: Path, capsys: pytest.CaptureFixture) -> None:
|
|
artifact, text, manifest_path = _write_inputs(tmp_path)
|
|
with pytest.raises(compiler.CompilerError, match="artifact path is not a regular file"):
|
|
compiler.compile_source_packet(tmp_path / "missing.bin", text, manifest_path)
|
|
|
|
text.write_bytes(b"\xff\xfe")
|
|
with pytest.raises(compiler.CompilerError, match="valid UTF-8"):
|
|
compiler.compile_source_packet(artifact, text, manifest_path)
|
|
|
|
status = compiler.main(
|
|
[
|
|
"--artifact",
|
|
str(artifact),
|
|
"--text",
|
|
str(text),
|
|
"--manifest",
|
|
str(manifest_path),
|
|
"--output",
|
|
str(artifact),
|
|
]
|
|
)
|
|
rejection = json.loads(capsys.readouterr().out)
|
|
assert status == 2
|
|
assert rejection["status"] == "rejected"
|
|
assert rejection["database_write_performed"] is False
|
|
assert "must not overwrite" in rejection["error"]
|
|
|
|
|
|
def test_cli_reports_output_write_failure_as_structured_rejection(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
artifact, text, manifest_path = _write_inputs(tmp_path)
|
|
output = tmp_path / "packet.json"
|
|
original_write_text = Path.write_text
|
|
|
|
def guarded_write_text(path: Path, data: str, *args, **kwargs):
|
|
if path == output:
|
|
raise PermissionError("fixture output denied")
|
|
return original_write_text(path, data, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Path, "write_text", guarded_write_text)
|
|
status = compiler.main(
|
|
[
|
|
"--artifact",
|
|
str(artifact),
|
|
"--text",
|
|
str(text),
|
|
"--manifest",
|
|
str(manifest_path),
|
|
"--output",
|
|
str(output),
|
|
]
|
|
)
|
|
|
|
rejection = json.loads(capsys.readouterr().out)
|
|
assert status == 2
|
|
assert rejection["status"] == "rejected"
|
|
assert rejection["database_write_performed"] is False
|
|
assert "could not write output" in rejection["error"]
|
|
|
|
|
|
def test_rejects_attempt_to_invent_source_from_chat_label(tmp_path: Path) -> None:
|
|
manifest = _manifest()
|
|
manifest["claims"][0]["evidence"][0]["source_label"] = "Alice in Telegram"
|
|
|
|
with pytest.raises(compiler.CompilerError, match="unsupported keys: source_label"):
|
|
_compile(tmp_path, manifest)
|
|
|
|
|
|
def test_rejects_quotes_that_cannot_fit_the_reviewable_canonical_excerpt(tmp_path: Path) -> None:
|
|
long_quote = "x" * 8100
|
|
text = f"prefix {long_quote} suffix"
|
|
artifact = tmp_path / "long.bin"
|
|
text_path = tmp_path / "long.txt"
|
|
manifest_path = tmp_path / "long.json"
|
|
artifact.write_bytes(ARTIFACT_BYTES)
|
|
text_path.write_text(text, encoding="utf-8")
|
|
manifest = _manifest()
|
|
manifest["extracted_text_sha256"] = _sha256(text.encode("utf-8"))
|
|
manifest["claims"][0]["quote"] = long_quote
|
|
manifest["claims"][0]["evidence"] = [{"quote": long_quote, "role": "grounds"}]
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
|
with pytest.raises(compiler.CompilerError, match=r"truncated|excerpt capacity"):
|
|
compiler.compile_source_packet(artifact, text_path, manifest_path)
|