portfolio-optimiser/tests/test_ingest_sql.py

"""I4 Step 2/3 — the ``sql`` connector (``read_sql``) + ``materialize`` dispatch.

Mirrors tests/test_ingest_materialize.py for the second source type. Covers the §4 connector
rules for ``sql`` (env-resolved ``connection_ref`` → sqlite path, read-only enforcement,
single-statement, streaming ``max_rows`` cap §8) and the §5 TYPED cell rendering the ``file``
path never exercised (all-strings): INTEGER→plain decimal, REAL→shortest round-trip, TEXT
verbatim, SQL NULL→empty string, BLOB/other→fail (never silent coercion). Offline: sqlite
fixtures under tmp_path, zero network, zero credentials, zero model calls.

Pinned SQL §5 decisions (spec-silent, delegated to I4 by the session plan; see brief):
``connection_ref`` names an env var whose value is a filesystem path to a sqlite database
opened read-only; ``float`` renders via ``repr`` (Python's shortest round-trip form).
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Any

import pytest

from conftest import expected_generated_stamp
from portfolio_optimiser import okf
from portfolio_optimiser.ingest import (
    IngestError,
    _sql_value_to_text,
    materialize,
    read_sql,
)

_INGESTED_AT = "2026-07-03T12:00:00Z"


def _make_db(path: Path, script: str) -> Path:
    """Build a sqlite fixture (read-write) from a DDL/DML script, then close it."""
    conn = sqlite3.connect(path)
    try:
        conn.executescript(script)
        conn.commit()
    finally:
        conn.close()
    return path


def _sql_manifest(
    tmp_path: Path, extractions: list[dict[str, Any]], *, connection_ref: str = "PROJ_DB"
) -> Path:
    manifest = {
        "manifest_version": 1,
        "source": {"type": "sql", "id": "db", "connection_ref": connection_ref},
        "bundle_summary": "SQL extracts from the project database.",
        "extractions": extractions,
    }
    path = tmp_path / "manifest.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    return path


# --- connector: env resolution + fail-fast (§4) --------------------------------------------------


def test_read_sql_unset_connection_ref_fails(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("PROJ_DB", raising=False)
    with pytest.raises(IngestError):
        read_sql("PROJ_DB", "SELECT 1", max_rows=10)


def test_read_sql_missing_db_file_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("PROJ_DB", str(tmp_path / "does-not-exist.sqlite"))
    with pytest.raises(IngestError):
        read_sql("PROJ_DB", "SELECT 1", max_rows=10)


# --- connector: §5 typed cell rendering ----------------------------------------------------------


def test_read_sql_renders_each_type_per_section5(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """INTEGER→plain decimal, REAL→shortest round-trip (``repr``, integral 4200.0 preserved),
    TEXT verbatim, SQL NULL→empty string — all in one ordered extract."""
    db = _make_db(
        tmp_path / "s.sqlite",
        """
        CREATE TABLE t (i INTEGER, r REAL, s TEXT, n TEXT);
        INSERT INTO t VALUES (7, 1.5, 'hei', NULL);
        INSERT INTO t VALUES (3, 4200.0, 'verden', 'x');
        """,
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    header, rows = read_sql("PROJ_DB", "SELECT i, r, s, n FROM t ORDER BY i", max_rows=100)
    assert header == ["i", "r", "s", "n"]
    assert rows == [
        ["3", "4200.0", "verden", "x"],
        ["7", "1.5", "hei", ""],
    ]


def test_read_sql_null_is_empty_string(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = _make_db(
        tmp_path / "n.sqlite",
        "CREATE TABLE t (v TEXT); INSERT INTO t VALUES (NULL);",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    _, rows = read_sql("PROJ_DB", "SELECT v FROM t", max_rows=10)
    assert rows == [[""]]  # SQL NULL → empty string, never "None"


def test_read_sql_blob_value_fails_never_silent(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    db = _make_db(
        tmp_path / "b.sqlite",
        "CREATE TABLE t (b BLOB); INSERT INTO t VALUES (x'00ff');",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    with pytest.raises(IngestError):  # §5: any other value type MUST fail
        read_sql("PROJ_DB", "SELECT b FROM t", max_rows=10)


def test_sql_value_to_text_rejects_bool_defensively() -> None:
    # sqlite3 never returns Python bool; the guard prevents str(True)->"True" silent coercion
    # (bool is an int subclass, so it would otherwise slip through the integer branch).
    with pytest.raises(IngestError):
        _sql_value_to_text(True)


# --- connector: §8 cap + §4 read-only / single-statement -----------------------------------------


def test_read_sql_max_rows_exceeded_is_error(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    db = _make_db(
        tmp_path / "m.sqlite",
        "CREATE TABLE t (i INTEGER); INSERT INTO t VALUES (1),(2),(3);",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    with pytest.raises(IngestError):  # §8: error, never silent truncation
        read_sql("PROJ_DB", "SELECT i FROM t ORDER BY i", max_rows=2)


def test_read_sql_is_read_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = _make_db(
        tmp_path / "ro.sqlite",
        "CREATE TABLE t (i INTEGER); INSERT INTO t VALUES (1);",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    with pytest.raises(IngestError):  # §4: read-only enforced at the DB (mode=ro)
        read_sql("PROJ_DB", "INSERT INTO t VALUES (2)", max_rows=10)
    _, rows = read_sql("PROJ_DB", "SELECT count(*) FROM t", max_rows=10)
    assert rows == [["1"]]  # the refused write never happened


def test_read_sql_rejects_multiple_statements(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    db = _make_db(
        tmp_path / "ms.sqlite",
        "CREATE TABLE t (i INTEGER); INSERT INTO t VALUES (1);",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    with pytest.raises(IngestError):  # §4: one statement
        read_sql("PROJ_DB", "SELECT i FROM t; SELECT i FROM t", max_rows=10)


# --- materialize dispatch: sql executes (was refused in I2) ---------------------------------------


def test_sql_manifest_materializes_with_provenance(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    db = _make_db(
        tmp_path / "s.sqlite",
        "CREATE TABLE costs (item TEXT, amount REAL); INSERT INTO costs VALUES ('led', 120.0);",
    )
    monkeypatch.setenv("PROJ_DB", str(db))
    manifest = _sql_manifest(
        tmp_path,
        [
            {
                "id": "costs",
                "title": "Project costs",
                "query": "SELECT item, amount FROM costs ORDER BY item",
                "okf_type": "dataset",
                "max_rows": 100,
            }
        ],
    )
    written = materialize(manifest, tmp_path / "bundle", ingested_at=_INGESTED_AT)
    assert len(written) == 1
    fm = okf.parse_frontmatter(written[0])
    assert fm["source_system"] == "db"
    assert fm["generated"] == expected_generated_stamp(_INGESTED_AT)
    assert fm["source_query"] == "SELECT item, amount FROM costs ORDER BY item"
    body = written[0].read_text(encoding="utf-8")
    assert "| item | amount |" in body
    assert "| led | 120.0 |" in body