64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
"""Tests for the F4 ledger -> schema_linking.json deterministic projection."""
|
|
import json
|
|
|
|
from tht.config import DatabaseConfig
|
|
from tht.decisions import append_decision
|
|
from tht.session.store import create_session, set_schema_linking, sync_schema_linking
|
|
|
|
|
|
def _db():
|
|
return DatabaseConfig(
|
|
database="testdb", user="u", password="p",
|
|
schema="public",
|
|
)
|
|
|
|
|
|
def test_projection_from_ledger(tmp_path):
|
|
m = create_session("domanda X", _db(), tmp_path)
|
|
sid = m.id
|
|
sdir = tmp_path / sid
|
|
# seed a prior schema_linking with a join to prove joins are preserved
|
|
set_schema_linking(sid, {"question": "domanda X", "candidates": [], "joins": [
|
|
{"from": "dim_patient.cod_paz", "to": "fact_x.cod_paz"}], "excluded": []}, tmp_path)
|
|
append_decision(sdir, type="table_promoted", subject="dim_patient")
|
|
append_decision(sdir, type="column_promoted", subject="dim_patient.cod_paz")
|
|
append_decision(sdir, type="column_promoted", subject="dim_patient.nome")
|
|
append_decision(sdir, type="table_excluded", subject="fact_sost")
|
|
|
|
sync_schema_linking(sid, tmp_path)
|
|
|
|
data = json.loads((sdir / "schema_linking.json").read_text())
|
|
tabs = {(c["kind"], c["name"], c["decision"]) for c in data["candidates"]}
|
|
assert ("table", "dim_patient", "promoted") in tabs
|
|
assert ("column", "dim_patient.cod_paz", "promoted") in tabs
|
|
assert ("column", "dim_patient.nome", "promoted") in tabs
|
|
assert {"kind": "table", "name": "fact_sost"} in [
|
|
{"kind": e["kind"], "name": e["name"]} for e in data["excluded"]]
|
|
assert data["joins"], "existing joins must be preserved"
|
|
|
|
|
|
def test_projection_last_decision_wins(tmp_path):
|
|
m = create_session("domanda Y", _db(), tmp_path)
|
|
sid = m.id
|
|
sdir = tmp_path / sid
|
|
append_decision(sdir, type="table_promoted", subject="dim_x")
|
|
append_decision(sdir, type="table_excluded", subject="dim_x")
|
|
|
|
sync_schema_linking(sid, tmp_path)
|
|
|
|
data = json.loads((sdir / "schema_linking.json").read_text())
|
|
assert data["candidates"] == []
|
|
assert [e["name"] for e in data["excluded"]] == ["dim_x"]
|
|
|
|
|
|
def test_projection_defaults_question_from_manifest(tmp_path):
|
|
m = create_session("domanda Z", _db(), tmp_path)
|
|
sid = m.id
|
|
|
|
path = sync_schema_linking(sid, tmp_path)
|
|
|
|
data = json.loads(path.read_text())
|
|
assert data["question"] == "domanda Z"
|
|
assert data["candidates"] == []
|
|
assert data["excluded"] == []
|