marcopan
f104c015a1
New evidence/formula_store.py: ConceptFormula (concept, columns, sql, status, sources) as a frontmatter-YAML + SQL-body unit, stored one-file-per-formula under <formulas>/<slug>-<n>.sql.md. save_formula is append-only (competing drafts and reviewed versions coexist); retrieve_formula(concept) returns all of them so the gate can surface candidates and let the reviewer choose. concept_formula_approved / concept_formula_rejected added to DecisionType (records the reviewer's choice; approved formulas travel with schema-linking). L1: test_formula (7 tests) -- retrieval by concept, save/reload roundtrip (SQL body preserved, frontmatter well-formed), multiple formulas per concept, empty on no-match / missing dir, decision-type existence, default draft status. Deferred: --kind formula on nsp search (needs search_cmd porting) wires retrieve_formula into the CLI; lands with the search command.
86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
"""L1: SQL formula evidence -- concept->formula units (spec D14b).
|
|
|
|
A concept (e.g. 'fascia pediatrica') maps to a SQL formula (CASE WHEN ...) that
|
|
derives it from physical columns. These are reusable, reviewable units: the gate
|
|
surfaces a candidate formula, the reviewer approves or rejects it (recorded via
|
|
concept_formula_approved / concept_formula_rejected), and approved formulas are
|
|
part of the schema-linking artifact. The store is frontmatter-YAML + SQL body.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nsp.evidence.formula_store import ConceptFormula, retrieve_formula, save_formula
|
|
|
|
|
|
def test_formula_retrieval_by_concept(tmp_path):
|
|
f = ConceptFormula(
|
|
concept="fascia pediatrica",
|
|
columns=["data_nascita"],
|
|
sql="CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulto' END",
|
|
status="reviewed",
|
|
sources=["src1"],
|
|
)
|
|
save_formula(tmp_path, f)
|
|
results = retrieve_formula(tmp_path, "fascia pediatrica")
|
|
assert len(results) == 1
|
|
assert results[0].sql.startswith("CASE WHEN")
|
|
assert results[0].concept == "fascia pediatrica"
|
|
assert results[0].status == "reviewed"
|
|
assert "data_nascita" in results[0].columns
|
|
|
|
|
|
def test_save_and_reload_roundtrip_preserves_sql_body(tmp_path):
|
|
f = ConceptFormula(
|
|
concept="fascia pediatrica",
|
|
columns=["data_nascita"],
|
|
sql="CASE\n WHEN x THEN 1\nEND",
|
|
status="draft",
|
|
sources=[],
|
|
)
|
|
path = save_formula(tmp_path, f)
|
|
assert path.exists()
|
|
# the file is frontmatter YAML + SQL body
|
|
text = path.read_text()
|
|
assert text.startswith("---")
|
|
assert "concept: fascia pediatrica" in text
|
|
assert "CASE" in text # SQL body preserved
|
|
|
|
|
|
def test_retrieve_multiple_formulas_for_concept(tmp_path):
|
|
# two competing formulas for the same concept (different sources/status)
|
|
save_formula(tmp_path, ConceptFormula(concept="ablazione", columns=["flag"],
|
|
sql="SELECT 1", status="draft", sources=["a"]))
|
|
save_formula(tmp_path, ConceptFormula(concept="ablazione", columns=["testo"],
|
|
sql="SELECT 2", status="reviewed", sources=["b"]))
|
|
results = retrieve_formula(tmp_path, "ablazione")
|
|
assert len(results) == 2
|
|
statuses = {r.status for r in results}
|
|
assert statuses == {"draft", "reviewed"}
|
|
|
|
|
|
def test_retrieve_empty_when_no_match(tmp_path):
|
|
save_formula(tmp_path, ConceptFormula(concept="altro", columns=["c"],
|
|
sql="SELECT 1", status="reviewed", sources=[]))
|
|
assert retrieve_formula(tmp_path, "inesistente") == []
|
|
|
|
|
|
def test_retrieve_empty_on_missing_dir(tmp_path):
|
|
# no formulas dir at all -> empty list, not error
|
|
assert retrieve_formula(tmp_path / "nope", "anything") == []
|
|
|
|
|
|
def test_concept_formula_decision_types_exist():
|
|
import typing
|
|
from nsp.decisions import DecisionType
|
|
args = typing.get_args(DecisionType)
|
|
assert "concept_formula_approved" in args
|
|
assert "concept_formula_rejected" in args
|
|
|
|
|
|
def test_concept_formula_default_status(tmp_path):
|
|
# status has a sensible default so an author can write a draft quickly
|
|
f = ConceptFormula(concept="x", columns=["c"], sql="SELECT 1")
|
|
assert f.status == "draft" # not yet reviewed
|
|
assert f.sources == []
|