36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
"""L1: store.set_schema_linking — validate against SchemaLinking, then write."""
|
|
import json
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from tht.config import DatabaseConfig
|
|
from tht.session.models import SchemaLinking
|
|
from tht.session.store import create_session, set_schema_linking
|
|
|
|
|
|
def _db():
|
|
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
|
|
|
|
|
|
def test_writes_and_revalidates(tmp_path):
|
|
m = create_session("q", _db(), tmp_path)
|
|
data = {
|
|
"question": "q riscritta",
|
|
"candidates": [{"kind": "table", "name": "fact_x", "decision": "promoted"}],
|
|
"joins": [{"from": "a.k", "to": "b.k"}],
|
|
}
|
|
path = set_schema_linking(m.id, data, tmp_path)
|
|
assert path.exists()
|
|
reloaded = json.loads(path.read_text())
|
|
assert reloaded["candidates"][0]["name"] == "fact_x"
|
|
assert reloaded["joins"][0]["from"] == "a.k" # 'from' alias round-trips
|
|
SchemaLinking.model_validate(reloaded) # re-validates clean
|
|
|
|
|
|
def test_rejects_invalid_and_writes_nothing(tmp_path):
|
|
m = create_session("q", _db(), tmp_path)
|
|
with pytest.raises(ValidationError):
|
|
set_schema_linking(m.id, {"question": "q", "bogus": 1}, tmp_path) # extra=forbid
|
|
assert not (tmp_path / m.id / "schema_linking.json").exists()
|