c93562e685
Sync (requests) and async (httpx) clients with identical API surface to the Node.js SDK. Delivered: - pyproject.toml — python>=3.9, hatchling build, mypy strict config - types.py — all 14-endpoint request/response dataclasses - errors.py — AgentIdPError with from_api_error, from_oauth2_error, network_error - token_manager.py — thread-safe sync TokenManager, 60s refresh buffer - async_token_manager.py — asyncio-safe AsyncTokenManager (httpx) - _request.py — shared sync/async request helper (DRY) - services/agents.py — AgentRegistryClient + AsyncAgentRegistryClient (5 methods each) - services/credentials.py — CredentialClient + AsyncCredentialClient (4 methods each) - services/token.py — TokenClient + AsyncTokenClient (introspect + revoke) - services/audit.py — AuditClient + AsyncAuditClient (query + get) - client.py — AgentIdPClient + AsyncAgentIdPClient - __init__.py — barrel exports - README.md — installation, quick start, full API reference QA gates: - mypy --strict: 0 errors (12 source files) - pytest: 57/57 passed - Coverage: 90.83% (required >= 80%) - All 14 endpoints covered (sync + async) - AgentIdPError raised on all failure paths Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""Tests for TokenManager (sync) and AsyncTokenManager."""
|
|
|
|
import time
|
|
import pytest
|
|
import responses as resp_lib
|
|
import respx
|
|
import httpx
|
|
|
|
from sentryagent_idp.token_manager import TokenManager, REFRESH_BUFFER_SECONDS
|
|
from sentryagent_idp.async_token_manager import AsyncTokenManager
|
|
from sentryagent_idp.errors import AgentIdPError
|
|
|
|
BASE_URL = "http://localhost:3000"
|
|
TOKEN_URL = f"{BASE_URL}/api/v1/token"
|
|
TOKEN_RESP = {
|
|
"access_token": "eyJ.abc.def",
|
|
"token_type": "Bearer",
|
|
"expires_in": 3600,
|
|
"scope": "agents:read",
|
|
}
|
|
|
|
|
|
# ─── Sync TokenManager ────────────────────────────────────────────────────────
|
|
|
|
@resp_lib.activate
|
|
def test_token_manager_issues_token() -> None:
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json=TOKEN_RESP, status=200)
|
|
tm = TokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
token = tm.get_token()
|
|
assert token == "eyJ.abc.def"
|
|
assert len(resp_lib.calls) == 1
|
|
|
|
|
|
@resp_lib.activate
|
|
def test_token_manager_caches_token() -> None:
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json=TOKEN_RESP, status=200)
|
|
tm = TokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
tm.get_token()
|
|
tm.get_token()
|
|
# Only one HTTP call because second call uses cache
|
|
assert len(resp_lib.calls) == 1
|
|
|
|
|
|
@resp_lib.activate
|
|
def test_token_manager_refreshes_near_expiry() -> None:
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json={**TOKEN_RESP, "expires_in": 30}, status=200)
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json=TOKEN_RESP, status=200)
|
|
tm = TokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
tm.get_token()
|
|
# Simulate cached token being nearly expired
|
|
assert tm._cached is not None
|
|
tm._cached.expires_at = time.time() + (REFRESH_BUFFER_SECONDS - 1)
|
|
tm.get_token()
|
|
assert len(resp_lib.calls) == 2
|
|
|
|
|
|
@resp_lib.activate
|
|
def test_token_manager_raises_on_auth_failure() -> None:
|
|
resp_lib.add(
|
|
resp_lib.POST, TOKEN_URL,
|
|
json={"error": "invalid_client", "error_description": "Bad creds."},
|
|
status=401,
|
|
)
|
|
tm = TokenManager(BASE_URL, "client-id", "bad-secret", "agents:read")
|
|
with pytest.raises(AgentIdPError) as exc_info:
|
|
tm.get_token()
|
|
assert exc_info.value.code == "invalid_client"
|
|
assert exc_info.value.http_status == 401
|
|
|
|
|
|
@resp_lib.activate
|
|
def test_token_manager_clear_cache() -> None:
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json=TOKEN_RESP, status=200)
|
|
resp_lib.add(resp_lib.POST, TOKEN_URL, json=TOKEN_RESP, status=200)
|
|
tm = TokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
tm.get_token()
|
|
tm.clear_cache()
|
|
tm.get_token()
|
|
assert len(resp_lib.calls) == 2
|
|
|
|
|
|
# ─── Async TokenManager ───────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_token_manager_issues_token() -> None:
|
|
with respx.mock:
|
|
respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=TOKEN_RESP))
|
|
tm = AsyncTokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
token = await tm.get_token()
|
|
assert token == "eyJ.abc.def"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_token_manager_caches_token() -> None:
|
|
with respx.mock:
|
|
route = respx.post(TOKEN_URL).mock(return_value=httpx.Response(200, json=TOKEN_RESP))
|
|
tm = AsyncTokenManager(BASE_URL, "client-id", "secret", "agents:read")
|
|
await tm.get_token()
|
|
await tm.get_token()
|
|
assert route.call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_token_manager_raises_on_auth_failure() -> None:
|
|
with respx.mock:
|
|
respx.post(TOKEN_URL).mock(return_value=httpx.Response(
|
|
401, json={"error": "invalid_client", "error_description": "Bad creds."}
|
|
))
|
|
tm = AsyncTokenManager(BASE_URL, "client-id", "bad-secret", "agents:read")
|
|
with pytest.raises(AgentIdPError) as exc_info:
|
|
await tm.get_token()
|
|
assert exc_info.value.code == "invalid_client"
|