177 lines
7.6 KiB
Python
177 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from groundrecall.ingest import run_groundrecall_import
|
|
from groundrecall.promotion import PromotionGateError, promote_import_to_store
|
|
from groundrecall.store import GroundRecallStore
|
|
|
|
|
|
def test_groundrecall_promotion_writes_canonical_objects(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "channel-capacity.md").write_text(
|
|
"# Channel Capacity\n\n"
|
|
"- Reliable rate upper bound for a noisy channel.\n\n"
|
|
"See also [[Shannon Entropy]].\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="promote-test")
|
|
review_path = result.out_dir / "review_session.json"
|
|
review_payload = json.loads(review_path.read_text(encoding="utf-8"))
|
|
for concept in review_payload["draft_pack"]["concepts"]:
|
|
concept["status"] = "trusted"
|
|
review_path.write_text(json.dumps(review_payload, indent=2), encoding="utf-8")
|
|
|
|
store_dir = tmp_path / "groundrecall-store"
|
|
payload = promote_import_to_store(result.out_dir, store_dir, reviewer="R")
|
|
|
|
store = GroundRecallStore(store_dir)
|
|
concepts = store.list_concepts()
|
|
claims = store.list_claims()
|
|
relations = store.list_relations()
|
|
promotions = store.list_promotions()
|
|
snapshots = store.list_snapshots()
|
|
|
|
assert payload["promoted_concept_count"] >= 1
|
|
assert payload["promoted_claim_count"] >= 1
|
|
assert len(concepts) >= 2
|
|
assert any(item.current_status == "promoted" for item in concepts)
|
|
assert any(item.current_status == "promoted" for item in claims)
|
|
assert len(relations) >= 1
|
|
assert len(promotions) == 1
|
|
assert promotions[0].reviewer == "R"
|
|
assert len(snapshots) == 1
|
|
assert snapshots[0].metadata["source_import_id"] == "promote-test"
|
|
|
|
|
|
def test_groundrecall_promotion_respects_rejected_review_status(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "solo.md").write_text(
|
|
"# Solo Concept\n\n- A solitary claim.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="reject-test")
|
|
review_path = result.out_dir / "review_session.json"
|
|
review_payload = json.loads(review_path.read_text(encoding="utf-8"))
|
|
review_payload["draft_pack"]["concepts"][0]["status"] = "rejected"
|
|
review_path.write_text(json.dumps(review_payload, indent=2), encoding="utf-8")
|
|
|
|
store_dir = tmp_path / "groundrecall-store"
|
|
promote_import_to_store(result.out_dir, store_dir, reviewer="R")
|
|
|
|
store = GroundRecallStore(store_dir)
|
|
assert store.list_concepts()[0].current_status == "rejected"
|
|
assert store.list_claims()[0].current_status == "rejected"
|
|
|
|
|
|
def test_groundrecall_promotion_preserves_contradiction_and_supersession_links(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "notes.md").write_text(
|
|
"# Notes\n\n"
|
|
"- [claim_id: base] Channel capacity bounds reliable communication rate.\n"
|
|
"- [claim_id: revised] [supersedes: base] Channel capacity bounds reliable communication rate for a specified channel model.\n"
|
|
"- [claim_id: dissent] [contradicts: revised] Channel capacity has no stable interpretation.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="graph-test")
|
|
review_path = result.out_dir / "review_session.json"
|
|
review_payload = json.loads(review_path.read_text(encoding="utf-8"))
|
|
for concept in review_payload["draft_pack"]["concepts"]:
|
|
concept["status"] = "trusted"
|
|
review_path.write_text(json.dumps(review_payload, indent=2), encoding="utf-8")
|
|
|
|
store_dir = tmp_path / "groundrecall-store"
|
|
promote_import_to_store(result.out_dir, store_dir, reviewer="R")
|
|
|
|
store = GroundRecallStore(store_dir)
|
|
claims = {item.claim_id: item for item in store.list_claims()}
|
|
assert claims["clm_revised"].supersedes_claim_ids == ["clm_base"]
|
|
assert claims["clm_dissent"].contradicts_claim_ids == ["clm_revised"]
|
|
|
|
|
|
def test_groundrecall_promotion_preserves_queue_graph_rationale(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "a.md").write_text("# A\n\nSee also [[B]].\n", encoding="utf-8")
|
|
(root / "wiki" / "b.md").write_text("# B\n\nSee also [[C]].\n", encoding="utf-8")
|
|
(root / "wiki" / "c.md").write_text("# C\n", encoding="utf-8")
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="promotion-queue-test")
|
|
review_path = result.out_dir / "review_session.json"
|
|
review_payload = json.loads(review_path.read_text(encoding="utf-8"))
|
|
for concept in review_payload["draft_pack"]["concepts"]:
|
|
concept["status"] = "trusted"
|
|
review_path.write_text(json.dumps(review_payload, indent=2), encoding="utf-8")
|
|
|
|
store_dir = tmp_path / "groundrecall-store"
|
|
promote_import_to_store(result.out_dir, store_dir, reviewer="R")
|
|
|
|
store = GroundRecallStore(store_dir)
|
|
review_candidates = {item.candidate_id: item for item in store.list_review_candidates()}
|
|
bridge_candidate = review_candidates["concept::b"]
|
|
|
|
assert bridge_candidate.triage_lane == "conflict_resolution"
|
|
assert "bridge_concept" in bridge_candidate.finding_codes
|
|
assert "lane=conflict_resolution" in bridge_candidate.rationale
|
|
assert "graph=bridge_concept" in bridge_candidate.rationale
|
|
|
|
|
|
def test_groundrecall_promotion_blocks_lint_errors_by_default(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "bad.md").write_text("# Bad\n\n- A claim with a synthetic lint error.\n", encoding="utf-8")
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="lint-gate-test")
|
|
lint_path = result.out_dir / "lint_findings.json"
|
|
lint_payload = json.loads(lint_path.read_text(encoding="utf-8"))
|
|
lint_payload["summary"]["error_count"] = 1
|
|
lint_payload["findings"].append(
|
|
{
|
|
"severity": "error",
|
|
"code": "claim_missing_observation",
|
|
"target_id": "clm_synthetic",
|
|
"message": "Synthetic test error.",
|
|
}
|
|
)
|
|
lint_path.write_text(json.dumps(lint_payload, indent=2), encoding="utf-8")
|
|
|
|
with pytest.raises(PromotionGateError) as excinfo:
|
|
promote_import_to_store(result.out_dir, tmp_path / "store", reviewer="R")
|
|
|
|
assert excinfo.value.payload["error_count"] == 1
|
|
assert excinfo.value.payload["errors"][0]["code"] == "claim_missing_observation"
|
|
|
|
|
|
def test_groundrecall_promotion_can_override_lint_error_gate(tmp_path: Path) -> None:
|
|
root = tmp_path / "llmwiki"
|
|
(root / "wiki").mkdir(parents=True)
|
|
(root / "wiki" / "override.md").write_text("# Override\n\n- A claim with an allowed lint error.\n", encoding="utf-8")
|
|
|
|
result = run_groundrecall_import(root, mode="quick", import_id="lint-gate-override-test")
|
|
lint_path = result.out_dir / "lint_findings.json"
|
|
lint_payload = json.loads(lint_path.read_text(encoding="utf-8"))
|
|
lint_payload["summary"]["error_count"] = 1
|
|
lint_payload["findings"].append(
|
|
{
|
|
"severity": "error",
|
|
"code": "relation_missing_source",
|
|
"target_id": "rel_synthetic",
|
|
"message": "Synthetic test error.",
|
|
}
|
|
)
|
|
lint_path.write_text(json.dumps(lint_payload, indent=2), encoding="utf-8")
|
|
|
|
payload = promote_import_to_store(result.out_dir, tmp_path / "store", reviewer="R", allow_lint_errors=True)
|
|
|
|
assert payload["lint_error_count"] == 1
|
|
assert payload["lint_errors_allowed"] is True
|