Spaces:
Running
Running
File size: 3,644 Bytes
e37d4de | 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 | """
Testy PostgreSQL kompatibilní vrstvy v db.py.
Bez reálného PostgreSQL serveru (není v CI/testovacím prostředí k dispozici)
nelze otestovat end-to-end zápis/čtení - tyto testy proto ověřují jádro té
části, která se mezi backendy liší: detekci backendu z DATABASE_URL a
překlad "?" -> "%s" v _execute(). Zbytek modulu (save_record, get_records,
stats, export_records) backend nerozlišuje a je už pokrytý testy nad SQLite
v test_main_llm_integration.py.
"""
from __future__ import annotations
import importlib
from unittest.mock import MagicMock
import db
def test_is_postgres_detects_postgres_urls():
for url in ("postgres://u:p@host/db", "postgresql://u:p@host:5432/db"):
assert url.startswith(("postgres://", "postgresql://"))
def test_is_postgres_false_when_unset(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
importlib.reload(db)
try:
assert db.IS_POSTGRES is False
assert db.DATABASE_URL is None
finally:
# vrátit modul do stavu, který očekává zbytek test suite (SQLite,
# CONV_DB_PATH nastavené na začátku test_main_llm_integration.py)
importlib.reload(db)
def test_execute_translates_placeholders_for_postgres(monkeypatch):
fake_cursor = MagicMock()
fake_conn = MagicMock()
fake_conn.cursor.return_value = fake_cursor
monkeypatch.setattr(db, "IS_POSTGRES", True)
monkeypatch.setattr(db, "_db_conn", fake_conn)
db._execute("SELECT * FROM records WHERE topic = ? AND sentiment = ?", ("x", "y"))
fake_conn.cursor.assert_called_once()
called_query, called_params = fake_cursor.execute.call_args[0]
assert called_query == "SELECT * FROM records WHERE topic = %s AND sentiment = %s"
assert called_params == ("x", "y")
def test_execute_keeps_sqlite_placeholders_untouched(monkeypatch):
fake_conn = MagicMock()
monkeypatch.setattr(db, "IS_POSTGRES", False)
monkeypatch.setattr(db, "_db_conn", fake_conn)
db._execute("SELECT * FROM records WHERE topic = ?", ("x",))
fake_conn.execute.assert_called_once_with("SELECT * FROM records WHERE topic = ?", ("x",))
def test_execute_reconnects_when_postgres_connection_was_dropped(monkeypatch):
"""
Reprodukuje reálně pozorovanou chybu: Neon (serverless PostgreSQL) může
kdykoli zavřít nečinné spojení ze své strany (scale-to-zero, idle
timeout). Jedno sdílené spojení pro celý běh procesu by tím zůstalo
navždy rozbité (psycopg2.InterfaceError: connection already closed) pro
každý další požadavek - _execute() se místo toho musí jednou přepojit a
dotaz zopakovat.
"""
import psycopg2
dead_cursor = MagicMock()
dead_cursor.execute.side_effect = psycopg2.InterfaceError("connection already closed")
dead_conn = MagicMock()
dead_conn.cursor.return_value = dead_cursor
fresh_cursor = MagicMock()
fresh_conn = MagicMock()
fresh_conn.cursor.return_value = fresh_cursor
monkeypatch.setattr(db, "IS_POSTGRES", True)
monkeypatch.setattr(db, "_db_conn", dead_conn)
monkeypatch.setattr(db, "_get_postgres_connection", lambda: fresh_conn)
result = db._execute("SELECT * FROM records WHERE topic = ?", ("x",))
dead_cursor.execute.assert_called_once()
fresh_cursor.execute.assert_called_once_with(
"SELECT * FROM records WHERE topic = %s", ("x",)
)
assert result is fresh_cursor
# _db_conn v modulu musí zůstat přepojené na nové (živé) spojení, ne se
# vrátit zpět k mrtvému - jinak by se stejná chyba opakovala příště.
assert db._db_conn is fresh_conn
|