bruno
2f2a5e9b4a
Introduce an admin portal (React + Nginx), WebSocket routing, and API versioning middleware with `/api/v1/` prefix deprecation. Add master API key authentication, new Prometheus metrics for AI token consumption and active WebSockets, and extend S3 config with a public endpoint URL. Update test paths and fixtures to align with the new routing structure.
53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_stats(async_client: AsyncClient, admin_headers: dict, admin_client):
|
|
response = await async_client.get("/admin/api/stats", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_images" in data
|
|
assert "total_storage_bytes" in data
|
|
assert "total_ai_tokens" in data
|
|
assert "total_clients" in data
|
|
|
|
assert data["total_clients"] >= 1 # at least the admin client
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_clients(async_client: AsyncClient, admin_headers: dict, admin_client):
|
|
response = await async_client.get("/admin/api/clients", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
assert data[0]["name"] == "Admin Client"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_queue_status(async_client: AsyncClient, admin_headers: dict, admin_client):
|
|
response = await async_client.get("/admin/api/queue/status", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "pending_jobs" in data
|
|
assert data["status"] == "active"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_toggle_and_reset(async_client: AsyncClient, admin_headers: dict, admin_client, client_a):
|
|
# Get all clients
|
|
resp = await async_client.get("/admin/api/clients", headers=admin_headers)
|
|
clients = resp.json()
|
|
|
|
# Find client_a to toggle (don't toggle the admin client we're using!)
|
|
target_client = next(c for c in clients if c["id"] != admin_client.id)
|
|
client_id = target_client["id"]
|
|
is_active_initial = target_client["is_active"]
|
|
|
|
# Toggle
|
|
toggle_resp = await async_client.post(f"/admin/api/clients/{client_id}/toggle", headers=admin_headers)
|
|
assert toggle_resp.status_code == 200
|
|
assert toggle_resp.json()["is_active"] == (not is_active_initial)
|
|
|
|
# Reset Quota
|
|
reset_resp = await async_client.post(f"/admin/api/clients/{client_id}/reset-quota", headers=admin_headers)
|
|
assert reset_resp.status_code == 200
|
|
assert reset_resp.json()["storage_used"] == 0
|