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.
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from imago_client import HubClient, ImagoError, APIError, QuotaError, AuthError
|
|
|
|
@pytest.fixture
|
|
def base_url():
|
|
return "http://testserver"
|
|
|
|
@pytest.fixture
|
|
def api_key():
|
|
return "sk-test-key"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_initialization(base_url, api_key):
|
|
client = HubClient(base_url, api_key=api_key)
|
|
assert client.base_url == "http://testserver"
|
|
assert client.api_key == "sk-test-key"
|
|
assert "X-API-Key" in client._http_client.headers
|
|
await client.close()
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_auth_error(base_url, api_key):
|
|
respx.get(f"{base_url}/api/v1/images").mock(return_value=Response(401, json={"detail": "Unauthorized"}))
|
|
|
|
async with HubClient(base_url, api_key="wrong") as client:
|
|
with pytest.raises(AuthError):
|
|
await client.images.list()
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_quota_error(base_url, api_key):
|
|
respx.get(f"{base_url}/api/v1/images").mock(return_value=Response(429, json={"detail": "Rate Limit Exceeded"}))
|
|
|
|
async with HubClient(base_url, api_key=api_key) as client:
|
|
with pytest.raises(QuotaError):
|
|
await client.images.list()
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_api_error_retry(base_url, api_key):
|
|
route = respx.get(f"{base_url}/api/v1/images")
|
|
route.side_effect = [
|
|
Response(502, json={"error": "Bad Gateway"}),
|
|
Response(200, json={"items": []})
|
|
]
|
|
|
|
async with HubClient(base_url, api_key=api_key) as client:
|
|
# Should automatically retry on 502
|
|
response = await client.images.list()
|
|
assert response == {"items": []}
|
|
assert route.call_count == 2
|