ThothII/harness/tests/test_cte_info.py

"""L1: `tht cte info <name>` — read-only CTE detail (plan position + sql + doc + last test).

Feeds the JS gate (WS2) building the cte_result v2 payload: it needs index/total from
cte_plan.json (same source as tht.phase.next_cte), the persisted .sql file, the optional
cte_plan_doc.json entry, and the last CteTestRecord for the CTE. --json output must be
pristine (only valid JSON on stdout).
"""
import json
from datetime import datetime

from typer.testing import CliRunner

from tht.cli.cte_cmd import cte_app
from tht.config import DatabaseConfig
from tht.ctetest import CteTestRecord, append_cte_test
from tht.decisions import append_decision
from tht.session.store import create_session


def _db():
    return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"})  # noqa: S106


def _patch_cfg(monkeypatch, tmp_path):
    from tht.cli import cte_cmd

    class FakePaths:
        sessions = tmp_path

    class FakeCfg:
        paths = FakePaths()
        database = _db()

    monkeypatch.setattr(cte_cmd, "_load_config_or_exit", lambda _: FakeCfg())


def _make_session(tmp_path, plan):
    m = create_session("q", _db(), tmp_path)
    sdir = tmp_path / m.id
    (sdir / "cte_plan.json").write_text(json.dumps(plan))
    (sdir / "ctes").mkdir(parents=True, exist_ok=True)
    return m.id, sdir


def test_info_happy_path_json(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a", "b"])
    (sdir / "ctes" / "b.sql").write_text("WITH b AS (SELECT 1)")
    append_cte_test(sdir, CteTestRecord(
        name="b", ts=datetime(2025, 1, 1, 12, 0), sql_hash="h1", status="ok",
        columns=["x"], row_sample=1, execution_ms=5, preview_rows=[[1]],
    ))
    _patch_cfg(monkeypatch, tmp_path)

    res = CliRunner().invoke(cte_app, ["info", "b", "--session", sid, "--json"])
    assert res.exit_code == 0, res.output
    data = json.loads(res.output)
    assert data["name"] == "b"
    assert data["index"] == 2
    assert data["total"] == 2
    assert data["plan"] == ["a", "b"]
    assert data["sql"] == "WITH b AS (SELECT 1)"
    assert data["approved"] is False
    assert data["doc"] is None
    assert data["last_test"]["sql_hash"] == "h1"
    assert data["last_test"]["preview_rows"] == [[1]]


def test_info_approved_true_when_decision_present(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
    append_decision(sdir, type="cte_approved", subject="a")
    _patch_cfg(monkeypatch, tmp_path)

    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
    assert res.exit_code == 0, res.output
    data = json.loads(res.output)
    assert data["approved"] is True


def test_info_doc_present(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
    doc = {"ctes": [{"name": "a", "purpose": "p", "rationale": "r", "depends_on": []}]}
    (sdir / "cte_plan_doc.json").write_text(json.dumps(doc, ensure_ascii=False))
    _patch_cfg(monkeypatch, tmp_path)

    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
    assert res.exit_code == 0, res.output
    data = json.loads(res.output)
    assert data["doc"] == {"purpose": "p", "rationale": "r", "depends_on": []}


def test_info_last_test_none_when_no_records(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
    _patch_cfg(monkeypatch, tmp_path)

    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
    assert res.exit_code == 0, res.output
    data = json.loads(res.output)
    assert data["last_test"] is None


def test_info_last_test_is_most_recent_record(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    (sdir / "ctes" / "a.sql").write_text("WITH a AS (SELECT 1)")
    append_cte_test(sdir, CteTestRecord(
        name="a", ts=datetime(2025, 1, 1), sql_hash="old", status="ok",
    ))
    append_cte_test(sdir, CteTestRecord(
        name="a", ts=datetime(2025, 1, 2), sql_hash="new", status="ok",
    ))
    _patch_cfg(monkeypatch, tmp_path)

    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
    assert res.exit_code == 0, res.output
    data = json.loads(res.output)
    assert data["last_test"]["sql_hash"] == "new"


def test_info_nonexistent_session_exit_1(tmp_path, monkeypatch):
    _patch_cfg(monkeypatch, tmp_path)
    res = CliRunner().invoke(cte_app, ["info", "a", "--session", "nope", "--json"])
    assert res.exit_code == 1
    assert res.stdout == ""  # stdout stays pristine JSON-or-nothing


def test_info_missing_plan_exit_1(tmp_path, monkeypatch):
    m = create_session("q", _db(), tmp_path)
    _patch_cfg(monkeypatch, tmp_path)
    res = CliRunner().invoke(cte_app, ["info", "a", "--session", m.id, "--json"])
    assert res.exit_code == 1
    assert res.stdout == ""


def test_info_name_not_in_plan_exit_1(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    _patch_cfg(monkeypatch, tmp_path)
    res = CliRunner().invoke(cte_app, ["info", "zzz", "--session", sid, "--json"])
    assert res.exit_code == 1
    assert res.stdout == ""


def test_info_missing_sql_file_exit_1(tmp_path, monkeypatch):
    sid, sdir = _make_session(tmp_path, ["a"])
    _patch_cfg(monkeypatch, tmp_path)
    res = CliRunner().invoke(cte_app, ["info", "a", "--session", sid, "--json"])
    assert res.exit_code == 1
    assert res.stdout == ""