ThothII/harness/tests/test_ctetest.py

"""L1: tht.ctetest — CTE file structure + test-record append-only ledger.

The CTE phase (F6) writes one .sql file per CTE block (WITH ... AS (...) only,
NO trailing SELECT) and a cte_tests.json ledger of execution outcomes. These tests
pin the structural helpers (has_trailing_select / last_cte_name / build_test_sql)
and the ledger I/O (load/append, tolerant of JSON-array and JSONL formats).
Pure logic, no DB.
"""
import json
from datetime import UTC, date, datetime
from decimal import Decimal

import pytest

from tht.ctetest import (
    CteError,
    CteTestRecord,
    _jsonable,
    append_cte_test,
    build_test_sql,
    has_trailing_select,
    last_cte_name,
    load_cte_tests,
)

# CTE file contents
_CTE_OK = "WITH a AS (SELECT 1), b AS (SELECT * FROM a)"
_CTE_WITH_SELECT = "WITH a AS (SELECT 1) SELECT * FROM a"
_CTE_SYNTAX_GARBAGE = "WITH a AS (SELECT FROM WHERE"


# --- has_trailing_select (semantics: True = file VIOLATES the rule) ----------

def test_correct_cte_file_has_no_trailing_select():
    # a correct CTE file has only WITH ... AS (...), no SELECT -> False
    assert has_trailing_select(_CTE_OK) is False


def test_file_with_trailing_select_flagged():
    # a file that already has a SELECT final VIOLATES the rule -> True
    assert has_trailing_select(_CTE_WITH_SELECT) is True


def test_unparsable_file_returns_false_no_hint():
    # garbage: no hint (can't tell), returns False
    assert has_trailing_select(_CTE_SYNTAX_GARBAGE) is False


# --- last_cte_name ----------------------------------------------------------

def test_last_cte_name_of_two_cte_block():
    assert last_cte_name(_CTE_OK) == "b"


def test_last_cte_name_single_cte():
    assert last_cte_name("WITH solo AS (SELECT 1)") == "solo"


def test_last_cte_name_rejects_non_with():
    with pytest.raises(CteError, match="WITH"):
        last_cte_name("SELECT 1")


def test_last_cte_name_rejects_unparsable():
    with pytest.raises(CteError, match="non parsabile"):
        last_cte_name("WITH a AS (SELECT FROM WHERE")


# --- build_test_sql ---------------------------------------------------------

def test_build_test_sql_appends_select_from_last_cte():
    sql = build_test_sql(_CTE_OK)
    assert sql.endswith("SELECT * FROM b")
    assert sql.startswith("WITH a AS")  # the WITH block preserved


def test_build_test_sql_single_cte():
    sql = build_test_sql("WITH solo AS (SELECT 1)")
    assert sql.rstrip().endswith("SELECT * FROM solo")


# --- ledger I/O: load_cte_tests / append_cte_test ---------------------------

def _record(**kw) -> CteTestRecord:
    base = {
        "name": "ablazione_q", "ts": datetime(2025, 1, 1, 12, 0, tzinfo=UTC), "sql_hash": "abc123",
        "status": "ok", "columns": ["x"], "row_sample": 5, "execution_ms": 42,
    }
    base.update(kw)
    return CteTestRecord(**base)


def test_load_empty_when_no_file(tmp_path):
    assert load_cte_tests(tmp_path) == []


def test_append_then_load_roundtrip(tmp_path):
    append_cte_test(tmp_path, _record(name="a", status="ok"))
    append_cte_test(tmp_path, _record(name="b", status="error", error="boom"))
    loaded = load_cte_tests(tmp_path)
    assert len(loaded) == 2
    assert loaded[0].name == "a" and loaded[0].status == "ok"
    assert loaded[1].name == "b" and loaded[1].status == "error"
    assert loaded[1].error == "boom"


def test_load_tolerates_json_array_format(tmp_path):
    # the ledger may be either JSONL (append-style) or a JSON array
    path = tmp_path / "cte_tests.json"
    path.write_text(json.dumps([
        {"name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok"},
    ]))
    loaded = load_cte_tests(tmp_path)
    assert len(loaded) == 1
    assert loaded[0].name == "x"


def test_load_tolerates_mixed_array_then_jsonl(tmp_path):
    # defensive: array + appended JSONL rows (the tolerant parser must handle both)
    path = tmp_path / "cte_tests.json"
    path.write_text(
        json.dumps([{"name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok"}])
        + "\n"
        + json.dumps({"name": "y", "ts": "2025-01-01T00:00:00", "sql_hash": "h2", "status": "ok"})
        + "\n"
    )
    loaded = load_cte_tests(tmp_path)
    assert {r.name for r in loaded} == {"x", "y"}


def test_corrupt_ledger_raises(tmp_path):
    (tmp_path / "cte_tests.json").write_text("{not valid json")
    with pytest.raises(CteError, match="corrotto"):
        load_cte_tests(tmp_path)


# --- _jsonable ----------------------------------------------------------

def test_jsonable_passes_through_native_types():
    assert _jsonable(None) is None
    assert _jsonable(True) is True
    assert _jsonable(1) == 1
    assert _jsonable(1.5) == 1.5
    assert _jsonable("abc") == "abc"


def test_jsonable_coerces_decimal_and_date_to_str():
    assert _jsonable(Decimal("12.34")) == "12.34"
    assert _jsonable(date(2025, 1, 1)) == "2025-01-01"
    assert _jsonable(datetime(2025, 1, 1, 12, 0, 0)) == "2025-01-01 12:00:00"  # noqa: DTZ001


def test_jsonable_truncates_long_strings():
    long_str = "x" * 500
    out = _jsonable(long_str)
    assert isinstance(out, str)
    assert len(out) <= 200


def test_jsonable_truncates_long_coerced_values():
    class Weird:
        def __str__(self):
            return "y" * 500

    out = _jsonable(Weird())
    assert len(out) <= 200


# --- preview_rows on CteTestRecord ---------------------------------------

def test_preview_rows_optional_default_none():
    r = _record()
    assert r.preview_rows is None


def test_preview_rows_roundtrip_through_ledger(tmp_path):
    rec = _record(name="a", preview_rows=[["x", 1], [None, 2.5]])
    append_cte_test(tmp_path, rec)
    loaded = load_cte_tests(tmp_path)
    assert loaded[0].preview_rows == [["x", 1], [None, 2.5]]


def test_old_record_without_preview_rows_still_validates(tmp_path):
    # back-compat: an old cte_tests.json row has no "preview_rows" key at all
    path = tmp_path / "cte_tests.json"
    path.write_text(json.dumps({
        "name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok",
    }))
    loaded = load_cte_tests(tmp_path)
    assert len(loaded) == 1
    assert loaded[0].preview_rows is None