128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT_PATH = REPO_ROOT / "scripts" / "prepare_kb_context_from_canonical_dump.py"
|
|
SPEC = importlib.util.spec_from_file_location("prepare_kb_context_from_canonical_dump", SCRIPT_PATH)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
context_builder = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = context_builder
|
|
SPEC.loader.exec_module(context_builder)
|
|
|
|
CLAIM_A = "11111111-1111-4111-8111-111111111111"
|
|
CLAIM_B = "22222222-2222-4222-8222-222222222222"
|
|
SOURCE_ID = "33333333-3333-4333-8333-333333333333"
|
|
|
|
|
|
def _copy(table: str, columns: list[str], rows: list[list[str]]) -> str:
|
|
rendered = [f"COPY public.{table} ({', '.join(columns)}) FROM stdin;"]
|
|
rendered.extend("\t".join(row) for row in rows)
|
|
rendered.append(r"\.")
|
|
return "\n".join(rendered) + "\n"
|
|
|
|
|
|
def test_build_context_matches_readonly_claim_ranking_and_counts(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
dump = tmp_path / "canonical.dump"
|
|
dump.write_bytes(b"fixture dump")
|
|
artifact = tmp_path / "segment.txt"
|
|
artifact.write_text("Exact source quotes keep staged proposals reviewable.\n", encoding="utf-8")
|
|
sql = {
|
|
"claims": _copy(
|
|
"claims",
|
|
[
|
|
"id",
|
|
"type",
|
|
"text",
|
|
"status",
|
|
"confidence",
|
|
"tags",
|
|
"created_by",
|
|
"superseded_by",
|
|
"created_at",
|
|
"updated_at",
|
|
],
|
|
[
|
|
[
|
|
CLAIM_A,
|
|
"structural",
|
|
"Exact source quotes keep proposals reviewable.",
|
|
"open",
|
|
"0.8",
|
|
"{provenance}",
|
|
r"\N",
|
|
r"\N",
|
|
"2026-01-01",
|
|
"2026-01-01",
|
|
],
|
|
[
|
|
CLAIM_B,
|
|
"structural",
|
|
"Staged proposals need guarded apply.",
|
|
"open",
|
|
"0.9",
|
|
"{review}",
|
|
r"\N",
|
|
r"\N",
|
|
"2026-01-01",
|
|
"2026-01-01",
|
|
],
|
|
],
|
|
),
|
|
"claim_evidence": _copy(
|
|
"claim_evidence",
|
|
["claim_id", "source_id", "role", "weight", "created_by", "created_at"],
|
|
[[CLAIM_A, SOURCE_ID, "grounds", "1", r"\N", "2026-01-01"]],
|
|
),
|
|
"claim_edges": _copy(
|
|
"claim_edges",
|
|
["id", "from_claim", "to_claim", "edge_type", "weight", "created_by", "created_at"],
|
|
[["44444444-4444-4444-8444-444444444444", CLAIM_A, CLAIM_B, "supports", "1", r"\N", "2026-01-01"]],
|
|
),
|
|
}
|
|
monkeypatch.setattr(context_builder, "restore_table", lambda _dump, table, _image: sql[table])
|
|
|
|
context, receipt = context_builder.build_context(
|
|
dump_path=dump,
|
|
artifact_path=artifact,
|
|
title="Telegram segment",
|
|
limit=12,
|
|
docker_image="fixture-postgres",
|
|
repo_root=REPO_ROOT,
|
|
)
|
|
|
|
assert context["candidate_claims"][0]["id"] == CLAIM_A
|
|
assert context["candidate_claims"][0]["evidence_count"] == 1
|
|
assert context["candidate_claims"][0]["edge_count"] == 1
|
|
assert receipt["database_write_performed"] is False
|
|
assert receipt["model_call_performed"] is False
|
|
assert receipt["runtime"]["network_disabled"] is True
|
|
assert receipt["runtime"]["pg_restore_output_retained"] is False
|
|
|
|
|
|
def test_copy_parser_preserves_escaped_source_text() -> None:
|
|
sql = _copy("claims", ["id", "text"], [[CLAIM_A, r"line one\nline two\tvalue\\tail"]])
|
|
|
|
rows = context_builder.parse_copy_table(sql, "claims")
|
|
|
|
assert rows == [{"id": CLAIM_A, "text": "line one\nline two\tvalue\\tail"}]
|
|
|
|
|
|
def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None:
|
|
output = tmp_path / "private" / "context.json"
|
|
|
|
context_builder._private_json(output, {"safe": True})
|
|
|
|
assert json.loads(output.read_text(encoding="utf-8")) == {"safe": True}
|
|
assert stat.S_IMODE(output.parent.stat().st_mode) == 0o700
|
|
assert stat.S_IMODE(output.stat().st_mode) == 0o600
|
|
with pytest.raises(context_builder.ContextError, match="refusing to overwrite"):
|
|
context_builder._private_json(output, {"safe": False})
|