Spaces:
Sleeping
Sleeping
File size: 5,536 Bytes
2e658e7 1f8cf56 2e658e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | 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
|