| """Transport contracts for the AffixIO production API client.""" |
|
|
| |
|
|
| from __future__ import annotations |
|
|
| import httpx |
| import pytest |
|
|
| from affix_huggingface.errors import AffixAPIError |
| from affix_huggingface.transport import AffixTransport |
|
|
|
|
| def test_prove_sends_auth_idempotency_and_encoded_circuit() -> None: |
| captured: list[httpx.Request] = [] |
|
|
| def handler(request: httpx.Request) -> httpx.Response: |
| captured.append(request) |
| return httpx.Response( |
| 200, |
| json={"proof": "proof_hex"}, |
| headers={"X-Request-Id": "req_1"}, |
| ) |
|
|
| http = httpx.Client(transport=httpx.MockTransport(handler)) |
| transport = AffixTransport("aio_test", client=http) |
| data, request_id = transport.prove( |
| circuit_id="yes/no", |
| fields={"condition_greater1": True}, |
| idempotency_key="idem_1", |
| ) |
|
|
| assert data["proof"] == "proof_hex" |
| assert request_id == "req_1" |
| assert captured[0].url.raw_path == b"/v1/circuits/yes%2Fno/prove" |
| assert captured[0].headers["Authorization"] == "Bearer aio_test" |
| assert captured[0].headers["X-API-Key"] == "aio_test" |
| assert captured[0].headers["Idempotency-Key"] == "idem_1" |
|
|
|
|
| def test_gate_returns_structured_denial_from_403() -> None: |
| def handler(_: httpx.Request) -> httpx.Response: |
| return httpx.Response( |
| 403, |
| json={ |
| "allow": False, |
| "reason_code": "REGION_MISMATCH", |
| "receipt_id": "rcpt_denied", |
| }, |
| ) |
|
|
| http = httpx.Client(transport=httpx.MockTransport(handler)) |
| transport = AffixTransport("aio_test", client=http) |
| data, _ = transport.gate( |
| proof="proof_hex", |
| circuit_id="yesno", |
| policy_id="gate.agent", |
| mode="consume", |
| gate_id="a" * 64, |
| idempotency_key="idem_gate", |
| ) |
| assert data["allow"] is False |
| assert data["reason_code"] == "REGION_MISMATCH" |
|
|
|
|
| def test_retryable_response_is_retried(monkeypatch: pytest.MonkeyPatch) -> None: |
| attempts = 0 |
|
|
| def handler(_: httpx.Request) -> httpx.Response: |
| nonlocal attempts |
| attempts += 1 |
| if attempts == 1: |
| return httpx.Response(503, json={"error": "unavailable"}) |
| return httpx.Response(200, json={"ok": True}) |
|
|
| monkeypatch.setattr(AffixTransport, "_sleep", staticmethod(lambda *_: None)) |
| http = httpx.Client(transport=httpx.MockTransport(handler)) |
| transport = AffixTransport("aio_test", client=http, max_retries=1) |
| data, _ = transport.health() |
| assert data == {"ok": True} |
| assert attempts == 2 |
|
|
|
|
| def test_non_retryable_error_preserves_api_fields() -> None: |
| def handler(_: httpx.Request) -> httpx.Response: |
| return httpx.Response( |
| 401, |
| json={"error": "unauthorised", "message": "API key rejected"}, |
| headers={"X-Request-Id": "req_auth"}, |
| ) |
|
|
| http = httpx.Client(transport=httpx.MockTransport(handler)) |
| transport = AffixTransport("bad_key", client=http) |
| with pytest.raises(AffixAPIError) as exc: |
| transport.health() |
| assert exc.value.status_code == 401 |
| assert exc.value.code == "unauthorised" |
| assert exc.value.request_id == "req_auth" |
|
|
|
|
