feat: add /api/admin/users/search endpoint

- Add endpoint to search users by email (case-insensitive)
- Limit results to 10 for autocomplete purposes
- Require VIEW_ALL_EXCHANGES permission (admin only)
- Add tests for search functionality and access control
This commit is contained in:
counterweight 2025-12-23 10:55:44 +01:00
parent 27896ed136
commit ef01a970d5
Signed by: counterweight
GPG key ID: 883EDBAA726BD96C
2 changed files with 89 additions and 1 deletions

View file

@ -813,5 +813,39 @@ async def admin_cancel_trade(
return _to_admin_exchange_response(exchange)
# =============================================================================
# Admin User Search Endpoint
# =============================================================================
admin_users_router = APIRouter(prefix="/api/admin/users", tags=["admin-users"])
class UserSearchResult(BaseModel):
"""Result item for user search."""
id: int
email: str
@admin_users_router.get("/search", response_model=list[UserSearchResult])
async def search_users(
q: str = Query(..., min_length=1, description="Search query for user email"),
db: AsyncSession = Depends(get_db),
_current_user: User = Depends(require_permission(Permission.VIEW_ALL_EXCHANGES)),
) -> list[UserSearchResult]:
"""
Search users by email for autocomplete.
Returns users whose email contains the search query (case-insensitive).
Limited to 10 results for autocomplete purposes.
"""
result = await db.execute(
select(User).where(User.email.ilike(f"%{q}%")).order_by(User.email).limit(10)
)
users = result.scalars().all()
return [UserSearchResult(id=u.id, email=u.email) for u in users]
# All routers from this module for easy registration
routers = [router, trades_router, admin_trades_router]
routers = [router, trades_router, admin_trades_router, admin_users_router]

View file

@ -862,3 +862,57 @@ class TestAdminCancelTrade:
response = await client.post(f"/api/admin/trades/{trade_id}/cancel")
assert response.status_code == 403
# =============================================================================
# User Search Tests
# =============================================================================
class TestAdminUserSearch:
"""Test admin user search endpoint."""
@pytest.mark.asyncio
async def test_admin_can_search_users(
self, client_factory, admin_user, regular_user
):
"""Admin can search for users by email."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
# Search for the regular user
response = await client.get(
f"/api/admin/users/search?q={regular_user['user']['email'][:5]}"
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
# Should find the regular user
emails = [u["email"] for u in data]
assert regular_user["user"]["email"] in emails
@pytest.mark.asyncio
async def test_search_returns_limited_results(self, client_factory, admin_user):
"""Search results are limited to 10."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
# Search with a common pattern
response = await client.get("/api/admin/users/search?q=@")
assert response.status_code == 200
data = response.json()
assert len(data) <= 10
@pytest.mark.asyncio
async def test_regular_user_cannot_search(self, client_factory, regular_user):
"""Regular user cannot access user search."""
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/admin/users/search?q=test")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_search_requires_query(self, client_factory, admin_user):
"""Search requires a query parameter."""
async with client_factory.create(cookies=admin_user["cookies"]) as client:
response = await client.get("/api/admin/users/search")
assert response.status_code == 422 # Validation error