ThothII/harness/tests/test_sql_preview_json.py

"""Task 6: `tht sql preview --json` + `--offset` paging.

Pure-logic tests (no DB needed):
  1. inject_limit_offset — subquery wrapping, non-destructive.
  2. preview_cmd JSON-purity — monkeypatched do_run, verifies stdout is valid JSON.
"""
import json

import pytest

from tht.execute.limit import inject_limit_offset


def test_inject_limit_offset_wraps_query():
    sql = "SELECT a FROM t ORDER BY a"
    out = inject_limit_offset(sql, limit=10, offset=20)
    assert "LIMIT 10" in out and "OFFSET 20" in out
    # original query stays as a subquery (no clobber of an existing user LIMIT)
    assert "SELECT a FROM t ORDER BY a" in out


def test_inject_limit_offset_zero_offset_no_offset_clause():
    out = inject_limit_offset("SELECT 1", limit=5, offset=0)
    assert "LIMIT 5" in out
    assert "OFFSET" not in out


def test_inject_limit_offset_strips_trailing_semicolon():
    out = inject_limit_offset("SELECT 1;", limit=3, offset=0)
    assert "LIMIT 3" in out
    # semicolon should not appear in the inner subquery
    assert ";" not in out


def test_inject_limit_offset_default_offset_zero():
    out = inject_limit_offset("SELECT x FROM y", limit=7)
    assert "LIMIT 7" in out
    assert "OFFSET" not in out


def test_do_run_offset_truncated_when_extra_row(monkeypatch):
    """Real do_run offset>0 path: runner returns N+1 rows → truncated True, exactly N returned.

    Stubs the transport layer (_run_transport) so do_run's own offset logic
    (probe_limit=N+1, compute truncated, slice to N) is exercised for real.
    """
    from types import SimpleNamespace

    from tht.cli import sql_cmd
    from tht.execute import ExecResult

    captured = {}

    def fake_transport(cfg, sql, *, limit):
        captured["sql"] = sql
        captured["limit"] = limit
        # runner sliced to probe_limit (N+1=3): it found the sentinel extra row.
        return ExecResult(columns=["a"], rows=[(1,), (2,), (3,)], execution_ms=5, truncated=False)

    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
    result = sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=10)

    # do_run must have probed for N+1 rows and wrapped with OFFSET.
    assert captured["limit"] == 3
    assert "OFFSET 10" in captured["sql"] and "LIMIT 3" in captured["sql"]
    # extra row detected → truncated True, sliced back to exactly N=2.
    assert result.truncated is True
    assert result.rows == [(1,), (2,)]


def test_do_run_offset_not_truncated_when_exactly_n(monkeypatch):
    """Real do_run offset>0 path: runner returns exactly N rows → truncated False."""
    from types import SimpleNamespace

    from tht.cli import sql_cmd
    from tht.execute import ExecResult

    def fake_transport(cfg, sql, *, limit):
        return ExecResult(columns=["a"], rows=[(1,), (2,)], execution_ms=5, truncated=False)

    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
    result = sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=10)

    assert result.truncated is False
    assert result.rows == [(1,), (2,)]


def test_do_run_offset_zero_path_unchanged(monkeypatch):
    """offset == 0 bypasses the wrapper entirely: SQL and limit reach the runner verbatim."""
    from types import SimpleNamespace

    from tht.cli import sql_cmd
    from tht.execute import ExecResult

    captured = {}

    def fake_transport(cfg, sql, *, limit):
        captured["sql"] = sql
        captured["limit"] = limit
        return ExecResult(columns=["a"], rows=[(1,)], execution_ms=5, truncated=False)

    monkeypatch.setattr(sql_cmd, "_run_transport", fake_transport)
    sql_cmd.do_run(SimpleNamespace(), "SELECT a FROM t", limit=2, offset=0)

    # no wrapper applied: original SQL and original limit passed straight through.
    assert captured["sql"] == "SELECT a FROM t"
    assert captured["limit"] == 2


def test_preview_session_no_file_resolves_sql_final(monkeypatch, tmp_path, capsys):
    """--session without a positional FILE resolves sql_final.sql via _session_sql_file."""
    from types import SimpleNamespace

    from tht.cli import sql_cmd

    sql_file = tmp_path / "sql_final.sql"
    sql_file.write_text("SELECT session_resolved")

    # Patch _session_sql_file to return our tmp file (no real workspace/DB needed).
    monkeypatch.setattr(sql_cmd, "_session_sql_file", lambda cfg, sid: sql_file)

    captured_sql = {}

    def fake_do_run(cfg, sql, *, limit, offset=0):
        captured_sql["sql"] = sql
        return SimpleNamespace(columns=["x"], rows=[[42]], execution_ms=1, truncated=False)

    monkeypatch.setattr(sql_cmd, "do_run", fake_do_run)
    monkeypatch.setattr(sql_cmd, "validate_or_exit", lambda *a, **k: SimpleNamespace(ast=None))
    monkeypatch.setattr(sql_cmd, "require_action", lambda *a, **k: None)
    monkeypatch.setattr(
        sql_cmd,
        "_load_config_or_exit",
        lambda *a, **k: SimpleNamespace(execution=SimpleNamespace(max_preview_rows=100)),
    )

    # Call with file=None, session="abc123" — should NOT raise.
    sql_cmd.preview_cmd(file=None, limit=None, offset=0, session="abc123", json_out=True, config=None)

    out = capsys.readouterr().out.strip()
    data = json.loads(out)
    assert data["columns"] == ["x"]
    assert data["rows"] == [[42]]
    # Confirm the SQL came from the file, not a None read.
    assert captured_sql["sql"] == "SELECT session_resolved"


def test_preview_json_pure_stdout(monkeypatch, tmp_path, capsys):
    from types import SimpleNamespace

    from tht.cli import sql_cmd

    fake = SimpleNamespace(columns=["a"], rows=[[1], [2]], execution_ms=3, truncated=False)
    monkeypatch.setattr(sql_cmd, "do_run", lambda *a, **k: fake)
    monkeypatch.setattr(sql_cmd, "validate_or_exit", lambda *a, **k: SimpleNamespace(ast=None))
    monkeypatch.setattr(sql_cmd, "require_action", lambda *a, **k: None)
    monkeypatch.setattr(
        sql_cmd,
        "_load_config_or_exit",
        lambda *a, **k: SimpleNamespace(execution=SimpleNamespace(max_preview_rows=100)),
    )
    f = tmp_path / "q.sql"
    f.write_text("SELECT 1")
    sql_cmd.preview_cmd(file=f, limit=None, offset=0, session=None, json_out=True, config=None)
    out = capsys.readouterr().out.strip()
    data = json.loads(out)  # must parse: stdout pristine
    assert data["columns"] == ["a"]
    assert data["rows"] == [[1], [2]]
    assert data["execution_ms"] == 3
    assert data["truncated"] is False
    assert "limit" in data
    assert data["offset"] == 0