You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from iti.service_client import (
|
|
ServiceClient,
|
|
ServiceBusinessError,
|
|
ServiceConfig,
|
|
ServiceHTTPError,
|
|
ServiceUnavailableError,
|
|
)
|
|
|
|
|
|
def test_service_client_sends_json_and_token():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["Authorization"] == "Bearer token"
|
|
return httpx.Response(200, json={"ok": True})
|
|
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping(
|
|
"demo",
|
|
{"base_url": "https://example.test", "token": "token"},
|
|
),
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
assert client.get("/ping") == {"ok": True}
|
|
|
|
|
|
def test_service_client_unwraps_envelope():
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping("demo", {"base_url": "https://example.test"}),
|
|
transport=httpx.MockTransport(
|
|
lambda request: httpx.Response(
|
|
200,
|
|
json={"data": {"ok": True}, "code": 200, "message": "成功"},
|
|
)
|
|
),
|
|
)
|
|
|
|
assert client.get("/ping") == {"ok": True}
|
|
|
|
|
|
def test_service_client_raises_for_business_errors():
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping("demo", {"base_url": "https://example.test"}),
|
|
transport=httpx.MockTransport(
|
|
lambda request: httpx.Response(
|
|
200,
|
|
json={"data": None, "code": 403, "message": "无权限"},
|
|
)
|
|
),
|
|
)
|
|
|
|
with pytest.raises(ServiceBusinessError) as exc:
|
|
client.get("/ping")
|
|
|
|
assert exc.value.code == 403
|
|
assert exc.value.status_code == 200
|
|
|
|
|
|
def test_service_client_prefers_envelope_code_on_http_error():
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping("demo", {"base_url": "https://example.test"}),
|
|
transport=httpx.MockTransport(
|
|
lambda request: httpx.Response(
|
|
401,
|
|
json={"data": {"reason": "expired"}, "code": 401, "message": "未认证"},
|
|
)
|
|
),
|
|
)
|
|
|
|
with pytest.raises(ServiceBusinessError) as exc:
|
|
client.get("/ping")
|
|
|
|
assert exc.value.code == 401
|
|
assert exc.value.status_code == 401
|
|
assert exc.value.data == {"reason": "expired"}
|
|
|
|
|
|
def test_service_client_raises_for_http_errors():
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping("demo", {"base_url": "https://example.test"}),
|
|
transport=httpx.MockTransport(lambda request: httpx.Response(503, text="down")),
|
|
)
|
|
|
|
with pytest.raises(ServiceHTTPError) as exc:
|
|
client.get("/ping")
|
|
|
|
assert exc.value.status_code == 503
|
|
|
|
|
|
def test_service_client_circuit_breaker_opens_after_failures():
|
|
client = ServiceClient(
|
|
ServiceConfig.from_mapping(
|
|
"demo",
|
|
{
|
|
"base_url": "https://example.test",
|
|
"circuit_breaker": {"enabled": True, "fail_max": 1},
|
|
},
|
|
),
|
|
transport=httpx.MockTransport(lambda request: httpx.Response(503, text="down")),
|
|
)
|
|
|
|
with pytest.raises(ServiceHTTPError):
|
|
client.get("/ping")
|
|
with pytest.raises(ServiceUnavailableError, match="circuit breaker"):
|
|
client.get("/ping")
|