546 lines
18 KiB
Python
546 lines
18 KiB
Python
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from tht.evidence import (
|
|
CuratedEvidence,
|
|
dump_curated_markdown,
|
|
load_curated_tree,
|
|
parse_curated_markdown,
|
|
)
|
|
|
|
COMMON = {
|
|
"schema_version": 1,
|
|
"id": "evidence:fascia-pediatrica",
|
|
"title": "Fascia pediatrica",
|
|
"purposes": ["sql_generation"],
|
|
"applies_to": {
|
|
"concepts": ["fascia pediatrica"],
|
|
"tables": ["clinical.patient"],
|
|
"columns": ["clinical.patient.birth_date"],
|
|
},
|
|
"language": "it",
|
|
"provenance": {
|
|
"source_file": "source/domain/patient.md",
|
|
"source_sha256": "sha256:" + "a" * 64,
|
|
"supporting_excerpts": ["I pazienti sotto i 18 anni sono pediatrici."],
|
|
},
|
|
"review_items": [],
|
|
}
|
|
|
|
|
|
def test_formula_requires_its_typed_payload():
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "formula",
|
|
"payload": {"concept": "fascia pediatrica"},
|
|
})
|
|
|
|
|
|
def test_reference_rejects_formula_payload():
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "reference",
|
|
"payload": {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": "CASE WHEN true THEN 1 END",
|
|
},
|
|
})
|
|
|
|
|
|
@pytest.mark.parametrize(("kind", "payload"), [
|
|
("glossary", {
|
|
"definition": "Un paziente con età inferiore a 18 anni.",
|
|
"synonyms": ["minore"],
|
|
"variants": ["pediatrico"],
|
|
}),
|
|
("domain", {"rule": "L'età è calcolata alla data di ricovero."}),
|
|
("enum", {
|
|
"column": "clinical.episode.discharge_status",
|
|
"values": {"D": "dimesso"},
|
|
}),
|
|
("example", {
|
|
"question": "Come riconosco un paziente pediatrico?",
|
|
"interpretation": "Applicare la formula della fascia pediatrica.",
|
|
}),
|
|
("mapping", {
|
|
"concept": "fascia pediatrica",
|
|
"tables": ["clinical.patient"],
|
|
"columns": ["clinical.patient.birth_date"],
|
|
}),
|
|
("normalization", {
|
|
"input": "PEDS",
|
|
"output": "pediatrico",
|
|
"rule": "Converte il codice abbreviato nella forma canonica.",
|
|
}),
|
|
("formula", {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": "CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
|
|
}),
|
|
("reference", {
|
|
"url": "https://example.test/linea-guida",
|
|
"label": "Linea guida",
|
|
"description": "Criteri clinici di riferimento.",
|
|
}),
|
|
])
|
|
def test_every_kind_requires_a_typed_payload(kind, payload):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": kind,
|
|
"payload": payload,
|
|
})
|
|
|
|
assert evidence.kind == kind
|
|
|
|
|
|
def test_canonical_evidence_rejects_unknown_envelope_and_payload_fields():
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "domain",
|
|
"payload": {"rule": "Una regola", "unknown": "no"},
|
|
"unknown": "no",
|
|
})
|
|
|
|
|
|
@pytest.mark.parametrize("purpose", ["disambiguation", "rewriting", "schema_linking", "sql_generation"])
|
|
def test_all_public_evidence_purposes_are_accepted(purpose):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "domain",
|
|
"purposes": [purpose],
|
|
"payload": {"rule": "Una regola di dominio."},
|
|
})
|
|
|
|
assert evidence.purposes == (purpose,)
|
|
|
|
|
|
@pytest.mark.parametrize("field, value", [
|
|
("id", "formula:fascia-pediatrica"),
|
|
("purposes", ["memory"]),
|
|
("provenance", {
|
|
**COMMON["provenance"],
|
|
"source_sha256": "sha256:not-a-digest",
|
|
}),
|
|
("provenance", {
|
|
**COMMON["provenance"],
|
|
"supporting_excerpts": [],
|
|
}),
|
|
("provenance", {
|
|
**COMMON["provenance"],
|
|
"supporting_excerpts": ["x" * 1001],
|
|
}),
|
|
])
|
|
def test_common_evidence_constraints_reject_invalid_values(field, value):
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
field: value,
|
|
"kind": "domain",
|
|
"payload": {"rule": "Una regola di dominio."},
|
|
})
|
|
|
|
|
|
def test_provenance_is_immutable_after_validation():
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "domain",
|
|
"payload": {"rule": "Una regola di dominio."},
|
|
})
|
|
|
|
with pytest.raises(ValidationError):
|
|
evidence.provenance.source_file = "source/other.md"
|
|
|
|
|
|
@pytest.mark.parametrize("sql", [
|
|
"SELECT clinical.patient.birth_date FROM clinical.patient",
|
|
"WITH patients AS (SELECT 1) SELECT * FROM patients",
|
|
"DELETE FROM clinical.patient",
|
|
"CREATE TABLE scratch (id integer)",
|
|
"TRUNCATE TABLE clinical.patient",
|
|
"GRANT SELECT ON clinical.patient TO reader",
|
|
"VALUES (1)",
|
|
"TABLE clinical.patient",
|
|
"SET search_path = clinical",
|
|
"CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END; SELECT 1",
|
|
])
|
|
def test_formula_rejects_complete_sql_statements(sql):
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "formula",
|
|
"payload": {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": sql,
|
|
},
|
|
})
|
|
|
|
|
|
def test_formula_accepts_one_composable_postgresql_expression():
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "formula",
|
|
"payload": {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": "CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
|
|
},
|
|
})
|
|
|
|
assert evidence.payload.sql.startswith("CASE WHEN")
|
|
|
|
|
|
def _formula_evidence() -> CuratedEvidence:
|
|
return CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "formula",
|
|
"payload": {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": "CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
|
|
},
|
|
})
|
|
|
|
|
|
def _domain_evidence_v2() -> CuratedEvidence:
|
|
return CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 2,
|
|
"title": "Dominio Ablazione",
|
|
"kind": "domain",
|
|
"payload": {
|
|
"rule": (
|
|
"Il dominio Ablazione rappresenta la procedura transcatetere.\n\n"
|
|
"La fact centrale è `clinical.fact_ablazione`."
|
|
),
|
|
},
|
|
})
|
|
|
|
|
|
def _domain_evidence_v3() -> CuratedEvidence:
|
|
return CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 3,
|
|
"title": "Dominio Ablazione",
|
|
"kind": "domain",
|
|
"purposes": ["disambiguation", "schema_linking", "rewriting"],
|
|
"applies_to": {
|
|
"concepts": ["ablazione", "studio elettrofisiologico", "SEE"],
|
|
"tables": ["clinical.fact_ablazione"],
|
|
"columns": ["clinical.fact_ablazione.patient_id"],
|
|
},
|
|
"payload": {
|
|
"rule": (
|
|
"Il dominio Ablazione rappresenta la procedura transcatetere.\n\n"
|
|
"La fact centrale è `clinical.fact_ablazione`."
|
|
),
|
|
},
|
|
})
|
|
|
|
|
|
def test_v3_curated_markdown_replaces_frontmatter_tables_with_readable_sections(tmp_path):
|
|
evidence = _domain_evidence_v3()
|
|
path = tmp_path / "curated" / "domain" / "dominio-ablazione.md"
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
parsed = parse_curated_markdown(text, path=path)
|
|
|
|
assert text.startswith("<!-- tht:metadata:")
|
|
assert not text.startswith("---\n")
|
|
assert not any(line.startswith("|") for line in text.splitlines())
|
|
assert "> **Dominio** · Italiano" in text
|
|
assert "**Scopi:** Disambiguazione · Collegamento allo schema · Riscrittura" in text
|
|
assert "## Ambito di applicazione" in text
|
|
assert "### Concetti\n\n- ablazione\n- studio elettrofisiologico\n- SEE" in text
|
|
assert "### Tabelle\n\n- `clinical.fact_ablazione`" in text
|
|
assert "### Colonne\n\n- `clinical.fact_ablazione.patient_id`" in text
|
|
assert "<summary>Dettagli tecnici e provenienza</summary>" in text
|
|
assert parsed == evidence
|
|
|
|
|
|
def test_v3_curated_markdown_renders_enum_values_as_a_list_instead_of_a_table(tmp_path):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 3,
|
|
"kind": "enum",
|
|
"payload": {
|
|
"column": "clinical.episode.discharge_status",
|
|
"values": {"D": "dimesso", "T": "trasferito | altra struttura"},
|
|
},
|
|
})
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
|
|
assert "## Valori\n\n- `D`: dimesso\n- `T`: trasferito | altra struttura" in text
|
|
assert not any(line.startswith("|") for line in text.splitlines())
|
|
assert parse_curated_markdown(
|
|
text,
|
|
path=tmp_path / "curated" / "enum" / "discharge-status.md",
|
|
) == evidence
|
|
|
|
|
|
def test_v3_curated_markdown_rejects_visible_metadata_that_drifted_from_canonical_data():
|
|
text = dump_curated_markdown(_domain_evidence_v3()).replace(
|
|
"**Scopi:** Disambiguazione",
|
|
"**Scopi:** Testo alterato",
|
|
1,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="not canonical"):
|
|
parse_curated_markdown(text)
|
|
|
|
|
|
def test_v2_curated_markdown_renders_domain_content_in_the_markdown_body(tmp_path):
|
|
evidence = _domain_evidence_v2()
|
|
path = tmp_path / "curated" / "domain" / "dominio-ablazione.md"
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
frontmatter = text.split("---\n", 2)[1]
|
|
parsed = parse_curated_markdown(text, path=path)
|
|
|
|
assert "domain:" not in frontmatter
|
|
assert "supporting_excerpts:" not in frontmatter
|
|
assert "review_items:" not in frontmatter
|
|
assert "# Dominio Ablazione\n" in text
|
|
assert "## Regola\n\nIl dominio Ablazione" in text
|
|
assert "## Estratti di supporto\n\n> I pazienti sotto i 18 anni sono pediatrici." in text
|
|
assert parsed == evidence
|
|
|
|
|
|
@pytest.mark.parametrize(("kind", "payload", "rendered"), [
|
|
("glossary", {
|
|
"definition": "Un paziente con età inferiore a 18 anni.",
|
|
"synonyms": ["minore"],
|
|
"variants": ["pediatrico"],
|
|
}, "## Sinonimi\n\n- minore"),
|
|
("enum", {
|
|
"column": "clinical.episode.discharge_status",
|
|
"values": {"D": "dimesso", "T": "trasferito | altra struttura"},
|
|
}, "| `D` | dimesso |"),
|
|
("example", {
|
|
"question": "Come riconosco un paziente pediatrico?",
|
|
"interpretation": "Applicare la formula della fascia pediatrica.",
|
|
}, "## Domanda\n\nCome riconosco un paziente pediatrico?"),
|
|
("mapping", {
|
|
"concept": "fascia pediatrica",
|
|
"tables": ["clinical.patient"],
|
|
"columns": ["clinical.patient.birth_date"],
|
|
}, "## Tabelle\n\n- `clinical.patient`"),
|
|
("normalization", {
|
|
"input": "PEDS",
|
|
"output": "pediatrico",
|
|
"rule": "Converte il codice abbreviato nella forma canonica.",
|
|
}, "## Output\n\npediatrico"),
|
|
("formula", {
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient.birth_date"],
|
|
"sql": "CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
|
|
}, "```sql\nCASE WHEN age < 18"),
|
|
("reference", {
|
|
"url": "https://example.test/linea-guida",
|
|
"label": "Linea guida",
|
|
"description": "Criteri clinici di riferimento.",
|
|
}, "## URL\n\n<https://example.test/linea-guida>"),
|
|
])
|
|
def test_v2_curated_markdown_renders_and_round_trips_every_typed_payload(
|
|
tmp_path, kind, payload, rendered,
|
|
):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 2,
|
|
"kind": kind,
|
|
"payload": payload,
|
|
})
|
|
path = tmp_path / "curated" / kind / "fascia-pediatrica.md"
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
|
|
assert rendered in text
|
|
assert parse_curated_markdown(text, path=path) == evidence
|
|
|
|
|
|
def test_v2_curated_markdown_renders_review_items_as_readable_blocks(tmp_path):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 2,
|
|
"kind": "domain",
|
|
"review_items": [{
|
|
"code": "ambiguous_source_statement",
|
|
"message": "Il sorgente non chiarisce la data di riferimento.",
|
|
"field": "domain.rule",
|
|
}],
|
|
"payload": {"rule": "L'età è calcolata alla data di ricovero."},
|
|
})
|
|
path = tmp_path / "curated" / "domain" / "fascia-pediatrica.md"
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
|
|
assert "## Elementi da rivedere" in text
|
|
assert "### `ambiguous_source_statement`" in text
|
|
assert "Il sorgente non chiarisce la data di riferimento." in text
|
|
assert "**Campo:** `domain.rule`" in text
|
|
assert parse_curated_markdown(text, path=path) == evidence
|
|
|
|
|
|
@pytest.mark.parametrize("legacy_field", [
|
|
"review_items: []\n",
|
|
"payload:\n rule: should-not-be-ignored\n",
|
|
])
|
|
def test_v2_curated_markdown_rejects_body_owned_fields_in_frontmatter(legacy_field):
|
|
text = dump_curated_markdown(_domain_evidence_v2()).replace(
|
|
"kind: domain\n",
|
|
f"kind: domain\n{legacy_field}",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="frontmatter"):
|
|
parse_curated_markdown(text)
|
|
|
|
|
|
def test_v2_curated_markdown_rejects_unstructured_body_content():
|
|
text = dump_curated_markdown(_domain_evidence_v2()) + "should-not-be-ignored\n"
|
|
|
|
with pytest.raises(ValueError, match="unstructured"):
|
|
parse_curated_markdown(text)
|
|
|
|
|
|
def test_v2_curated_markdown_round_trips_an_empty_enum_as_an_explicit_empty_state(tmp_path):
|
|
evidence = CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"schema_version": 2,
|
|
"kind": "enum",
|
|
"payload": {
|
|
"column": "clinical.episode.discharge_status",
|
|
"values": {},
|
|
},
|
|
})
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
|
|
assert "Nessun elemento" in text
|
|
assert parse_curated_markdown(
|
|
text,
|
|
path=tmp_path / "curated" / "enum" / "discharge-status.md",
|
|
) == evidence
|
|
|
|
|
|
def test_curated_markdown_round_trip_uses_the_kind_specific_key(tmp_path):
|
|
evidence = _formula_evidence()
|
|
path = tmp_path / "curated" / "formula" / "fascia-pediatrica.md"
|
|
|
|
text = dump_curated_markdown(evidence)
|
|
parsed = parse_curated_markdown(text, path=path)
|
|
|
|
assert "formula:" in text
|
|
assert "payload:" not in text
|
|
assert parsed == evidence
|
|
|
|
|
|
def test_curated_markdown_rejects_a_kind_that_disagrees_with_its_directory(tmp_path):
|
|
text = dump_curated_markdown(_formula_evidence())
|
|
|
|
with pytest.raises(ValueError, match="directory"):
|
|
parse_curated_markdown(text, path=tmp_path / "curated" / "reference" / "guide.md")
|
|
|
|
|
|
def test_load_curated_tree_returns_documents_in_path_order_and_ignores_readmes(tmp_path):
|
|
root = tmp_path / "curated"
|
|
formula_path = root / "formula" / "fascia-pediatrica.md"
|
|
domain_path = root / "domain" / "eta.md"
|
|
formula_path.parent.mkdir(parents=True)
|
|
domain_path.parent.mkdir(parents=True)
|
|
formula_path.write_text(dump_curated_markdown(_formula_evidence()), encoding="utf-8")
|
|
domain_path.write_text(dump_curated_markdown(CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"id": "evidence:eta-ricovero",
|
|
"title": "Età al ricovero",
|
|
"kind": "domain",
|
|
"payload": {"rule": "L'età è calcolata al ricovero."},
|
|
})), encoding="utf-8")
|
|
(root / "README.md").write_text("solo navigazione", encoding="utf-8")
|
|
|
|
loaded = load_curated_tree(root)
|
|
|
|
assert [item.id for item in loaded] == [
|
|
"evidence:eta-ricovero",
|
|
"evidence:fascia-pediatrica",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(("source_file", "reference_url"), [
|
|
("source/domain/patient.pdf", "https://example.test/linea-guida"),
|
|
("../patient.md", "https://example.test/linea-guida"),
|
|
("source/domain/patient.md", "https://user:secret@example.test/linea-guida"),
|
|
("source/domain/patient.md", "https://example.test/linea-guida?access_token=secret"),
|
|
])
|
|
def test_canonical_evidence_rejects_unsafe_source_or_url(source_file, reference_url):
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "reference",
|
|
"provenance": {**COMMON["provenance"], "source_file": source_file},
|
|
"payload": {
|
|
"url": reference_url,
|
|
"label": "Linea guida",
|
|
"description": "Criteri clinici.",
|
|
},
|
|
})
|
|
|
|
|
|
def test_curated_markdown_rejects_nonempty_body_instead_of_ignoring_it():
|
|
text = dump_curated_markdown(_formula_evidence()) + "password: should-not-be-ignored\n"
|
|
|
|
with pytest.raises(ValueError, match="body"):
|
|
parse_curated_markdown(text)
|
|
|
|
|
|
@pytest.mark.parametrize(("scope", "payload"), [
|
|
(
|
|
{"tables": ["patient"], "columns": ["clinical.patient.birth_date"]},
|
|
{"rule": "Una regola."},
|
|
),
|
|
(
|
|
{"tables": ["clinical.patient"], "columns": ["clinical.patient.date.of.birth"]},
|
|
{"rule": "Una regola."},
|
|
),
|
|
(
|
|
{"tables": ["clinical.patient"], "columns": ["clinical.patient.birth_date"]},
|
|
{
|
|
"concept": "fascia pediatrica",
|
|
"columns": ["clinical.patient"],
|
|
"sql": "CASE WHEN age < 18 THEN 1 ELSE 0 END",
|
|
},
|
|
),
|
|
])
|
|
def test_schema_identifiers_require_table_or_column_shape(scope, payload):
|
|
with pytest.raises(ValidationError):
|
|
CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "formula" if "sql" in payload else "domain",
|
|
"applies_to": scope,
|
|
"payload": payload,
|
|
})
|
|
|
|
|
|
def test_load_curated_tree_rejects_non_utf8_and_oversized_files(tmp_path, monkeypatch):
|
|
root = tmp_path / "curated"
|
|
path = root / "domain" / "eta.md"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_bytes(b"\xff")
|
|
|
|
with pytest.raises(ValueError, match="UTF-8"):
|
|
load_curated_tree(root)
|
|
|
|
path.write_text(dump_curated_markdown(CuratedEvidence.model_validate({
|
|
**COMMON,
|
|
"kind": "domain",
|
|
"payload": {"rule": "Una regola."},
|
|
})), encoding="utf-8")
|
|
monkeypatch.setattr("tht.evidence.canonical.MAX_CURATED_FILE_BYTES", 1)
|
|
|
|
with pytest.raises(ValueError, match="size limit"):
|
|
load_curated_tree(root)
|