ThothII/harness/tests/test_sqlcheck.py

"""L1: tht.sqlcheck.validate_sql — static validation of reviewer-SQL (read-only, no DDL).

The gate runs generated SQL through validate_sql before it ever touches the DB.
A bug here is dangerous (lets a DELETE through, or flags a valid SELECT). These
tests pin the read-only contract + forbidden functions + object-existence checks.
Pure logic, no DB.
"""
import pytest

from tht.mschema.models import (
    ColumnPhysical,
    PhysicalSchema,
    TablePhysical,
)
from tht.sqlcheck import validate_sql


def _schema() -> PhysicalSchema:
    """A tiny known schema for the object-existence checks."""
    from datetime import datetime

    return PhysicalSchema(
        database="db",
        schema="dw",
        introspected_at=datetime(2025, 1, 1),
        tables={
            "dim_pazienti": TablePhysical(
                columns={
                    "id_paziente": ColumnPhysical(type="bigint", pk=True),
                    "citta": ColumnPhysical(type="varchar(100)"),
                }
            ),
            "fct_ricoveri": TablePhysical(
                columns={
                    "id_ricovero": ColumnPhysical(type="bigint", pk=True),
                    "fk_paziente": ColumnPhysical(type="bigint"),
                },
            ),
        },
    )


# --- parse + single-statement -----------------------------------------------

def test_valid_simple_select_ok():
    r = validate_sql("SELECT 1 AS x")
    assert r.ok
    assert r.errors == []


def test_unparsable_sql_returns_clear_error():
    r = validate_sql("SELECT FROM WHERE")
    assert not r.ok
    assert any("non parsabile" in e for e in r.errors)


def test_two_statements_rejected():
    r = validate_sql("SELECT 1; SELECT 2")
    assert not r.ok
    assert any("uno solo statement" in e for e in r.errors)


# --- read-only enforcement (the load-bearing contract) -----------------------

@pytest.mark.parametrize("ddl", [
    "INSERT INTO t VALUES (1)",
    "UPDATE t SET x = 1",
    "DELETE FROM t",
    "CREATE TABLE t (x int)",
    "DROP TABLE t",
    "ALTER TABLE t ADD COLUMN x int",
    "TRUNCATE t",
    "GRANT SELECT ON t TO public",
])
def test_ddl_rejected(ddl):
    r = validate_sql(ddl)
    assert not r.ok
    assert any("read-only" in e or "SELECT" in e for e in r.errors)


def test_with_select_ok():
    r = validate_sql("WITH cte AS (SELECT 1) SELECT * FROM cte")
    assert r.ok


def test_union_ok():
    r = validate_sql("SELECT 1 UNION SELECT 2")
    assert r.ok


# --- forbidden functions ----------------------------------------------------

def test_default_forbidden_function_rejected():
    # dblink is in the default blacklist (side-effect: outbound DB connection)
    r = validate_sql("SELECT dblink('host=x', 'SELECT 1')")
    assert not r.ok
    assert any("dblink" in e for e in r.errors)


def test_custom_forbidden_function_set_used():
    # caller narrows the blacklist; regexp_match is an Anonymous function whose
    # .name resolves to "regexp_match", so the matcher catches it.
    r = validate_sql("SELECT regexp_match(col, 'x')", forbidden_functions={"regexp_match"})
    assert not r.ok
    assert any("regexp_match" in e for e in r.errors)


def test_allowed_function_not_flagged():
    r = validate_sql("SELECT coalesce(NULL, 0)")
    assert r.ok


def test_specialized_aggregate_not_caught_by_name_matcher():
    # KNOWN LIMITATION (documents real behavior): sqlglot models count/sum/etc. as
    # specialized Func nodes whose `.name` is the ARGUMENT ('*' for count(*)), not
    # the function name. So the name-based matcher in validate_sql does NOT flag
    # specialized aggregates even when listed. Only Anonymous functions (coalesce,
    # dblink, set_config...) are matched. If a customer needs to forbid count(), a
    # separate node-type check would be required -- not in scope here.
    r = validate_sql("SELECT count(*)", forbidden_functions={"count"})
    assert r.ok  # NOT caught -- this is the documented limitation


# --- object existence (when physical schema given) --------------------------

def test_nonexistent_table_flagged():
    r = validate_sql("SELECT * FROM non_esiste", physical=_schema())
    assert not r.ok
    assert any("inesistente" in e and "non_esiste" in e for e in r.errors)


def test_existing_table_ok():
    r = validate_sql("SELECT * FROM dim_pazienti", physical=_schema())
    assert r.ok


def test_cte_table_not_flagged_as_nonexistent():
    # a CTE is not a real table; must not be flagged as missing
    r = validate_sql("WITH cte AS (SELECT 1) SELECT * FROM cte", physical=_schema())
    assert r.ok


def test_promoted_perimeter_warning():
    # dim_pazienti is NOT in the promoted perimeter -> warning (not error)
    r = validate_sql("SELECT * FROM dim_pazienti", physical=_schema(), promoted_tables={"fct_ricoveri"})
    assert r.ok  # warning, not error
    assert any("fuori dal perimetro" in w for w in r.warnings)


def test_nonexistent_column_flagged():
    r = validate_sql("SELECT p.colonna_inesistente FROM dim_pazienti p", physical=_schema())
    assert not r.ok
    assert any("colonna inesistente" in e for e in r.errors)


def test_existing_column_with_alias_ok():
    r = validate_sql("SELECT p.citta FROM dim_pazienti p", physical=_schema())
    assert r.ok


def test_no_physical_schema_skips_object_checks():
    # without physical, object existence is not checked (trusted context)
    r = validate_sql("SELECT * FROM anything")
    assert r.ok