120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
"""Tests for the shared HTTP transport layer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from conduit_client import (
|
|
AsyncConduitClient,
|
|
AuthenticationError,
|
|
ConduitClient,
|
|
NotFoundError,
|
|
RateLimitError,
|
|
ValidationError,
|
|
)
|
|
from conduit_client._base_client import _BaseClientMixin
|
|
|
|
|
|
@respx.mock
|
|
def test_sync_auth_header(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/auth/me").mock(return_value=Response(200, json={}))
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client._transport.request("GET", "/auth/me")
|
|
assert route.calls[0].request.headers["Authorization"] == f"Bearer {api_key}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_async_auth_header(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/auth/me").mock(return_value=Response(200, json={}))
|
|
async with AsyncConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
await client._transport.request("GET", "/auth/me")
|
|
assert route.calls[0].request.headers["Authorization"] == f"Bearer {api_key}"
|
|
|
|
|
|
@respx.mock
|
|
def test_sync_json_response(api_key: str) -> None:
|
|
respx.get("https://api.example.com/auth/me").mock(return_value=Response(200, json={"id": 1}))
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client._transport.request("GET", "/auth/me")
|
|
assert result == {"id": 1}
|
|
|
|
|
|
@respx.mock
|
|
def test_sync_text_response(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._transport.request("GET", "/health")
|
|
assert result == "ok"
|
|
|
|
|
|
@respx.mock
|
|
def test_sync_204_returns_none(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._transport.request("DELETE", "/keys/abc")
|
|
assert result is None
|
|
|
|
|
|
@respx.mock
|
|
def test_authentication_error(api_key: str) -> None:
|
|
respx.get("https://api.example.com/auth/me").mock(
|
|
return_value=Response(401, json={"error": "unauthorized"})
|
|
)
|
|
with pytest.raises(AuthenticationError) as exc, ConduitClient(
|
|
api_key, base_url="https://api.example.com"
|
|
) as client:
|
|
client._transport.request("GET", "/auth/me")
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
@respx.mock
|
|
def test_not_found_error(api_key: str) -> None:
|
|
respx.get("https://api.example.com/wiki/pages/missing").mock(
|
|
return_value=Response(404, text="not found")
|
|
)
|
|
with pytest.raises(NotFoundError) as exc, ConduitClient(
|
|
api_key, base_url="https://api.example.com"
|
|
) as client:
|
|
client._transport.request("GET", "/wiki/pages/missing")
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
@respx.mock
|
|
def test_validation_error(api_key: str) -> None:
|
|
respx.post("https://api.example.com/keys").mock(
|
|
return_value=Response(422, json={"error": "unknown scopes"})
|
|
)
|
|
with pytest.raises(ValidationError) as exc, ConduitClient(
|
|
api_key, base_url="https://api.example.com"
|
|
) as client:
|
|
client._transport.request("POST", "/keys", json={"scopes": ["bad"]})
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
@respx.mock
|
|
def test_rate_limit_retry_then_success(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/health").mock(
|
|
side_effect=[Response(429, text="slow down"), Response(200, text="ok")]
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com", max_retries=2) as client:
|
|
result = client._transport.request("GET", "/health")
|
|
assert result == "ok"
|
|
assert route.call_count == 2
|
|
|
|
|
|
@respx.mock
|
|
def test_rate_limit_exhausted(api_key: str) -> None:
|
|
respx.get("https://api.example.com/health").mock(return_value=Response(429, text="slow down"))
|
|
with pytest.raises(RateLimitError), ConduitClient(
|
|
api_key, base_url="https://api.example.com", max_retries=1
|
|
) as client:
|
|
client._transport.request("GET", "/health")
|
|
|
|
|
|
def test_encode_path_param() -> None:
|
|
assert _BaseClientMixin._encode_path_param("foo/bar") == "foo%2Fbar"
|
|
assert _BaseClientMixin._encode_path_param("example.com") == "example.com"
|