image_management_api/tests/api/test_users.py

import pytest
from fastapi.testclient import TestClient
from bson import ObjectId

from src.models.user import UserModel
from src.models.team import TeamModel
from src.db.repositories.user_repository import user_repository
from src.db.repositories.team_repository import team_repository


@pytest.mark.asyncio
async def test_create_user(client: TestClient, admin_api_key: tuple, test_team: TeamModel):
    """Test creating a new user (admin only)"""
    raw_key, _ = admin_api_key
    
    # Set up the headers with the admin API key
    headers = {"X-API-Key": raw_key}
    
    # Create a new user
    response = client.post(
        "/api/v1/users",
        headers=headers,
        json={
            "email": "newuser@example.com",
            "name": "New User",
            "team_id": str(test_team.id),
            "is_admin": False
        }
    )
    
    # Check response
    assert response.status_code == 201
    data = response.json()
    assert "id" in data
    assert data["email"] == "newuser@example.com"
    assert data["name"] == "New User"
    assert data["team_id"] == str(test_team.id)
    assert data["is_admin"] is False
    assert "created_at" in data


@pytest.mark.asyncio
async def test_create_user_non_admin(client: TestClient, user_api_key: tuple, test_team: TeamModel):
    """Test that non-admin users cannot create users"""
    raw_key, _ = user_api_key
    
    # Set up the headers with a non-admin API key
    headers = {"X-API-Key": raw_key}
    
    # Try to create a new user
    response = client.post(
        "/api/v1/users",
        headers=headers,
        json={
            "email": "unauthorized@example.com",
            "name": "Unauthorized User",
            "team_id": str(test_team.id),
            "is_admin": False
        }
    )
    
    # Check that the request is forbidden
    assert response.status_code == 403
    assert "detail" in response.json()
    assert "admin" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_list_users_in_team(client: TestClient, admin_api_key: tuple, test_team: TeamModel):
    """Test listing users in a team"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # List users in the team
    response = client.get(
        f"/api/v1/users?team_id={test_team.id}",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert "users" in data
    assert "total" in data
    assert data["total"] >= 1  # Should include at least the admin user


@pytest.mark.asyncio
async def test_list_all_users_admin_only(client: TestClient, admin_api_key: tuple):
    """Test that only admins can list all users across teams"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # List all users (no team filter)
    response = client.get(
        "/api/v1/users",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert "users" in data
    assert "total" in data


@pytest.mark.asyncio
async def test_list_users_non_admin_restricted(client: TestClient, user_api_key: tuple):
    """Test that non-admin users can only see their own team"""
    raw_key, _ = user_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to list all users
    response = client.get(
        "/api/v1/users",
        headers=headers
    )
    
    # Should only return users from their own team
    assert response.status_code == 200
    data = response.json()
    assert "users" in data
    # All returned users should be from the same team
    if data["users"]:
        user_team_ids = set(user["team_id"] for user in data["users"])
        assert len(user_team_ids) == 1  # Only one team represented


@pytest.mark.asyncio
async def test_get_user_by_id(client: TestClient, admin_api_key: tuple, admin_user: UserModel):
    """Test getting a specific user by ID"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Get the user
    response = client.get(
        f"/api/v1/users/{admin_user.id}",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == str(admin_user.id)
    assert data["email"] == admin_user.email
    assert data["name"] == admin_user.name
    assert data["team_id"] == str(admin_user.team_id)
    assert data["is_admin"] == admin_user.is_admin


@pytest.mark.asyncio
async def test_get_user_self(client: TestClient, user_api_key: tuple, regular_user: UserModel):
    """Test that users can get their own information"""
    raw_key, _ = user_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Get own user information
    response = client.get(
        f"/api/v1/users/{regular_user.id}",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == str(regular_user.id)
    assert data["email"] == regular_user.email


@pytest.mark.asyncio
async def test_get_user_other_team_forbidden(client: TestClient, user_api_key: tuple):
    """Test that users cannot access users from other teams"""
    raw_key, _ = user_api_key
    
    # Create a user in another team
    other_team = TeamModel(
        name="Other Team",
        description="Another team for testing"
    )
    created_team = await team_repository.create(other_team)
    
    other_user = UserModel(
        email="other@example.com",
        name="Other User",
        team_id=created_team.id,
        is_admin=False
    )
    created_user = await user_repository.create(other_user)
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to get the other user
    response = client.get(
        f"/api/v1/users/{created_user.id}",
        headers=headers
    )
    
    # Check that the request is forbidden
    assert response.status_code == 403
    assert "detail" in response.json()


@pytest.mark.asyncio
async def test_update_user(client: TestClient, admin_api_key: tuple, regular_user: UserModel):
    """Test updating a user (admin only)"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Update the user
    response = client.put(
        f"/api/v1/users/{regular_user.id}",
        headers=headers,
        json={
            "name": "Updated User Name",
            "email": "updated@example.com",
            "is_admin": True
        }
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == str(regular_user.id)
    assert data["name"] == "Updated User Name"
    assert data["email"] == "updated@example.com"
    assert data["is_admin"] is True
    assert "updated_at" in data


@pytest.mark.asyncio
async def test_update_user_self(client: TestClient, user_api_key: tuple, regular_user: UserModel):
    """Test that users can update their own basic information"""
    raw_key, _ = user_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Update own information (limited fields)
    response = client.put(
        f"/api/v1/users/{regular_user.id}",
        headers=headers,
        json={
            "name": "Self Updated Name",
            "email": "self_updated@example.com"
            # Note: is_admin should not be updatable by self
        }
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Self Updated Name"
    assert data["email"] == "self_updated@example.com"
    # Admin status should remain unchanged
    assert data["is_admin"] == regular_user.is_admin


@pytest.mark.asyncio
async def test_update_user_non_admin_forbidden(client: TestClient, user_api_key: tuple, admin_user: UserModel):
    """Test that non-admin users cannot update other users"""
    raw_key, _ = user_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to update another user
    response = client.put(
        f"/api/v1/users/{admin_user.id}",
        headers=headers,
        json={
            "name": "Unauthorized Update",
            "is_admin": False
        }
    )
    
    # Check that the request is forbidden
    assert response.status_code == 403
    assert "detail" in response.json()


@pytest.mark.asyncio
async def test_delete_user(client: TestClient, admin_api_key: tuple):
    """Test deleting a user (admin only)"""
    raw_key, _ = admin_api_key
    
    # Create a user to delete
    user_to_delete = UserModel(
        email="delete@example.com",
        name="User to Delete",
        team_id=ObjectId(),
        is_admin=False
    )
    created_user = await user_repository.create(user_to_delete)
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Delete the user
    response = client.delete(
        f"/api/v1/users/{created_user.id}",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 204
    
    # Verify the user has been deleted
    deleted_user = await user_repository.get_by_id(created_user.id)
    assert deleted_user is None


@pytest.mark.asyncio
async def test_delete_user_non_admin(client: TestClient, user_api_key: tuple, admin_user: UserModel):
    """Test that non-admin users cannot delete users"""
    raw_key, _ = user_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to delete a user
    response = client.delete(
        f"/api/v1/users/{admin_user.id}",
        headers=headers
    )
    
    # Check that the request is forbidden
    assert response.status_code == 403
    assert "detail" in response.json()
    assert "admin" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_get_user_activity(client: TestClient, admin_api_key: tuple, regular_user: UserModel):
    """Test getting user activity/statistics"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Get user activity
    response = client.get(
        f"/api/v1/users/{regular_user.id}/activity",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert "user_id" in data
    assert "images_uploaded" in data
    assert "last_login" in data
    assert "api_key_usage" in data


@pytest.mark.asyncio
async def test_change_user_team(client: TestClient, admin_api_key: tuple, regular_user: UserModel):
    """Test moving a user to a different team (admin only)"""
    raw_key, _ = admin_api_key
    
    # Create another team
    new_team = TeamModel(
        name="New Team",
        description="A new team for testing"
    )
    created_team = await team_repository.create(new_team)
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Move user to new team
    response = client.put(
        f"/api/v1/users/{regular_user.id}/team",
        headers=headers,
        json={
            "team_id": str(created_team.id)
        }
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["team_id"] == str(created_team.id)


@pytest.mark.asyncio
async def test_user_search(client: TestClient, admin_api_key: tuple):
    """Test searching for users by email or name"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Search for users
    response = client.get(
        "/api/v1/users/search?query=admin",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert "users" in data
    assert "total" in data
    
    # Results should contain users matching the query
    for user in data["users"]:
        assert "admin" in user["email"].lower() or "admin" in user["name"].lower()


@pytest.mark.asyncio
async def test_user_pagination(client: TestClient, admin_api_key: tuple):
    """Test user listing with pagination"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Test pagination
    response = client.get(
        "/api/v1/users?page=1&limit=10",
        headers=headers
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert "users" in data
    assert "pagination" in data
    assert "total" in data["pagination"]
    assert "page" in data["pagination"]
    assert "pages" in data["pagination"]
    assert len(data["users"]) <= 10


@pytest.mark.asyncio
async def test_invalid_user_id(client: TestClient, admin_api_key: tuple):
    """Test handling of invalid user IDs"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to get user with invalid ID
    response = client.get(
        "/api/v1/users/invalid-id",
        headers=headers
    )
    
    # Check that the request returns 400 Bad Request
    assert response.status_code == 400
    assert "detail" in response.json()
    assert "invalid" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_nonexistent_user(client: TestClient, admin_api_key: tuple):
    """Test handling of nonexistent user IDs"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to get nonexistent user
    nonexistent_id = str(ObjectId())
    response = client.get(
        f"/api/v1/users/{nonexistent_id}",
        headers=headers
    )
    
    # Check that the request returns 404 Not Found
    assert response.status_code == 404
    assert "detail" in response.json()
    assert "not found" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_duplicate_email_validation(client: TestClient, admin_api_key: tuple, regular_user: UserModel):
    """Test that duplicate emails are not allowed"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Try to create user with existing email
    response = client.post(
        "/api/v1/users",
        headers=headers,
        json={
            "email": regular_user.email,  # Duplicate email
            "name": "Duplicate Email User",
            "team_id": str(regular_user.team_id),
            "is_admin": False
        }
    )
    
    # Check that the request is rejected
    assert response.status_code == 400
    assert "detail" in response.json()
    assert "email" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_user_role_management(client: TestClient, admin_api_key: tuple, regular_user: UserModel):
    """Test managing user roles and permissions"""
    raw_key, _ = admin_api_key
    
    # Set up the headers
    headers = {"X-API-Key": raw_key}
    
    # Promote user to admin
    response = client.put(
        f"/api/v1/users/{regular_user.id}/role",
        headers=headers,
        json={
            "is_admin": True
        }
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["is_admin"] is True
    
    # Demote user back to regular
    response = client.put(
        f"/api/v1/users/{regular_user.id}/role",
        headers=headers,
        json={
            "is_admin": False
        }
    )
    
    # Check response
    assert response.status_code == 200
    data = response.json()
    assert data["is_admin"] is False