Kjell Tore Guttormsen
c94ed6d525
TDD step 4: read_sql executes one read-only SELECT against the sqlite database whose path is resolved at run time from the env var named by connection_ref (credentials never in the manifest). Read-only enforced by the connection mode (file:...?mode=ro), single statement via Connection.execute, §5 cell text conversion, streaming max_rows cap, and every sqlite failure wrapped in SourceError. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""The `sql` connector (ingest-spec §4, §8).
|
|
|
|
sqlite via connection_ref env-var resolution (the ref names the variable;
|
|
its value is the database path), read-only open, single statement,
|
|
streaming max_rows cap, typed errors for every sqlite failure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from llm_ingestion_okf.connectors import read_sql
|
|
from llm_ingestion_okf.errors import IngestError, SourceError
|
|
|
|
REF = "OKF_TEST_DB"
|
|
|
|
|
|
@pytest.fixture
|
|
def db_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
path = tmp_path / "fixture.db"
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("CREATE TABLE t (id INTEGER, name TEXT, score REAL, note TEXT)")
|
|
conn.executemany(
|
|
"INSERT INTO t VALUES (?, ?, ?, ?)",
|
|
[(1, "alpha", 1.5, None), (2, "beta", 0.1, "x|y")],
|
|
)
|
|
monkeypatch.setenv(REF, str(path))
|
|
return path
|
|
|
|
|
|
# --- happy path: §5 text form, source order ---
|
|
|
|
|
|
def test_reads_header_and_rows_as_text(db_path: Path) -> None:
|
|
header, rows = read_sql(REF, "SELECT id, name, score, note FROM t ORDER BY id", max_rows=10)
|
|
assert header == ["id", "name", "score", "note"]
|
|
# NULL -> empty string; int plain decimal; float shortest round-trip; text verbatim.
|
|
assert rows == [["1", "alpha", "1.5", ""], ["2", "beta", "0.1", "x|y"]]
|
|
|
|
|
|
# --- env-var resolution (credentials never in the manifest, §4) ---
|
|
|
|
|
|
def test_unset_connection_ref_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv(REF, raising=False)
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELECT 1", max_rows=10)
|
|
|
|
|
|
def test_empty_connection_ref_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv(REF, "")
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELECT 1", max_rows=10)
|
|
|
|
|
|
def test_missing_database_file_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv(REF, str(tmp_path / "nope.db"))
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELECT 1", max_rows=10)
|
|
|
|
|
|
# --- read-only, single statement, typed sqlite errors ---
|
|
|
|
|
|
def test_write_statement_fails_at_the_database(db_path: Path) -> None:
|
|
"""Read-only is enforced by the connection mode, not string parsing."""
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "DELETE FROM t", max_rows=10)
|
|
|
|
|
|
def test_multiple_statements_rejected(db_path: Path) -> None:
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELECT 1; SELECT 2", max_rows=10)
|
|
|
|
|
|
def test_syntax_error_wrapped_in_typed_error(db_path: Path) -> None:
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELEC nonsense", max_rows=10)
|
|
|
|
|
|
def test_blob_cell_fails_typed(db_path: Path, tmp_path: Path) -> None:
|
|
"""Never silent coercion (§5): a BLOB cell is a typed IngestError."""
|
|
with sqlite3.connect(tmp_path / "fixture.db") as conn:
|
|
conn.execute("INSERT INTO t VALUES (3, 'blob', 0.0, x'DEADBEEF')")
|
|
with pytest.raises(IngestError):
|
|
read_sql(REF, "SELECT note FROM t WHERE id = 3", max_rows=10)
|
|
|
|
|
|
# --- max_rows cap (§8): error, never silent truncation ---
|
|
|
|
|
|
def test_row_count_at_cap_is_allowed(db_path: Path) -> None:
|
|
_, rows = read_sql(REF, "SELECT id FROM t ORDER BY id", max_rows=2)
|
|
assert len(rows) == 2
|
|
|
|
|
|
def test_exceeding_cap_is_an_error(db_path: Path) -> None:
|
|
with pytest.raises(SourceError):
|
|
read_sql(REF, "SELECT id FROM t ORDER BY id", max_rows=1)
|