325 lines
10 KiB
Python
325 lines
10 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 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)
|