55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.services import keycloak_user_sync
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_fetch_keycloak_users_uses_admin_token_and_paginates():
|
|
user_ids = [uuid4() for _ in range(101)]
|
|
requested_first: list[int] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["authorization"] == "Bearer admin-token"
|
|
first = int(request.url.params["first"])
|
|
requested_first.append(first)
|
|
page_ids = user_ids[first : first + 100]
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"id": str(user_id),
|
|
"username": f"user-{first + index}",
|
|
"email": f"user-{first + index}@example.com",
|
|
}
|
|
for index, user_id in enumerate(page_ids)
|
|
],
|
|
)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
users = await keycloak_user_sync.fetch_keycloak_users(
|
|
issuer="https://keycloak.example.com/realms/tjwater",
|
|
bearer_token="admin-token",
|
|
client=client,
|
|
)
|
|
|
|
assert len(users) == 101
|
|
assert requested_first == [0, 100]
|
|
assert users[-1].username == "user-100"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_fetch_keycloak_users_requires_realm_issuer():
|
|
with pytest.raises(keycloak_user_sync.KeycloakUserSyncConfigurationError):
|
|
await keycloak_user_sync.fetch_keycloak_users(
|
|
issuer="",
|
|
bearer_token="admin-token",
|
|
)
|