96 lines
4.7 KiB
Python
96 lines
4.7 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from banban.service import traffic_handoff as adapter
|
|
|
|
SOURCE = "banban:12345678-1234-1234-1234-123456789abc"
|
|
RELEASED = {"sourceBindingId": SOURCE, "status": "unbound", "relationState": "closing",
|
|
"pendingHandoff": False, "acceptingHandoff": False, "canUnbind": True,
|
|
"reason": "ORIGINAL_BINDING_RELEASED", "checkedAt": "2026-09-23T08:00:00Z"}
|
|
|
|
|
|
def transport(monkeypatch, callback):
|
|
client = httpx.AsyncClient
|
|
monkeypatch.setattr(adapter.settings, "banban_handoff_service_token", "test-only")
|
|
monkeypatch.setattr(adapter.httpx, "AsyncClient", lambda **kwargs: client(transport=httpx.MockTransport(callback), **kwargs))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_and_close_encode_existing_ids_and_use_service_auth(monkeypatch):
|
|
requests = []
|
|
def reply(request):
|
|
requests.append(request)
|
|
return httpx.Response(200, json={"code": "OK", "data": RELEASED})
|
|
transport(monkeypatch, reply)
|
|
await adapter.binding_relation(SOURCE, "query-1")
|
|
for _ in range(2): await adapter.binding_relation(SOURCE, "close-1", close=True)
|
|
assert requests[0].method == "GET" and requests[0].content == b""
|
|
assert b"banban%3A" in requests[0].url.raw_path
|
|
assert all(request.headers["X-Service-Token"] == "test-only" for request in requests)
|
|
assert all("Authorization" not in request.headers for request in requests)
|
|
assert requests[0].headers["Cache-Control"] == "no-cache"
|
|
assert requests[1].url.path.endswith("/close") and requests[1].method == "POST"
|
|
assert json.loads(requests[1].content) == {"requestId": "close-1"}
|
|
assert requests[1].content == requests[2].content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("changes", [
|
|
{"sourceBindingId": "another"}, {"canUnbind": "true"}, {"pendingHandoff": 0},
|
|
{"pendingHandoff": True}, {"acceptingHandoff": True}, {"relationState": "open"},
|
|
{"status": "bound"}, {"status": "unknown"}, {"status": {}},
|
|
{"relationState": []}, {"checkedAt": None}, {"checkedAt": "badZ"},
|
|
{"checkedAt": "2026-09-23T08:00:00"}, {"canUnbind": False},
|
|
{"checkedAt": "2026-02-30T08:00:00.123456789Z"},
|
|
{"checkedAt": "2026-09-23T08:00:00.Z"},
|
|
{"checkedAt": "2026-09-23T08:00:00.123456789+00:00"},
|
|
{"checkedAt": "2026-09-23T08:00:00.123456789Zjunk"},
|
|
])
|
|
async def test_inconsistent_or_malformed_relation_never_authorizes_unbind(monkeypatch, changes):
|
|
transport(monkeypatch, lambda request: httpx.Response(200, json={"code": "OK", "data": {**RELEASED, **changes}}))
|
|
with pytest.raises(adapter.TrafficError) as failure: await adapter.binding_relation(SOURCE, "query")
|
|
assert failure.value.code == "INVALID_RESPONSE"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("status,code", [(401, "HANDOFF_SERVICE_UNAUTHORIZED"), (429, "RATE_LIMITED"),
|
|
(500, "INTERNAL_ERROR"), (404, "NOT_FOUND"), (200, "HANDOFF_RELATION_UNKNOWN")])
|
|
async def test_http_or_business_errors_never_become_unbound(monkeypatch, status, code):
|
|
transport(monkeypatch, lambda request: httpx.Response(status, json={"code": code, "data": RELEASED}))
|
|
with pytest.raises(adapter.TrafficError): await adapter.binding_relation(SOURCE, "close", close=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("state", [
|
|
{**RELEASED, "status": "never_bound"},
|
|
{**RELEASED, "status": "bound", "canUnbind": False},
|
|
{**RELEASED, "status": "unknown", "canUnbind": False},
|
|
{**RELEASED, "status": "unknown", "relationState": "unknown", "canUnbind": False},
|
|
{**RELEASED, "status": "never_bound", "relationState": "open", "acceptingHandoff": True, "pendingHandoff": True, "canUnbind": False},
|
|
])
|
|
async def test_valid_non_success_business_states_are_preserved(monkeypatch, state):
|
|
transport(monkeypatch, lambda request: httpx.Response(200, json={"code": "OK", "data": state}))
|
|
assert await adapter.binding_relation(SOURCE, "query") == state
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("checked", [
|
|
"2026-09-23T09:52:51Z", "2026-09-23T09:52:51.2Z",
|
|
"2026-09-23T09:52:51.262Z", "2026-09-23T09:52:51.262251Z",
|
|
"2026-09-23T09:52:51.262251915Z",
|
|
])
|
|
async def test_java_instant_precision_is_accepted_and_preserved(monkeypatch, checked):
|
|
state = {**RELEASED, "checkedAt": checked}
|
|
transport(monkeypatch, lambda request: httpx.Response(200, json={"code": "OK", "data": state}))
|
|
assert await adapter.binding_relation(SOURCE, "query") == state
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_relation_network_timeout_is_explicit(monkeypatch):
|
|
def timeout(request): raise httpx.ReadTimeout("test timeout")
|
|
transport(monkeypatch, timeout)
|
|
with pytest.raises(adapter.TrafficError) as failure: await adapter.binding_relation(SOURCE, "query")
|
|
assert failure.value.code == "TRAFFIC_UNAVAILABLE"
|