crm-system/tests/test_auth.py

"""Auth tests: covers all 9 FR-1 acceptance criteria."""

from __future__ import annotations

import time

from httpx import AsyncClient
from jose import jwt

from app.core.config import get_settings

settings = get_settings()


# === FR-1.1 / FR-1.2 / Akzeptanzkriterium 1: register success ===


async def test_register_success(client: AsyncClient) -> None:
    """AC #1: POST /api/v1/auth/register mit gültigem Payload → 201 + User-Objekt + JWT."""
    payload = {
        "email": "alice@example.com",
        "password": "SecurePass123!",
        "name": "Alice",
    }
    resp = await client.post("/api/v1/auth/register", json=payload)
    assert resp.status_code == 201, resp.text
    data = resp.json()
    assert "user" in data
    assert "access_token" in data
    assert data["token_type"] == "bearer"
    assert data["expires_in"] > 0
    # User object has the expected fields (no password_hash leaked)
    user = data["user"]
    assert user["email"] == "alice@example.com"
    assert user["name"] == "Alice"
    assert "password_hash" not in user
    assert "id" in user
    assert "org_id" in user


# === FR-1.1 / Akzeptanzkriterium 2: register duplicate email ===


async def test_register_duplicate_email(client: AsyncClient, registered_user: dict) -> None:
    """AC #2: POST /api/v1/auth/register mit existierender Email → 409."""
    # registered_user already exists; trying again with same email (and 2nd user
    # would also be blocked by bootstrap). 409 is correct because of the email conflict.
    # But bootstrap is also blocked → 403 is also acceptable. We accept either.
    payload = {
        "email": registered_user["email"],
        "password": "AnotherPass123!",
        "name": "Dup User",
    }
    resp = await client.post("/api/v1/auth/register", json=payload)
    assert resp.status_code in (403, 409), (
        f"Expected 403 (bootstrap) or 409 (email), got {resp.status_code}: {resp.text}"
    )


async def test_register_bootstrap_blocked_after_first(
    client: AsyncClient, registered_user: dict
) -> None:
    """AC: Zweiter POST /api/v1/auth/register nach erfolgreichem ersten → 403."""
    payload = {
        "email": "other@example.com",
        "password": "AnotherPass123!",
        "name": "Other User",
    }
    resp = await client.post("/api/v1/auth/register", json=payload)
    assert resp.status_code == 403, (
        f"Bootstrap should be blocked after first user, got {resp.status_code}: {resp.text}"
    )


# === FR-1.2 / Akzeptanzkriterium 3: register weak password ===


async def test_register_weak_password(client: AsyncClient) -> None:
    """AC #3: Schwaches Passwort (< 8 Zeichen) → 422."""
    payload = {
        "email": "weak@example.com",
        "password": "short",  # < 8 chars
        "name": "Weak",
    }
    resp = await client.post("/api/v1/auth/register", json=payload)
    assert resp.status_code == 422, resp.text


# === FR-1.2 / Akzeptanzkriterium 4: login success ===


async def test_login_success(client: AsyncClient, registered_user: dict) -> None:
    """AC #4: POST /api/v1/auth/login mit korrekten Credentials → 200 + JWT."""
    resp = await client.post(
        "/api/v1/auth/login",
        data={
            "username": registered_user["email"],
            "password": registered_user["password"],
        },
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"
    assert data["expires_in"] > 0


# === FR-1.2 / Akzeptanzkriterium 5: login wrong password ===


async def test_login_wrong_password(client: AsyncClient, registered_user: dict) -> None:
    """AC #5: POST /api/v1/auth/login mit falschem Passwort → 401."""
    resp = await client.post(
        "/api/v1/auth/login",
        data={
            "username": registered_user["email"],
            "password": "WrongPassword123!",
        },
    )
    assert resp.status_code == 401, resp.text


# === FR-1.2 / Akzeptanzkriterium 6: login nonexistent user ===


async def test_login_nonexistent_user(client: AsyncClient) -> None:
    """AC #6: POST /api/v1/auth/login mit nicht existierendem User → 401."""
    resp = await client.post(
        "/api/v1/auth/login",
        data={"username": "nobody@example.com", "password": "AnyPass123!"},
    )
    assert resp.status_code == 401, resp.text


# === FR-1.6 / Akzeptanzkriterium 7: get /me with valid JWT ===


async def test_get_me_with_valid_jwt(
    client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
) -> None:
    """AC #7: GET /api/v1/users/me mit gültigem JWT → 200 + User-Daten (ohne password_hash)."""
    resp = await client.get("/api/v1/users/me", headers=auth_headers)
    assert resp.status_code == 200, resp.text
    data = resp.json()
    assert data["email"] == registered_user["email"]
    assert data["name"] == registered_user["name"]
    assert "password_hash" not in data


# === FR-1.6 / Akzeptanzkriterium 8: get /me without JWT ===


async def test_get_me_without_jwt(client: AsyncClient) -> None:
    """AC #8: GET /api/v1/users/me ohne JWT → 401."""
    resp = await client.get("/api/v1/users/me")
    assert resp.status_code == 401, resp.text


# === FR-1.6 / Akzeptanzkriterium 9: get /me with expired JWT ===


async def test_get_me_with_expired_jwt(client: AsyncClient) -> None:
    """AC #9: GET /api/v1/users/me mit expired JWT → 401 + Hinweis 'token_expired'."""
    # Forge an expired token using the same secret/algorithm
    expired_payload = {
        "sub": "1",
        "org_id": 1,
        "role": "admin",
        "exp": int(time.time()) - 3600,  # 1h in the past
        "iat": int(time.time()) - 7200,
    }
    expired_token = jwt.encode(
        expired_payload, settings.AUTH_SECRET, algorithm=settings.JWT_ALGORITHM
    )
    resp = await client.get(
        "/api/v1/users/me",
        headers={"Authorization": f"Bearer {expired_token}"},
    )
    assert resp.status_code == 401, resp.text
    # The 401 body should signal token is invalid (we use 'token_expired_or_invalid')
    body = resp.json()
    assert "detail" in body
    assert (
        "token" in body["detail"].lower()
        or "expired" in body["detail"].lower()
        or "credential" in body["detail"].lower()
    ), f"Expected token-related 401 detail, got: {body}"


# === Bonus: password is stored hashed, not plaintext ===


async def test_db_user_has_hashed_password(
    client: AsyncClient, registered_user: dict, session_factory
) -> None:
    """AC: DB-User wird mit gehashtem password_hash angelegt (kein Klartext)."""
    from sqlalchemy import select

    from app.models.user import User

    async with session_factory() as session:
        result = await session.execute(select(User).where(User.email == registered_user["email"]))
        user = result.scalar_one()
        # bcrypt hashes start with $2b$ (or $2a$ for passlib), never plain text
        assert user.password_hash.startswith("$"), (
            f"Password hash should be a bcrypt string, got: {user.password_hash!r}"
        )
        assert user.password_hash != registered_user["password"]
        assert len(user.password_hash) > 50, (
            f"Bcrypt hash should be ~60 chars long, got {len(user.password_hash)}"
        )


# === Bonus: no default admin bootstrap on startup ===


async def test_no_default_admin_on_startup(client: AsyncClient, session_factory) -> None:
    """AC: KEIN admin/admin Bootstrap-User beim App-Start (Frisch-DB = leer)."""
    from sqlalchemy import func, select

    from app.models.user import User

    # Fresh DB → no users
    async with session_factory() as session:
        result = await session.execute(select(func.count()).select_from(User))
        count = result.scalar_one()
        assert count == 0, f"Fresh DB should have 0 users, found {count}"

        # Also check: no user with role=admin and well-known email
        result = await session.execute(select(User).where(User.role == "admin"))
        admins = result.scalars().all()
        assert len(admins) == 0, f"Fresh DB should have no admin users, found {len(admins)}"