199 lines
7.5 KiB
Python
199 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import prepare_kb_source_manifest as preparer # noqa: E402
|
|
|
|
SOURCE_TEXT = (
|
|
"Source preparation note.\n\n"
|
|
"A staged proposal remains non-canonical until review and guarded apply both succeed.\n\n"
|
|
"Every prepared claim must remain bound to an exact source quote.\n"
|
|
)
|
|
CLAIM_QUOTE = "Every prepared claim must remain bound to an exact source quote."
|
|
DUPLICATE_QUOTE = "A staged proposal remains non-canonical until review and guarded apply both succeed."
|
|
CANDIDATE_ID = "11111111-1111-4111-8111-111111111111"
|
|
|
|
|
|
def _args(tmp_path: Path, artifact: Path, context: Path) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
artifact=artifact,
|
|
text=None,
|
|
kb_context=context,
|
|
identity="document:source-preparation-test-v1",
|
|
source_key="source_preparation_test_v1",
|
|
source_type="article",
|
|
title="Source preparation test",
|
|
locator="artifact://source-preparation/test-v1",
|
|
output_dir=tmp_path / "private-output",
|
|
key_file=tmp_path / "openrouter-key",
|
|
model="fixture/model",
|
|
openrouter_url="https://openrouter.invalid/v1/chat/completions",
|
|
timeout_seconds=30,
|
|
)
|
|
|
|
|
|
def _inputs(tmp_path: Path) -> tuple[Path, Path]:
|
|
artifact = tmp_path / "source.md"
|
|
artifact.write_text(SOURCE_TEXT, encoding="utf-8")
|
|
context = tmp_path / "context.json"
|
|
context.write_text(
|
|
json.dumps(
|
|
{
|
|
"database_search_query": "staged proposal exact source quote",
|
|
"candidate_claims": [
|
|
{
|
|
"id": CANDIDATE_ID,
|
|
"text": "A staged proposal is not canonical before guarded apply.",
|
|
"score": 3,
|
|
"evidence_count": 2,
|
|
"edge_count": 1,
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "openrouter-key").write_text("fixture-secret", encoding="utf-8")
|
|
return artifact, context
|
|
|
|
|
|
def _model_output(*, claims: bool = True, bad_reference: bool = False) -> str:
|
|
claim_rows = []
|
|
if claims:
|
|
quote_ref = "L99999" if bad_reference else "L00005"
|
|
claim_rows = [
|
|
{
|
|
"claim_key": "exact_quotes_bind_prepared_claims",
|
|
"type": "structural",
|
|
"text": "Prepared claims remain auditable when their evidence is bound to exact source quotes.",
|
|
"quote_ref": quote_ref,
|
|
"confidence": 0.7,
|
|
"tags": ["provenance", "review"],
|
|
"evidence": [{"quote_ref": quote_ref, "role": "grounds"}],
|
|
}
|
|
]
|
|
return json.dumps(
|
|
{
|
|
"schema": preparer.MODEL_OUTPUT_SCHEMA,
|
|
"claims": claim_rows,
|
|
"duplicates": [
|
|
{
|
|
"claim_id": CANDIDATE_ID,
|
|
"source_quote_ref": "L00003",
|
|
"reason": "The canonical candidate already captures the review-before-apply argument.",
|
|
}
|
|
],
|
|
"extraction_notes": "One duplicate argument and one novel exact-quote provenance claim.",
|
|
}
|
|
)
|
|
|
|
|
|
def test_prepare_builds_private_compiler_validated_manifest_without_db_write(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
artifact, context = _inputs(tmp_path)
|
|
monkeypatch.setattr(
|
|
preparer,
|
|
"call_openrouter",
|
|
lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}),
|
|
)
|
|
|
|
receipt = preparer.prepare(_args(tmp_path, artifact, context))
|
|
|
|
assert receipt["status"] == "prepared"
|
|
assert receipt["database_write_performed"] is False
|
|
assert receipt["canonical_apply_performed"] is False
|
|
assert receipt["retrieval"]["canonical_candidate_ids"] == [CANDIDATE_ID]
|
|
assert receipt["extraction"]["claim_count"] == 1
|
|
assert receipt["compiler"]["validated"] is True
|
|
assert receipt["compiler"]["strict_child_proposal_id"]
|
|
output_dir = Path(receipt["outputs"]["output_dir"])
|
|
manifest = json.loads(Path(receipt["outputs"]["manifest"]).read_text(encoding="utf-8"))
|
|
assert manifest["schema"] == preparer.compiler.MANIFEST_SCHEMA_V2
|
|
assert manifest["dedupe"]["duplicates"][0]["claim_id"] == CANDIDATE_ID
|
|
assert manifest["claims"][0]["quote"] == CLAIM_QUOTE
|
|
assert manifest["dedupe"]["duplicates"][0]["source_quote"] == DUPLICATE_QUOTE
|
|
assert receipt["extraction"]["selected_source_references"] == [
|
|
{"kind": "claim", "reference": "L00005"},
|
|
{"kind": "evidence", "reference": "L00005"},
|
|
{"kind": "duplicate", "reference": "L00003"},
|
|
]
|
|
assert stat.S_IMODE(output_dir.stat().st_mode) == 0o700
|
|
for path in output_dir.iterdir():
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
|
|
|
|
def test_prepare_retries_once_when_model_invents_a_source_reference(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
artifact, context = _inputs(tmp_path)
|
|
outputs = iter((_model_output(bad_reference=True), _model_output()))
|
|
prompts: list[str] = []
|
|
|
|
def fake_call(prompt: str, **_kwargs):
|
|
prompts.append(prompt)
|
|
return next(outputs), {}
|
|
|
|
monkeypatch.setattr(preparer, "call_openrouter", fake_call)
|
|
|
|
receipt = preparer.prepare(_args(tmp_path, artifact, context))
|
|
|
|
assert receipt["status"] == "prepared"
|
|
assert len(receipt["extraction"]["attempts"]) == 2
|
|
assert "previous extraction failed deterministic validation" in prompts[1]
|
|
assert "must select one of the bracketed SOURCE line IDs" in prompts[1]
|
|
|
|
|
|
def test_source_fragments_preserve_exact_unicode_and_punctuation() -> None:
|
|
text = "Heading\n\nIdentity pins use an exact version - never a floating alias.\nCurly: \u201creviewed\u201d.\n"
|
|
|
|
fragments = preparer.build_source_fragments(text)
|
|
|
|
assert fragments == {
|
|
"L00001": "Heading",
|
|
"L00003": "Identity pins use an exact version - never a floating alias.",
|
|
"L00004": "Curly: \u201creviewed\u201d.",
|
|
}
|
|
assert all(value in text for value in fragments.values())
|
|
|
|
|
|
def test_prepare_returns_no_novel_claims_without_manifest_or_stageable_packet(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
artifact, context = _inputs(tmp_path)
|
|
monkeypatch.setattr(preparer, "call_openrouter", lambda *_args, **_kwargs: (_model_output(claims=False), {}))
|
|
|
|
receipt = preparer.prepare(_args(tmp_path, artifact, context))
|
|
|
|
assert receipt["status"] == "no_novel_claims"
|
|
assert receipt["outputs"]["manifest"] is None
|
|
assert receipt["compiler"]["validated"] is False
|
|
assert "do not stage an empty proposal" in receipt["next_action"]
|
|
|
|
|
|
def test_binary_artifact_requires_explicit_utf8_extraction(tmp_path: Path) -> None:
|
|
artifact = tmp_path / "source.pdf"
|
|
artifact.write_bytes(b"%PDF-1.7\x00binary")
|
|
|
|
with pytest.raises(preparer.PreparationError, match=r"supply --text for \.pdf"):
|
|
preparer.extract_utf8_text(artifact, None)
|
|
|
|
|
|
def test_openrouter_key_cannot_be_redirected_to_an_unapproved_endpoint(tmp_path: Path) -> None:
|
|
with pytest.raises(preparer.PreparationError, match=r"must equal https://openrouter\.ai"):
|
|
preparer.call_openrouter(
|
|
"prompt",
|
|
model="fixture/model",
|
|
key_file=tmp_path / "private-file",
|
|
url="https://example.invalid/collect",
|
|
timeout_seconds=1,
|
|
)
|