moneyfy/backend/tests/test_auth.py

"""Tests der Anmeldung, Token-Rotation und Zugriffsbeschränkung."""

from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.cookies import ACCESS_COOKIE, REFRESH_COOKIE
from app.core.security import hash_password
from app.models import AppUser, RefreshToken
from tests.conftest import TEST_PASSWORD


async def test_anmeldung_setzt_beide_cookies(client: AsyncClient, user: AppUser) -> None:
    response = await client.post(
        "/api/auth/login", json={"username": "tester", "password": TEST_PASSWORD}
    )

    assert response.status_code == 200
    assert response.json()["username"] == "tester"
    assert ACCESS_COOKIE in response.cookies
    assert REFRESH_COOKIE in response.cookies

    # Beide Cookies sind httpOnly und auf SameSite=Lax gesetzt.
    header = "; ".join(response.headers.get_list("set-cookie"))
    assert header.count("HttpOnly") == 2
    assert header.count("SameSite=lax") == 2


async def test_anmeldung_ist_unabhaengig_von_der_gross_schreibung(
    client: AsyncClient, user: AppUser
) -> None:
    response = await client.post(
        "/api/auth/login", json={"username": "TESTER", "password": TEST_PASSWORD}
    )
    assert response.status_code == 200


async def test_falsches_passwort_wird_abgewiesen(client: AsyncClient, user: AppUser) -> None:
    response = await client.post(
        "/api/auth/login", json={"username": "tester", "password": "falsch"}
    )

    assert response.status_code == 401
    assert response.json()["code"] == "invalid_credentials"
    assert ACCESS_COOKIE not in response.cookies


async def test_unbekannter_benutzer_erhaelt_dieselbe_meldung(client: AsyncClient) -> None:
    response = await client.post(
        "/api/auth/login", json={"username": "gibtesnicht", "password": "egal"}
    )

    assert response.status_code == 401
    assert response.json()["code"] == "invalid_credentials"


async def test_geschuetzte_route_ohne_anmeldung(client: AsyncClient) -> None:
    response = await client.get("/api/accounts")

    assert response.status_code == 401
    assert response.json()["code"] == "not_authenticated"


async def test_geschuetzte_route_mit_anmeldung(auth_client: AsyncClient) -> None:
    assert (await auth_client.get("/api/accounts")).status_code == 200


async def test_me_liefert_den_angemeldeten_benutzer(auth_client: AsyncClient) -> None:
    body = (await auth_client.get("/api/me")).json()

    assert body["username"] == "tester"
    assert body["email"] == "tester@example.com"
    assert body["must_change_password"] is False
    assert "password_hash" not in body


async def test_systemendpunkte_bleiben_offen(client: AsyncClient) -> None:
    assert (await client.get("/api/health")).status_code == 200
    assert (await client.get("/api/version")).status_code == 200


async def test_refresh_rotiert_das_token(auth_client: AsyncClient, session: AsyncSession) -> None:
    altes_token = auth_client.cookies[REFRESH_COOKIE]

    response = await auth_client.post("/api/auth/refresh")

    assert response.status_code == 200
    assert auth_client.cookies[REFRESH_COOKIE] != altes_token

    tokens = (await session.execute(select(RefreshToken))).scalars().all()
    assert len(tokens) == 2
    assert sum(1 for token in tokens if token.revoked_at is None) == 1


async def test_wiederverwendetes_refresh_token_beendet_alle_sitzungen(
    auth_client: AsyncClient, session: AsyncSession
) -> None:
    altes_token = auth_client.cookies[REFRESH_COOKIE]
    await auth_client.post("/api/auth/refresh")

    # Das bereits verbrauchte Token noch einmal einlösen.
    auth_client.cookies.set(REFRESH_COOKIE, altes_token, path="/api/auth")
    response = await auth_client.post("/api/auth/refresh")

    assert response.status_code == 401
    assert response.json()["code"] == "token_reused"

    tokens = (await session.execute(select(RefreshToken))).scalars().all()
    assert all(token.revoked_at is not None for token in tokens)


async def test_refresh_ohne_cookie(client: AsyncClient) -> None:
    response = await client.post("/api/auth/refresh")

    assert response.status_code == 401
    assert response.json()["code"] == "not_authenticated"


async def test_abmelden_loescht_die_cookies(auth_client: AsyncClient) -> None:
    response = await auth_client.post("/api/auth/logout")

    assert response.status_code == 200
    assert not auth_client.cookies.get(ACCESS_COOKIE)
    assert (await auth_client.get("/api/accounts")).status_code == 401


async def test_passwortwechsel_beendet_alle_sitzungen(
    auth_client: AsyncClient, session: AsyncSession, user: AppUser
) -> None:
    response = await auth_client.post(
        "/api/auth/change-password",
        json={"current_password": TEST_PASSWORD, "new_password": "noch-sicherer-456"},
    )

    assert response.status_code == 200
    tokens = (await session.execute(select(RefreshToken))).scalars().all()
    assert all(token.revoked_at is not None for token in tokens)

    # Anmeldung nur noch mit dem neuen Passwort.
    assert (
        await auth_client.post(
            "/api/auth/login", json={"username": "tester", "password": TEST_PASSWORD}
        )
    ).status_code == 401
    assert (
        await auth_client.post(
            "/api/auth/login", json={"username": "tester", "password": "noch-sicherer-456"}
        )
    ).status_code == 200


async def test_passwortwechsel_prueft_das_alte_passwort(auth_client: AsyncClient) -> None:
    response = await auth_client.post(
        "/api/auth/change-password",
        json={"current_password": "falsch", "new_password": "noch-sicherer-456"},
    )

    assert response.status_code == 401
    assert response.json()["code"] == "invalid_credentials"


async def test_zu_kurzes_passwort_wird_abgewiesen(auth_client: AsyncClient) -> None:
    response = await auth_client.post(
        "/api/auth/change-password",
        json={"current_password": TEST_PASSWORD, "new_password": "kurz"},
    )

    assert response.status_code == 422
    assert response.json()["code"] == "validation_error"


async def test_gleiches_passwort_wird_abgewiesen(auth_client: AsyncClient) -> None:
    response = await auth_client.post(
        "/api/auth/change-password",
        json={"current_password": TEST_PASSWORD, "new_password": TEST_PASSWORD},
    )

    assert response.status_code == 422
    assert response.json()["code"] == "password_unchanged"


async def test_erzwungener_passwortwechsel_sperrt_die_fachrouten(
    client: AsyncClient, session: AsyncSession
) -> None:
    session.add(
        AppUser(
            username="neuling",
            password_hash=hash_password("start-passwort-1"),
            must_change_password=True,
        )
    )
    await session.flush()

    await client.post(
        "/api/auth/login", json={"username": "neuling", "password": "start-passwort-1"}
    )

    # /api/me bleibt erreichbar, damit die Oberfläche den Zustand erkennt.
    me = await client.get("/api/me")
    assert me.status_code == 200
    assert me.json()["must_change_password"] is True

    gesperrt = await client.get("/api/accounts")
    assert gesperrt.status_code == 403
    assert gesperrt.json()["code"] == "password_change_required"

    # Nach dem Wechsel ist der Zugriff frei.
    await client.post(
        "/api/auth/change-password",
        json={"current_password": "start-passwort-1", "new_password": "richtig-sicher-9"},
    )
    await client.post(
        "/api/auth/login", json={"username": "neuling", "password": "richtig-sicher-9"}
    )
    assert (await client.get("/api/accounts")).status_code == 200