Files
ThothII/harness/tests/test_formula_wiring.py
T

120 lines
5.0 KiB
Python

"""D14b wiring: status=auto, search_formulas, and evidence loader skips formula files.
Completes the formula layer beyond the store: the `auto` status the spec requires
(§4.7.2), the concept-substring retrieval that `tht search find --kind formula` uses,
and the guarantee that load_evidence_dir does NOT choke on *.sql.md formula files when
they live under the evidence root.
"""
import json
from types import SimpleNamespace
from typer.testing import CliRunner
from tht.cli import app, search_cmd
from tht.evidence import formula_store
from tht.evidence.formula_store import ConceptFormula, save_formula, search_formulas
from tht.evidence.model import EvidenceDoc, load_evidence_dir
from tht.evidence.search import EvidenceSearchOutcome
def test_status_auto_is_valid():
f = ConceptFormula(concept="x", sql="SELECT 1", status="auto")
assert f.status == "auto"
# round-trips through parse/dump
assert ConceptFormula.parse(f.dump()).status == "auto"
def test_search_formulas_substring_case_insensitive(tmp_path):
save_formula(tmp_path, ConceptFormula(concept="fascia pediatrica", sql="SELECT 1"))
save_formula(tmp_path, ConceptFormula(concept="indice di Charlson", sql="SELECT 2"))
hits = search_formulas(tmp_path, "PEDIATRICA")
assert len(hits) == 1
assert hits[0].concept == "fascia pediatrica"
assert search_formulas(tmp_path, "charlson")[0].concept == "indice di Charlson"
def test_load_evidence_dir_skips_formula_files(tmp_path):
# a real evidence doc + a formula file under the same root
(tmp_path / "ev1.md").write_text(
"---\nid: ev1\ntitle: T\n---\nbody text\n"
)
save_formula(tmp_path, ConceptFormula(concept="ablazione", sql="SELECT 1"))
docs = load_evidence_dir(tmp_path)
ids = [d.id for d in docs]
assert ids == ["ev1"] # the .sql.md formula file is skipped, no crash
assert all(isinstance(d, EvidenceDoc) for d in docs)
def test_unreviewed_legacy_formulas_cannot_become_curated_evidence():
for status in ("auto", "draft"):
formula = ConceptFormula(
concept="fascia pediatrica",
columns=["clinical.patient.birth_date"],
sql="CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
status=status,
)
assert formula_store.legacy_formula_to_curated(
formula, legacy_path="formulas/pediatric-1.sql.md",
) is None
def test_formula_search_uses_typed_evidence_with_a_formula_constraint():
class Searcher:
vector_generation = "gen:" + "a" * 32
def __init__(self):
self.calls = []
def search(self, embedding, **kwargs):
self.calls.append((embedding, kwargs))
return [SimpleNamespace(
id="fragment:formula", similarity=0.9, content="formula excerpt",
title="Fascia pediatrica", metadata={
"evidence_id": "evidence:fascia-pediatrica",
"evidence_kind": "formula",
"document_id": "doc:formula",
"ordinal": 0,
"source_uri": "file:///curated/formula/fascia-pediatrica.md",
"provenance": {},
},
)]
class Embedder:
def embed_query(self, query):
return [0.25]
searcher = Searcher()
outcome = search_cmd.search_formula_evidence(
"fascia pediatrica", searcher=searcher, embedder=Embedder(), top=3,
)
assert outcome.status == "available"
assert [result.evidence_id for result in outcome.results] == ["evidence:fascia-pediatrica"]
assert searcher.calls[0][1]["metadata_filter"]["required_kinds"] == ["formula"]
def test_formula_search_json_is_pristine_while_human_output_warns_about_legacy_store(monkeypatch):
monkeypatch.setattr(search_cmd, "_load_config_or_exit", lambda _path: SimpleNamespace(embeddings=object()))
monkeypatch.setattr(search_cmd, "workspace_id_for_config", lambda _cfg, _path: "workspace-a")
monkeypatch.setattr(search_cmd, "search_formula_evidence", lambda *args, **kwargs: EvidenceSearchOutcome(
"available", "gen:" + "a" * 32,
))
monkeypatch.setattr("tht.cli.vector_cmd.require_vector_cfg", lambda _cfg: None)
monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", lambda _cfg: object())
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda _cfg: object())
monkeypatch.setattr("tht.evidence.active_searcher", lambda *args, **kwargs: object())
monkeypatch.setattr("tht.evidence.validate_corpus_workspace", lambda _cfg, _workspace: None)
json_result = CliRunner().invoke(
app, ["search", "find", "fascia pediatrica", "--kind", "formula", "--top", "1", "--json"],
)
human_result = CliRunner().invoke(
app, ["search", "find", "fascia pediatrica", "--kind", "formula", "--top", "1"],
)
assert json_result.exit_code == 0
assert json.loads(json_result.stdout) == []
assert "ATTENZIONE" not in json_result.stdout
assert human_result.exit_code == 0
assert "ATTENZIONE" in human_result.output