File size: 2,220 Bytes
6aecb2e | 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 | import importlib
import sys
from contextlib import contextmanager
import pytest
from sqlalchemy.exc import OperationalError
DB_MODULE_NAME = "landppt.database.database"
MAIN_MODULE_NAME = "landppt.main"
LEGACY_POSTGRES_URL = "postgresql://landppt:landppt@postgres:5432/landppt"
def _reload_database_module(monkeypatch, database_url=None):
if database_url is None:
monkeypatch.delenv("DATABASE_URL", raising=False)
else:
monkeypatch.setenv("DATABASE_URL", database_url)
sys.modules.pop(DB_MODULE_NAME, None)
return importlib.import_module(DB_MODULE_NAME)
def test_database_module_defaults_to_sqlite(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
from landppt.core.config import AppConfig
config = AppConfig()
assert config.database_url == "sqlite:///./landppt.db"
def test_should_fallback_for_legacy_postgres_connectivity_error(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
module = _reload_database_module(monkeypatch)
err = OperationalError(
"SELECT 1",
{},
Exception('could not translate host name "postgres" to address: Name or service not known'),
)
assert module._should_fallback_to_sqlite(err, LEGACY_POSTGRES_URL) is True
def test_should_not_fallback_for_explicit_custom_postgres_connectivity_error(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
module = _reload_database_module(monkeypatch)
custom_url = "postgresql://landppt:landppt@db.internal:5432/landppt"
err = OperationalError(
"SELECT 1",
{},
Exception("connection refused"),
)
assert module._should_fallback_to_sqlite(err, custom_url) is False
@pytest.mark.asyncio
async def test_startup_initialization_runs_in_order(monkeypatch):
sys.modules.pop(MAIN_MODULE_NAME, None)
import landppt.main as main_module
calls = []
async def fake_run_startup_initialization():
calls.append("run_startup_initialization")
return True
monkeypatch.setattr(main_module, "run_startup_initialization", fake_run_startup_initialization)
await main_module.startup_event()
assert calls == ["run_startup_initialization"]
|