146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
"""Tests for the gateway client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from conduit_client import AsyncConduitClient, ConduitClient
|
|
from conduit_client.models.common import CreateKeyRequest
|
|
from conduit_client.models.gateway import AuditLogQuery
|
|
|
|
|
|
@respx.mock
|
|
def test_health(api_key: str) -> None:
|
|
respx.get("https://api.example.com/health").mock(return_value=Response(200, text="ok"))
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.health()
|
|
assert result == "ok"
|
|
|
|
|
|
@respx.mock
|
|
def test_me(api_key: str) -> None:
|
|
respx.get("https://api.example.com/auth/me").mock(
|
|
return_value=Response(
|
|
200,
|
|
json={
|
|
"subject": "user:123",
|
|
"actor_type": "user",
|
|
"auth_method": "api_key",
|
|
"scopes": ["gateway:read"],
|
|
"grantable_scopes": ["gateway:read", "sms:send"],
|
|
},
|
|
)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.me()
|
|
assert result.subject == "user:123"
|
|
assert result.scopes == ["gateway:read"]
|
|
|
|
|
|
@respx.mock
|
|
def test_services(api_key: str) -> None:
|
|
respx.get("https://api.example.com/services").mock(
|
|
return_value=Response(
|
|
200,
|
|
json=[
|
|
{"id": "sms", "name": "SMS", "prefixes": ["/sms"], "enabled": True},
|
|
],
|
|
)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.services()
|
|
assert len(result) == 1
|
|
assert result[0].id == "sms"
|
|
|
|
|
|
@respx.mock
|
|
def test_list_keys(api_key: str) -> None:
|
|
created = datetime.now(timezone.utc).isoformat()
|
|
respx.get("https://api.example.com/keys").mock(
|
|
return_value=Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"key_prefix": "abc123",
|
|
"display_name": "test",
|
|
"scopes": ["gateway:read"],
|
|
"created_at": created,
|
|
},
|
|
],
|
|
)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.list_keys()
|
|
assert result[0].key_prefix == "abc123"
|
|
|
|
|
|
@respx.mock
|
|
def test_create_key(api_key: str) -> None:
|
|
created = datetime.now(timezone.utc).isoformat()
|
|
route = respx.post("https://api.example.com/keys").mock(
|
|
return_value=Response(
|
|
200,
|
|
json={
|
|
"key": "mega_sk_live_abc_secret",
|
|
"key_prefix": "abc",
|
|
"display_name": "new key",
|
|
"scopes": ["sms:read"],
|
|
"created_at": created,
|
|
},
|
|
)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.create_key(CreateKeyRequest(display_name="new key", scopes=["sms:read"]))
|
|
assert result.key == "mega_sk_live_abc_secret"
|
|
assert json.loads(route.calls[0].request.content) == {
|
|
"display_name": "new key",
|
|
"scopes": ["sms:read"],
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_revoke_key(api_key: str) -> None:
|
|
respx.delete("https://api.example.com/keys/abc").mock(return_value=Response(204))
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.revoke_key("abc")
|
|
assert result is None
|
|
|
|
|
|
@respx.mock
|
|
def test_admin_audit(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/admin/audit").mock(
|
|
return_value=Response(200, json=[])
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.gateway.admin_audit(
|
|
AuditLogQuery(actor_subject="user:123", limit=50)
|
|
)
|
|
assert result == []
|
|
assert route.calls[0].request.url.params["actor_subject"] == "user:123"
|
|
assert route.calls[0].request.url.params["limit"] == "50"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_async_me(api_key: str) -> None:
|
|
respx.get("https://api.example.com/auth/me").mock(
|
|
return_value=Response(
|
|
200,
|
|
json={
|
|
"subject": "user:123",
|
|
"actor_type": "user",
|
|
"auth_method": "api_key",
|
|
"scopes": ["gateway:read"],
|
|
"grantable_scopes": ["gateway:read", "sms:send"],
|
|
},
|
|
)
|
|
)
|
|
async with AsyncConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = await client.gateway.me()
|
|
assert result.subject == "user:123"
|