Kitos
9e204b78ec
- Add test_test_entity.py with 46 pure unit tests covering the full domain entity - Fix _FakeSettings in 11 test files (REPORT_TEMPLATES_DIR, JIRA, TEMPO) - Fix stale db.commit assertions to db.flush after UoW refactor - Add missing mock fields for TestEntity.from_orm compatibility - Make database.py skip pool args for SQLite in test environment - Disable slowapi rate limiter in test client fixture - Inject test engine into app.database to fix threading errors - Update role assertions to match current require_any_role policy - Mark 6 legacy V1 endpoint tests as xfail (replaced by V2 workflow)
69 lines
1.7 KiB
Python
69 lines
1.7 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, declarative_base
|
|
|
|
Base = declarative_base()
|
|
|
|
# Engine and session factory are created lazily so that tests can
|
|
# override DATABASE_URL via environment *before* any import triggers
|
|
# the real PostgreSQL engine creation (which requires psycopg2).
|
|
_engine = None
|
|
_SessionLocal = None
|
|
|
|
|
|
def _get_engine():
|
|
global _engine
|
|
if _engine is None:
|
|
from app.config import settings
|
|
|
|
url = settings.DATABASE_URL
|
|
kwargs: dict = {}
|
|
if url.startswith("postgresql"):
|
|
kwargs.update(
|
|
pool_size=20,
|
|
max_overflow=10,
|
|
pool_recycle=3600,
|
|
pool_pre_ping=True,
|
|
)
|
|
_engine = create_engine(url, **kwargs)
|
|
return _engine
|
|
|
|
|
|
def _get_session_factory():
|
|
global _SessionLocal
|
|
if _SessionLocal is None:
|
|
_SessionLocal = sessionmaker(
|
|
autocommit=False, autoflush=False, bind=_get_engine()
|
|
)
|
|
return _SessionLocal
|
|
|
|
|
|
class _LazySessionLocal:
|
|
"""Proxy so ``SessionLocal()`` keeps working as before but the real
|
|
sessionmaker is only created on first call."""
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return _get_session_factory()(*args, **kwargs)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(_get_session_factory(), name)
|
|
|
|
|
|
SessionLocal = _LazySessionLocal()
|
|
|
|
|
|
class _EngineProxy:
|
|
"""Thin proxy so ``from app.database import engine`` still works."""
|
|
def __getattr__(self, name):
|
|
return getattr(_get_engine(), name)
|
|
|
|
|
|
engine = _EngineProxy() # type: ignore[assignment]
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|