patrickandClaude Sonnet 5
446e4f912d
CI / backend-tests (push) Failing after 1s
Installer (deploy/): - install_server.sh: PostgreSQL+nginx auf Debian 13, idempotente Rolle/DB-Anlage, pgcrypto-Extension, Speicher-Tuning für 4GB-VPS, Zugangsdaten in chmod-600-Datei statt stdout (postgres-expert/owasp-Review) - nginx-Template mit Security-Headern + Rate-Limit auf /auth/login - systemd-Unit-Template mit Sandboxing (NoNewPrivileges/ProtectSystem/PrivateTmp) Backend-Fixes (fastapi-expert-Review): - get_db: einheitliche commit/rollback-Konvention statt Endpunkt-Copy-Paste - Test-Fixtures auf SQLAlchemy-2.0-Savepoint-Pattern umgestellt (join_transaction_mode), da get_db jetzt selbst committet - Lifespan-Handler: Startup-Guard gegen JWT-Secret-Platzhalter, engine.dispose() beim Shutdown - JWT-Payload ohne ungenutztes roles-Claim (Rollen kommen immer frisch aus der DB) Zusätzlich: Subagenten-Definitionen (~/.claude/agents/) auf lauffähiges Modell fixiert. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
import pytest
|
||
import pytest_asyncio
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
from app.core.app_settings import settings
|
||
from app.core.security import hash_password
|
||
from app.db.session import get_db
|
||
from app.main import app
|
||
from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
|
||
|
||
# Erwartet eine bereits per Alembic migrierte Test-Datenbank (CI: install -> migrate -> pytest,
|
||
# siehe testphasen.md Phase 0). Jeder Test läuft in einer Transaktion, die am Ende zurückgerollt
|
||
# wird, damit Tests sich nicht gegenseitig beeinflussen.
|
||
engine = create_async_engine(settings.database_url, pool_pre_ping=True)
|
||
TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_session():
|
||
"""Jeder Test läuft in einer äußeren Transaktion, die am Ende zurückgerollt wird.
|
||
|
||
Die Session ist per `join_transaction_mode="create_savepoint"` an die äußere
|
||
Connection gebunden: ein `commit()` innerhalb des Tests (z. B. durch `get_db`,
|
||
das jetzt selbst committet) schließt nur eine SAVEPOINT ab, nicht die äußere
|
||
Transaktion – das äußere `connection.rollback()` verwirft am Ende trotzdem alles.
|
||
(SQLAlchemy 2.0 "Joining a Session into an External Transaction" – Review-Fund
|
||
fastapi-expert: ohne dieses Pattern hätte ein commit() in Endpunkten die
|
||
Testisolation gebrochen.)
|
||
"""
|
||
async with engine.connect() as connection:
|
||
await connection.begin()
|
||
session = AsyncSession(
|
||
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
|
||
)
|
||
try:
|
||
yield session
|
||
finally:
|
||
await session.close()
|
||
await connection.rollback()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def client(db_session):
|
||
async def _get_db_override():
|
||
yield db_session
|
||
|
||
app.dependency_overrides[get_db] = _get_db_override
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
yield ac
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def mitarbeiter_user(db_session):
|
||
benutzer = Benutzer(
|
||
name="Test Mitarbeiter",
|
||
login="mitarbeiter1",
|
||
passwort_hash=hash_password("test-passwort-123"),
|
||
aktiv=True,
|
||
)
|
||
db_session.add(benutzer)
|
||
await db_session.flush()
|
||
db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.mitarbeiter))
|
||
await db_session.flush()
|
||
return benutzer
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def admin_user(db_session):
|
||
benutzer = Benutzer(
|
||
name="Test Administration",
|
||
login="admin1",
|
||
passwort_hash=hash_password("test-passwort-123"),
|
||
aktiv=True,
|
||
)
|
||
db_session.add(benutzer)
|
||
await db_session.flush()
|
||
db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.administration))
|
||
await db_session.flush()
|
||
return benutzer
|