qonto-assistant/tests/test_tenant_engine_client.py
tegwick b9349782f4 feat(audit): publish sequenced heartbeat and reconciliation evidence
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ec5-7e2b-7743-ac08-719e1b0f42e2
2026-09-05 01:39:48 +02:00

72 lines
2.7 KiB
Python

import httpx
from qonto_assistant.tenant_engine_client import TenantEngineClient
def _client(handler) -> TenantEngineClient:
return TenantEngineClient(
base_url="https://tenant-engine.example.test",
timeout_seconds=1,
transport=httpx.MockTransport(handler),
)
def test_active_roles_returns_roles_on_200() -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/tenants/tenant:friendly:binky/roles/live"
return httpx.Response(
200, json={"tenant_id": "tenant:friendly:binky", "roles": ["VEN", "CUS"]}
)
roles = _client(handler).active_roles("tenant:friendly:binky")
assert roles == frozenset({"VEN", "CUS"})
def test_active_roles_empty_when_no_roles() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": []})
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
def test_active_roles_fails_closed_on_404_tenant_not_found() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, json={"detail": "tenant_not_found"})
assert _client(handler).active_roles("tenant:friendly:nobody") == frozenset()
def test_active_roles_fails_closed_on_503_store_unavailable() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(503, json={"detail": "tenant_roles_unavailable"})
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
def test_active_roles_fails_closed_on_malformed_body() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, content=b"not json")
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
def test_active_roles_fails_closed_on_non_list_roles_field() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": "VEN"})
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
def test_active_roles_fails_closed_on_connection_error() -> None:
def handler(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("connection refused", request=request)
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
def test_active_roles_fails_closed_on_timeout() -> None:
def handler(request: httpx.Request) -> httpx.Response:
raise httpx.TimeoutException("timed out", request=request)
assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()