hermes_overlay/tests/test_database_backends.py

from __future__ import annotations

from pathlib import Path

import pytest

from trading.domain.database import (
    HybridRow,
    PostgreSQLBackend,
    SQLiteBackend,
    configured_backend,
    validate_database_configuration,
)
from trading.domain.schema import migration_plan


class FakeCursor:
    def __init__(self, connection: "FakeRawConnection") -> None:
        self.connection = connection
        self.rowcount = 1
        self.description = [("first",), ("second",)]
        self._rows = [("a", 2)]

    def execute(self, sql, params=(), **kwargs):
        self.connection.calls.append((sql, tuple(params), dict(kwargs)))
        return self

    def fetchone(self):
        return self._rows[0] if self._rows else None

    def fetchall(self):
        return list(self._rows)

    def __iter__(self):
        return iter(self._rows)


class FakeRawConnection:
    def __init__(self) -> None:
        self.calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
        self.commits = 0
        self.rollbacks = 0
        self.closed = False

    def cursor(self):
        return FakeCursor(self)

    def commit(self):
        self.commits += 1

    def rollback(self):
        self.rollbacks += 1

    def close(self):
        self.closed = True


def test_postgres_backend_is_lazy_redacted_and_repository_compatible(monkeypatch):
    captured = {}
    raw = FakeRawConnection()

    def connector(dsn, **kwargs):
        captured["dsn"] = dsn
        captured.update(kwargs)
        return raw

    secret_dsn = "postgresql://operator:super-secret@db.internal/hermes?sslmode=require"
    backend = PostgreSQLBackend(secret_dsn, connector=connector)
    assert "super-secret" not in repr(backend)
    assert backend.supports_advisory_locks() is True

    connection = backend.connect()
    assert captured["dsn"] == secret_dsn
    assert captured["autocommit"] is False
    assert captured["application_name"] == "hermesface-v5"

    cursor = connection.execute("SELECT a,b FROM t WHERE id=?", (7,))
    assert raw.calls[-1][0] == "SELECT a,b FROM t WHERE id=%s"
    row = cursor.fetchone()
    assert isinstance(row, HybridRow)
    assert row[0] == "a" and row["second"] == 2 and dict(row) == {"first": "a", "second": 2}

    connection.execute("BEGIN IMMEDIATE")
    assert raw.calls[-1][0] == "BEGIN"
    connection.execute("INSERT OR IGNORE INTO roles(role_id) VALUES (?)", ("r",))
    assert raw.calls[-1][0] == "INSERT INTO roles(role_id) VALUES (%s) ON CONFLICT DO NOTHING"

    connection.executescript("CREATE TABLE demo(id TEXT);")
    assert raw.calls[-1][2] == {"prepare": False}
    connection.commit()
    connection.rollback()
    connection.close()
    assert raw.commits == 1 and raw.rollbacks == 1 and raw.closed is True


def test_postgres_connection_errors_do_not_echo_dsn():
    def failing_connector(*args, **kwargs):
        raise RuntimeError("driver echoed postgresql://user:secret@example/db")

    backend = PostgreSQLBackend("postgresql://user:secret@example/db", connector=failing_connector)
    with pytest.raises(RuntimeError, match="PostgreSQL connection failed") as exc_info:
        backend.connect()
    assert "secret" not in str(exc_info.value)


def test_sqlite_is_paper_only_and_memory_database_is_rejected(tmp_path, monkeypatch):
    monkeypatch.setenv("TRADING_MODE", "testnet")
    with pytest.raises(RuntimeError, match="restricted to isolated Paper"):
        SQLiteBackend(tmp_path / "db.sqlite3").connect()

    monkeypatch.setenv("TRADING_MODE", "paper")
    monkeypatch.setenv("HERMES_DATABASE_URL", "sqlite:///:memory:")
    with pytest.raises(ValueError, match="in-memory SQLite"):
        configured_backend()


def test_backend_selection_requires_postgres_for_nonpaper(monkeypatch):
    monkeypatch.delenv("HERMES_DATABASE_URL", raising=False)
    monkeypatch.setenv("TRADING_MODE", "live")
    with pytest.raises(RuntimeError, match="requires PostgreSQL"):
        configured_backend()
    ok, messages = validate_database_configuration()
    assert ok is False and "requires PostgreSQL" in " ".join(messages)

    monkeypatch.setenv("HERMES_DATABASE_URL", "postgresql://db/hermes?sslmode=require")
    backend = configured_backend()
    assert isinstance(backend, PostgreSQLBackend)
    ok, messages = validate_database_configuration()
    assert ok is True and "PostgreSQL architecture selected" in " ".join(messages)


def test_dialect_migration_plans_are_immutable_and_postgres_native():
    sqlite_plan = migration_plan("sqlite")
    postgres_plan = migration_plan("postgresql")
    assert len(sqlite_plan) >= 10
    assert [migration.version for migration in postgres_plan] == [
        "001_baseline", "002_operational_indexes",
        "003_scoped_lifecycle_metadata", "004_position_events",
        "005_execution_audit_session", "006_rate_limit_audit_integrity",
        "007_audit_append_only", "008_advisory_outcomes",
    ]
    assert all(len(migration.sha256) == 64 for migration in sqlite_plan + postgres_plan)

    sql = postgres_plan[0].path.read_text(encoding="utf-8").upper()
    for forbidden in ("AUTOINCREMENT", "PRAGMA", "RANDOMBLOB", "BEGIN IMMEDIATE", "INSERT OR"):
        assert forbidden not in sql
    for required in (
        "CREATE TABLE IF NOT EXISTS USERS",
        "CREATE TABLE IF NOT EXISTS ANALYSIS_REQUESTS",
        "CREATE TABLE IF NOT EXISTS EXECUTION_HISTORY",
        "CREATE TABLE IF NOT EXISTS POSITIONS",
        "CREATE TABLE IF NOT EXISTS TRADING_MODE_POLICIES",
    ):
        assert required in sql