95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
"""Regression: recon SQL validator catches LLM-generated schema mismatches.
|
|
|
|
Bug captured: drill_down generated `SELECT district_id FROM loan` —
|
|
syntactically valid, parses, but `loan` doesn't have a `district_id` column.
|
|
We want to catch this BEFORE Postgres does, with a clear message naming
|
|
where the column actually lives.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from api.recon.types import Column, Metric, Recon, Relationship, Table
|
|
from api.recon.validate import ReconValidationError, validate_sql
|
|
|
|
|
|
def _recon() -> Recon:
|
|
return Recon(
|
|
schema="financial",
|
|
tables={
|
|
"loan": Table("loan", None, [
|
|
Column("loan_id", "BIGINT", False),
|
|
Column("account_id", "BIGINT", True),
|
|
Column("amount", "BIGINT", True),
|
|
Column("status", "TEXT", True),
|
|
]),
|
|
"account": Table("account", None, [
|
|
Column("account_id", "BIGINT", False),
|
|
Column("district_id", "BIGINT", True),
|
|
]),
|
|
"district": Table("district", None, [
|
|
Column("district_id", "BIGINT", False),
|
|
Column("A2", "TEXT", True),
|
|
Column("A13", "BIGINT", True),
|
|
]),
|
|
},
|
|
metrics={},
|
|
column_to_tables={
|
|
"loan_id": ["loan"],
|
|
"account_id": ["account", "loan"],
|
|
"amount": ["loan"],
|
|
"status": ["loan"],
|
|
"district_id": ["account", "district"],
|
|
"A2": ["district"],
|
|
"A13": ["district"],
|
|
},
|
|
relationships=[
|
|
Relationship("account", "district_id", "district", "district_id"),
|
|
Relationship("loan", "account_id", "account", "account_id"),
|
|
],
|
|
)
|
|
|
|
|
|
def test_valid_sql_passes():
|
|
sql = "SELECT account_id, amount FROM loan WHERE status = 'B'"
|
|
validate_sql(sql, _recon())
|
|
|
|
|
|
def test_valid_join_passes():
|
|
sql = """
|
|
SELECT d.A2, AVG(l.amount)
|
|
FROM loan l
|
|
JOIN account a ON l.account_id = a.account_id
|
|
JOIN district d ON a.district_id = d.district_id
|
|
GROUP BY d.A2
|
|
"""
|
|
validate_sql(sql, _recon())
|
|
|
|
|
|
def test_unknown_column_on_table_raises():
|
|
# The exact bug we hit in production: district_id picked from loan
|
|
# without joining through account.
|
|
sql = """
|
|
WITH x AS (SELECT district_id, AVG(amount) FROM loan GROUP BY district_id)
|
|
SELECT * FROM x
|
|
"""
|
|
with pytest.raises(ReconValidationError) as exc:
|
|
validate_sql(sql, _recon())
|
|
msg = str(exc.value)
|
|
assert "district_id" in msg
|
|
# The hint should name where district_id actually lives.
|
|
assert "account" in msg and "district" in msg
|
|
|
|
|
|
def test_misspelled_column_raises():
|
|
sql = "SELECT amnt FROM loan"
|
|
with pytest.raises(ReconValidationError) as exc:
|
|
validate_sql(sql, _recon())
|
|
assert "amnt" in str(exc.value)
|
|
|
|
|
|
def test_qualified_column_on_wrong_table_raises():
|
|
# district.amount doesn't exist (amount is on loan).
|
|
sql = "SELECT d.amount FROM district d"
|
|
with pytest.raises(ReconValidationError):
|
|
validate_sql(sql, _recon())
|