moneyfyandClaude Opus 5
b586d27b77
- Anmeldung über Argon2id und JWT in httpOnly-Cookies, Refresh mit echter Rotation über die neue Tabelle refresh_token - AuthProvider-Protokoll als Vorbereitung für OIDC, Administrator-Anlage beim Erststart mit erzwungenem Passwortwechsel - CRUD für Konten, Kategorien (zweistufiger Baum), Firmen, Recurrences, Preisversionen, Buchungen, Budgets, Vorlagen und Sparziele - Fälligkeiten mit Overlay-Logik: abrufen, bestätigen, auslassen, zurücksetzen - Kontosalden zum Stichtag, Monatsübersicht mit Plan-Ist-Vergleich - SECRET_KEY jetzt mindestens 32 Zeichen; Platzhalter in Produktion abgelehnt - 61 neue Integrationstests, insgesamt 148 grün Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
"""Gemeinsame Test-Fixtures. Nutzt eine separate Testdatenbank."""
|
||
|
||
import os
|
||
|
||
# Muss vor dem ersten Import der Anwendung gesetzt sein, da die Settings gecacht werden.
|
||
os.environ.setdefault(
|
||
"DATABASE_URL",
|
||
os.environ.get(
|
||
"TEST_DATABASE_URL",
|
||
"postgresql+asyncpg://moneyfy:moneyfy@127.0.0.1:5432/moneyfy_test",
|
||
),
|
||
)
|
||
os.environ.setdefault("SECRET_KEY", "test-secret-key-mindestens-32-zeichen-lang")
|
||
os.environ.setdefault("ENVIRONMENT", "test")
|
||
|
||
from collections.abc import AsyncGenerator
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import (
|
||
AsyncSession,
|
||
async_sessionmaker,
|
||
create_async_engine,
|
||
)
|
||
|
||
from app.core.config import settings
|
||
from app.core.security import hash_password
|
||
from app.db.session import get_session
|
||
from app.main import app as fastapi_app
|
||
from app.models import Account, AppUser, Base, Category
|
||
from app.models.enums import AccountType
|
||
from app.services.seed import seed_all
|
||
|
||
TEST_PASSWORD = "sicher-genug-123"
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def anyio_backend() -> str:
|
||
return "asyncio"
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
async def engine():
|
||
"""Legt das Schema einmal pro Testlauf frisch an."""
|
||
test_engine = create_async_engine(str(settings.database_url), poolclass=None)
|
||
async with test_engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.drop_all)
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
yield test_engine
|
||
await test_engine.dispose()
|
||
|
||
|
||
@pytest.fixture
|
||
async def session(engine) -> AsyncGenerator[AsyncSession, None]:
|
||
"""Eine Session je Test, am Ende wird zurückgerollt."""
|
||
connection = await engine.connect()
|
||
transaction = await connection.begin()
|
||
# `create_savepoint` sorgt dafür, dass session.commit() nur den Savepoint freigibt
|
||
# und der abschließende Rollback wirklich alle Testdaten entfernt.
|
||
maker = async_sessionmaker(
|
||
bind=connection,
|
||
expire_on_commit=False,
|
||
autoflush=False,
|
||
join_transaction_mode="create_savepoint",
|
||
)
|
||
async with maker() as db_session:
|
||
yield db_session
|
||
await transaction.rollback()
|
||
await connection.close()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client(session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
|
||
"""HTTP-Client gegen die App, alle Requests laufen in der Test-Transaktion."""
|
||
|
||
async def _override() -> AsyncGenerator[AsyncSession, None]:
|
||
yield session
|
||
|
||
fastapi_app.dependency_overrides[get_session] = _override
|
||
transport = ASGITransport(app=fastapi_app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as http_client:
|
||
yield http_client
|
||
fastapi_app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest.fixture
|
||
async def user(session: AsyncSession) -> AppUser:
|
||
"""Ein angemeldefähiger Benutzer mit bekanntem Passwort."""
|
||
account = AppUser(
|
||
username="tester",
|
||
email="tester@example.com",
|
||
password_hash=hash_password(TEST_PASSWORD),
|
||
must_change_password=False,
|
||
)
|
||
session.add(account)
|
||
await session.flush()
|
||
return account
|
||
|
||
|
||
@pytest.fixture
|
||
async def auth_client(client: AsyncClient, user: AppUser) -> AsyncClient:
|
||
"""Bereits angemeldeter Client – die Cookies bleiben am Client hängen."""
|
||
response = await client.post(
|
||
"/api/auth/login", json={"username": user.username, "password": TEST_PASSWORD}
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
return client
|
||
|
||
|
||
@pytest.fixture
|
||
async def seeded(session: AsyncSession) -> dict[str, int]:
|
||
"""Kategoriebaum plus ein Konto – die Grundlage der meisten Integrationstests."""
|
||
await seed_all(session)
|
||
|
||
account = Account(
|
||
name="Girokonto",
|
||
type=AccountType.CHECKING,
|
||
opening_balance=Decimal("1000.00"),
|
||
opening_balance_date=date(2026, 1, 1),
|
||
)
|
||
session.add(account)
|
||
await session.flush()
|
||
|
||
async def category_id(name: str) -> int:
|
||
stmt = select(Category).where(Category.name == name)
|
||
return (await session.execute(stmt)).scalars().first().id
|
||
|
||
return {
|
||
"account_id": account.id,
|
||
"streaming": await category_id("Streaming"),
|
||
"miete": await category_id("Miete"),
|
||
"kredite": await category_id("Kredite"),
|
||
"gehalt": await category_id("Gehalt"),
|
||
"lebensmittel": await category_id("Lebensmittel"),
|
||
}
|