kmtnflower/backend/tests/test_admin_users.py

"""Tests for /api/admin/users admin user management endpoints."""

import pytest
from httpx import AsyncClient, ASGITransport
from jose import jwt
from datetime import datetime, timedelta, timezone

from app.auth import hash_password
from app.config import settings
from app.database import async_session
from app.main import app
from app.user_models import User, Role


@pytest.fixture
async def admin_token():
    """Create a valid admin JWT."""
    expire = datetime.now(timezone.utc) + timedelta(minutes=30)
    payload = {
        "sub": "99999",
        "is_admin": True,
        "exp": expire,
    }
    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")


@pytest.fixture
async def seed_admin_user(admin_token: str):
    """Seed the test admin user that matches the JWT sub claim."""
    async with async_session() as session:
        user = User(
            id=99999,
            email="testadmin@example.com",
            display_name="Test Admin",
            auth_provider="local",
            is_admin=True,
        )
        session.add(user)
        await session.commit()
    return user


@pytest.fixture
async def client():
    """Create an AsyncClient for the FastAPI app."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_list_admin_users_empty(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """GET /api/admin/users returns empty list when only JWT user (no others) exists."""
    resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"})
    assert resp.status_code == 200
    # The seed_admin_user IS an admin, so it shows up
    data = resp.json()
    assert len(data) == 1
    assert data[0]["email"] == "testadmin@example.com"


@pytest.mark.asyncio
async def test_list_admin_users_filters_non_admins(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """GET /api/admin/users returns only users with admin access."""
    # Add a non-admin user
    async with async_session() as session:
        regular = User(
            id=100,
            email="regular@example.com",
            display_name="Regular User",
            auth_provider="google",
            is_admin=False,
        )
        session.add(regular)
        await session.commit()

    resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    assert data[0]["email"] == "testadmin@example.com"


@pytest.mark.asyncio
async def test_create_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """POST /api/admin/users creates a new admin with hashed password."""
    resp = await client.post(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={
            "email": "newadmin@example.com",
            "display_name": "New Admin",
            "password": "securepass123",
        },
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["email"] == "newadmin@example.com"
    assert data["display_name"] == "New Admin"
    assert data["is_admin"] is True


@pytest.mark.asyncio
async def test_create_admin_user_duplicate_email(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """POST /api/admin/users returns 409 for duplicate email."""
    # Create first user
    await client.post(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={
            "email": "dup@example.com",
            "display_name": "Dup Admin",
            "password": "pass123",
        },
    )

    # Attempt duplicate
    resp = await client.post(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={
            "email": "dup@example.com",
            "display_name": "Dup Admin 2",
            "password": "pass123",
        },
    )
    assert resp.status_code == 409


@pytest.mark.asyncio
async def test_delete_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """DELETE /api/admin/users/{id} removes the user."""
    # Create another admin via the API
    await client.post(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={
            "email": "otheradmin@example.com",
            "display_name": "Other Admin",
            "password": "pass123",
        },
    )

    resp = await client.delete(
        "/api/admin/users/88888",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    # The new user won't have id=88888 in the test DB; we need to get the actual ID
    # Let's list users first to find the ID
    list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"})
    users = list_resp.json()
    other_user = next((u for u in users if u["email"] == "otheradmin@example.com"), None)
    if other_user:
        resp = await client.delete(
            f"/api/admin/users/{other_user['id']}",
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        assert resp.status_code == 204

    # Verify only seed admin remains
    list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"})
    remaining = list_resp.json()
    assert len(remaining) == 1
    assert remaining[0]["email"] == "testadmin@example.com"


@pytest.mark.asyncio
async def test_delete_self_admin_user_forbidden(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """DELETE /api/admin/users/{id} blocks deleting yourself."""
    resp = await client.delete(
        "/api/admin/users/99999",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 400
    assert "own account" in resp.json()["detail"].lower()


@pytest.mark.asyncio
async def test_delete_last_admin_forbidden(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """DELETE /api/admin/users/{id} blocks if it's the last admin."""
    # Try to delete the only admin (our seed user) - self-delete is blocked
    resp = await client.delete(
        "/api/admin/users/99999",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_update_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """PUT /api/admin/users/{id} updates display_name."""
    resp = await client.put(
        "/api/admin/users/99999",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={"display_name": "Updated Name"},
    )
    assert resp.status_code == 200
    assert resp.json()["display_name"] == "Updated Name"


@pytest.mark.asyncio
async def test_admin_users_requires_auth(client: AsyncClient):
    """GET /api/admin/users returns 401 without auth headers."""
    resp = await client.get("/api/admin/users")
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_create_admin_assigns_admin_role(client: AsyncClient, admin_token: str, seed_admin_user: User):
    """POST /api/admin/users assigns the 'admin' role and sets is_admin=True."""
    resp = await client.post(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={
            "email": "roletest@example.com",
            "display_name": "Role Test",
            "password": "pass123",
        },
    )
    assert resp.status_code == 201
    data = resp.json()
    assert data["is_admin"] is True

    # Verify via list endpoint that the user shows up as admin
    list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"})
    users = list_resp.json()
    role_user = next((u for u in users if u["email"] == "roletest@example.com"), None)
    assert role_user is not None
    assert role_user["is_admin"] is True