from __future__ import annotations from datetime import timedelta import pytest from httpx import AsyncClient from app.services.auth_service import create_access_token from conftest import TestUserData, create_test_user, login_payload pytestmark = [pytest.mark.asyncio, pytest.mark.auth] async def test_register_success_creates_user(client: AsyncClient, db_session) -> None: """Successful registration should create a user and hide password data.""" user_data = { "email": "newuser@example.com", "password": "SecurePass123", "role": "viewer", "full_name": "New User", } response = await client.post("/api/auth/register", json=user_data) assert response.status_code == 201, f"Expected successful registration, got {response.text}" payload = response.json() assert payload["email"] == user_data["email"], "Registered email should match the request payload" assert payload["role"] == "viewer", "Public registration should always create viewer accounts" assert "password" not in payload, "Registration response must not expose the password" async def test_register_duplicate_email_returns_400(client: AsyncClient, viewer_user: TestUserData) -> None: """Registering an existing email should return a conflict-style validation error.""" response = await client.post( "/api/auth/register", json={"email": viewer_user.user.email, "password": "AnotherPass123", "role": "viewer"}, ) assert response.status_code == 400, f"Duplicate email should be rejected, got {response.text}" assert response.json()["detail"] == "Email already registered", "Duplicate email message should be explicit" @pytest.mark.parametrize("password", ["short", "allletters", "12345678"]) async def test_register_weak_password_returns_422(client: AsyncClient, password: str) -> None: """Weak passwords should fail schema validation.""" response = await client.post( "/api/auth/register", json={"email": f"{password}@example.com", "password": password}, ) assert response.status_code == 422, f"Weak password {password!r} should fail validation" assert response.json()["detail"] == "Validation error", "Weak password should use the normalized 422 format" async def test_register_invalid_email_format_returns_422(client: AsyncClient) -> None: """Invalid email addresses should be rejected by Pydantic.""" response = await client.post( "/api/auth/register", json={"email": "not-an-email", "password": "SecurePass123"}, ) assert response.status_code == 422, f"Invalid email should return 422, got {response.text}" assert response.json()["errors"][0]["field"] == "email", "Validation error should point at the email field" async def test_register_missing_required_fields_returns_422(client: AsyncClient) -> None: """Missing required registration fields should return the normalized error payload.""" response = await client.post("/api/auth/register", json={}) assert response.status_code == 422, f"Missing registration fields should return 422, got {response.text}" fields = {error["field"] for error in response.json()["errors"]} assert {"email", "password"}.issubset(fields), "Both email and password fields should be reported missing" async def test_register_default_role_is_viewer(client: AsyncClient) -> None: """Omitting the role should create a viewer by default.""" response = await client.post( "/api/auth/register", json={"email": "defaultrole@example.com", "password": "SecurePass123"}, ) assert response.status_code == 201, f"Registration without an explicit role should succeed, got {response.text}" assert response.json()["role"] == "viewer", "Default role should be viewer" async def test_cannot_register_as_admin(client: AsyncClient) -> None: """Registration should ignore privileged role requests and keep the account as viewer.""" response = await client.post( "/api/auth/register", json={"email": "hacker@example.com", "password": "SecurePass123", "role": "admin"}, ) assert response.status_code == 201, f"Role escalation attempt should not fail registration, got {response.text}" assert response.json()["role"] == "viewer", "Privileged self-registration should be downgraded to viewer" async def test_login_success_returns_token(client: AsyncClient, viewer_user: TestUserData) -> None: """Valid credentials should return a bearer token.""" response = await client.post("/api/auth/login", data=login_payload(viewer_user.user.email, viewer_user.password)) assert response.status_code == 200, f"Expected login success, got {response.text}" payload = response.json() assert payload["token_type"] == "bearer", "Token type should be bearer" assert payload["access_token"], "Access token should be present in the login response" @pytest.mark.parametrize( ("email", "password"), [("viewer@example.com", "WrongPass123"), ("missing@example.com", "WrongPass123")], ) async def test_login_invalid_credentials_returns_401( client: AsyncClient, viewer_user: TestUserData, email: str, password: str, ) -> None: """Incorrect or unknown credentials should return 401.""" response = await client.post("/api/auth/login", data=login_payload(email, password)) assert response.status_code == 401, f"Invalid credentials should return 401, got {response.text}" assert response.json()["detail"] == "Invalid credentials", "Invalid login should use the expected error message" async def test_login_inactive_user_returns_401(client: AsyncClient, db_session) -> None: """Inactive users should not be able to log in.""" inactive_user = create_test_user( db_session, email="inactive@example.com", password="InactivePass123", is_active=False, ) response = await client.post("/api/auth/login", data=login_payload(inactive_user.user.email, inactive_user.password)) assert response.status_code == 401, f"Inactive users should be blocked from login, got {response.text}" assert response.json()["detail"] == "User account is inactive", "Inactive login should explain the failure" async def test_token_contains_valid_claims(client: AsyncClient, viewer_user: TestUserData, decode_jwt) -> None: """Returned JWT tokens should include the subject and expiration claims.""" response = await client.post("/api/auth/login", data=login_payload(viewer_user.user.email, viewer_user.password)) assert response.status_code == 200, f"Expected login success for token claims test, got {response.text}" token = response.json()["access_token"] claims = decode_jwt(token) assert claims["sub"] == viewer_user.user.email, "JWT subject should be the user email" assert claims["user_id"] == viewer_user.user.id, "JWT should include the user id" assert claims["role"] == viewer_user.user.role.value, "JWT should include the current role" assert "exp" in claims, "JWT should contain an expiration claim" async def test_expired_token_returns_401(client: AsyncClient, viewer_user: TestUserData) -> None: """Expired JWTs should be rejected by authenticated endpoints.""" expired_token = create_access_token( subject=viewer_user.user.email, user_id=viewer_user.user.id, role=viewer_user.user.role.value, expires_delta=timedelta(minutes=-5), ) response = await client.get("/api/users/me", headers={"Authorization": f"Bearer {expired_token}"}) assert response.status_code == 401, f"Expired token should return 401, got {response.text}" assert response.json()["detail"] == "Could not validate credentials", "Expired token should use auth failure message" async def test_invalid_token_format_returns_401(client: AsyncClient) -> None: """Malformed tokens should be rejected consistently.""" response = await client.get("/api/users/me", headers={"Authorization": "Bearer not.a.valid.token"}) assert response.status_code == 401, f"Invalid token should return 401, got {response.text}" assert response.json()["detail"] == "Could not validate credentials", "Malformed token should be treated as invalid" async def test_missing_token_returns_401(client: AsyncClient) -> None: """Protected routes should require a bearer token.""" response = await client.get("/api/users/me") assert response.status_code == 401, f"Missing token should return 401, got {response.text}" assert response.json()["detail"] == "Not authenticated", "Missing token should trigger the OAuth2 auth error"