633 lines
21 KiB
Python
633 lines
21 KiB
Python
"""Tests for admin user endpoints."""
|
|
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.models.plan import Plan
|
|
from app.models.user import User
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_user_repository():
|
|
"""Mock user repository."""
|
|
return Mock()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_plan_repository():
|
|
"""Mock plan repository."""
|
|
return Mock()
|
|
|
|
|
|
@pytest.fixture
|
|
def regular_user():
|
|
"""Create regular user for testing."""
|
|
return User(
|
|
id=2,
|
|
email="user@example.com",
|
|
name="Regular User",
|
|
role="user",
|
|
credits=100,
|
|
plan_id=1,
|
|
is_active=True,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def test_plan():
|
|
"""Create test plan."""
|
|
return Plan(
|
|
id=1,
|
|
name="Basic",
|
|
max_credits=100,
|
|
features=["feature1", "feature2"],
|
|
)
|
|
|
|
|
|
class TestAdminUserEndpoints:
|
|
"""Test admin user endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
regular_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test listing users successfully."""
|
|
with patch(
|
|
"app.repositories.user.UserRepository.get_all_with_plan_paginated",
|
|
) as mock_get_all:
|
|
# Create mock user objects that don't trigger database saves
|
|
mock_admin = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": admin_user.id,
|
|
"email": admin_user.email,
|
|
"name": admin_user.name,
|
|
"picture": None,
|
|
"role": admin_user.role,
|
|
"credits": admin_user.credits,
|
|
"is_active": admin_user.is_active,
|
|
"created_at": admin_user.created_at,
|
|
"updated_at": admin_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
|
|
mock_regular = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": regular_user.name,
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": regular_user.credits,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
|
|
# Mock returns tuple (users, total_count)
|
|
mock_get_all.return_value = ([mock_admin, mock_regular], 2)
|
|
|
|
response = await authenticated_admin_client.get("/api/v1/admin/users/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "users" in data
|
|
assert "total" in data
|
|
assert "page" in data
|
|
assert "limit" in data
|
|
assert "total_pages" in data
|
|
assert len(data["users"]) == 2
|
|
assert data["users"][0]["email"] == "admin@example.com"
|
|
assert data["users"][1]["email"] == "user@example.com"
|
|
assert data["total"] == 2
|
|
assert data["page"] == 1
|
|
assert data["limit"] == 50
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_with_pagination(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test listing users with pagination."""
|
|
from app.repositories.user import SortOrder, UserSortField, UserStatus
|
|
|
|
with patch(
|
|
"app.repositories.user.UserRepository.get_all_with_plan_paginated",
|
|
) as mock_get_all:
|
|
mock_admin = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": admin_user.id,
|
|
"email": admin_user.email,
|
|
"name": admin_user.name,
|
|
"picture": None,
|
|
"role": admin_user.role,
|
|
"credits": admin_user.credits,
|
|
"is_active": admin_user.is_active,
|
|
"created_at": admin_user.created_at,
|
|
"updated_at": admin_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
# Mock returns tuple (users, total_count)
|
|
mock_get_all.return_value = ([mock_admin], 1)
|
|
|
|
response = await authenticated_admin_client.get(
|
|
"/api/v1/admin/users/?page=2&limit=10",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "users" in data
|
|
assert data["page"] == 2
|
|
assert data["limit"] == 10
|
|
mock_get_all.assert_called_once_with(
|
|
page=2,
|
|
limit=10,
|
|
search=None,
|
|
sort_by=UserSortField.NAME,
|
|
sort_order=SortOrder.ASC,
|
|
status_filter=UserStatus.ALL,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_unauthenticated(self, client: AsyncClient) -> None:
|
|
"""Test listing users without authentication."""
|
|
response = await client.get("/api/v1/admin/users/")
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_users_non_admin(
|
|
self,
|
|
client: AsyncClient,
|
|
regular_user: User,
|
|
) -> None:
|
|
"""Test listing users as non-admin user."""
|
|
with patch(
|
|
"app.core.dependencies.get_current_active_user",
|
|
return_value=regular_user,
|
|
):
|
|
response = await client.get("/api/v1/admin/users/")
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
regular_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test getting specific user successfully."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
) as mock_get_by_id,
|
|
):
|
|
mock_user = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": regular_user.name,
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": regular_user.credits,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
mock_get_by_id.return_value = mock_user
|
|
|
|
response = await authenticated_admin_client.get("/api/v1/admin/users/2")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == 2
|
|
assert data["email"] == "user@example.com"
|
|
assert data["name"] == "Regular User"
|
|
mock_get_by_id.assert_called_once_with(2)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_not_found(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
) -> None:
|
|
"""Test getting non-existent user."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
return_value=None,
|
|
),
|
|
):
|
|
response = await authenticated_admin_client.get("/api/v1/admin/users/999")
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "User not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
regular_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test updating user successfully."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
) as mock_get_by_id,
|
|
patch("app.repositories.user.UserRepository.update") as mock_update,
|
|
patch(
|
|
"app.repositories.plan.PlanRepository.get_by_id",
|
|
return_value=test_plan,
|
|
),
|
|
):
|
|
mock_user = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": regular_user.name,
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": regular_user.credits,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
|
|
updated_mock = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": "Updated Name",
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": 200,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
|
|
mock_get_by_id.return_value = mock_user
|
|
mock_update.return_value = updated_mock
|
|
|
|
# Mock session.refresh to prevent actual database calls
|
|
async def mock_refresh(instance, attributes=None):
|
|
pass
|
|
|
|
with patch(
|
|
"sqlmodel.ext.asyncio.session.AsyncSession.refresh",
|
|
side_effect=mock_refresh,
|
|
):
|
|
response = await authenticated_admin_client.patch(
|
|
"/api/v1/admin/users/2",
|
|
json={
|
|
"name": "Updated Name",
|
|
"credits": 200,
|
|
"plan_id": 1,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == "Updated Name"
|
|
assert data["credits"] == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user_not_found(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
) -> None:
|
|
"""Test updating non-existent user."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
return_value=None,
|
|
),
|
|
):
|
|
response = await authenticated_admin_client.patch(
|
|
"/api/v1/admin/users/999",
|
|
json={"name": "Updated Name"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "User not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user_invalid_plan(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
regular_user: User,
|
|
) -> None:
|
|
"""Test updating user with invalid plan."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
) as mock_get_by_id,
|
|
patch("app.repositories.plan.PlanRepository.get_by_id", return_value=None),
|
|
):
|
|
mock_user = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": regular_user.name,
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": regular_user.credits,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": 1,
|
|
"name": "Basic",
|
|
"max_credits": 100,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
mock_get_by_id.return_value = mock_user
|
|
response = await authenticated_admin_client.patch(
|
|
"/api/v1/admin/users/2",
|
|
json={"plan_id": 999},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "Plan not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_user_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
regular_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test disabling user successfully."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
) as mock_get_by_id,
|
|
patch("app.repositories.user.UserRepository.update") as mock_update,
|
|
):
|
|
mock_user = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": regular_user.id,
|
|
"email": regular_user.email,
|
|
"name": regular_user.name,
|
|
"picture": None,
|
|
"role": regular_user.role,
|
|
"credits": regular_user.credits,
|
|
"is_active": regular_user.is_active,
|
|
"created_at": regular_user.created_at,
|
|
"updated_at": regular_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
mock_get_by_id.return_value = mock_user
|
|
mock_update.return_value = mock_user
|
|
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/admin/users/2/disable",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "User disabled successfully"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_user_not_found(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
) -> None:
|
|
"""Test disabling non-existent user."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
return_value=None,
|
|
),
|
|
):
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/admin/users/999/disable",
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "User not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_user_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test enabling user successfully."""
|
|
disabled_user = User(
|
|
id=3,
|
|
email="disabled@example.com",
|
|
name="Disabled User",
|
|
role="user",
|
|
credits=100,
|
|
plan_id=1,
|
|
is_active=False,
|
|
)
|
|
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
) as mock_get_by_id,
|
|
patch("app.repositories.user.UserRepository.update") as mock_update,
|
|
):
|
|
mock_disabled_user = type(
|
|
"User",
|
|
(),
|
|
{
|
|
"id": disabled_user.id,
|
|
"email": disabled_user.email,
|
|
"name": disabled_user.name,
|
|
"picture": None,
|
|
"role": disabled_user.role,
|
|
"credits": disabled_user.credits,
|
|
"is_active": disabled_user.is_active,
|
|
"created_at": disabled_user.created_at,
|
|
"updated_at": disabled_user.updated_at,
|
|
"plan": type(
|
|
"Plan",
|
|
(),
|
|
{
|
|
"id": test_plan.id,
|
|
"name": test_plan.name,
|
|
"max_credits": test_plan.max_credits,
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
mock_get_by_id.return_value = mock_disabled_user
|
|
mock_update.return_value = mock_disabled_user
|
|
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/admin/users/3/enable",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "User enabled successfully"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_user_not_found(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
) -> None:
|
|
"""Test enabling non-existent user."""
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.user.UserRepository.get_by_id_with_plan",
|
|
return_value=None,
|
|
),
|
|
):
|
|
response = await authenticated_admin_client.post(
|
|
"/api/v1/admin/users/999/enable",
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "User not found" in data["detail"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_plans_success(
|
|
self,
|
|
authenticated_admin_client: AsyncClient,
|
|
admin_user: User,
|
|
test_plan: Plan,
|
|
) -> None:
|
|
"""Test listing plans successfully."""
|
|
basic_plan = Plan(id=1, name="Basic", max_credits=100, features=["basic"])
|
|
premium_plan = Plan(id=2, name="Premium", max_credits=500, features=["premium"])
|
|
|
|
with (
|
|
patch("app.core.dependencies.get_admin_user", return_value=admin_user),
|
|
patch(
|
|
"app.repositories.plan.PlanRepository.get_all",
|
|
return_value=[basic_plan, premium_plan],
|
|
),
|
|
):
|
|
response = await authenticated_admin_client.get(
|
|
"/api/v1/admin/users/plans/list",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert data[0]["name"] == "Basic"
|
|
assert data[1]["name"] == "Premium"
|