fix(binding): accept 4G nanosecond timestamps on Python 3.10

This commit is contained in:
stu2not
2026-09-23 17:57:48 +08:00
parent 515567f6d4
commit 9cb71958f6
2 changed files with 21 additions and 2 deletions
@@ -80,8 +80,11 @@ async def binding_relation(source_binding_id: str, request_id: str, *, close=Fal
valid = valid and data.get("status") == "unknown" and not data.get("acceptingHandoff")
try:
checked = data.get("checkedAt")
valid = valid and isinstance(checked, str) and checked.endswith("Z")
datetime.fromisoformat(checked.replace("Z", "+00:00"))
valid = valid and isinstance(checked, str) and bool(re.fullmatch(
r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]+)?Z", checked))
# Java Instant can return nanoseconds; Python 3.10 fromisoformat rejects
# that precision. Validate the calendar separately and preserve the value.
datetime.strptime(checked[:19], "%Y-%m-%dT%H:%M:%S")
except (AttributeError, TypeError, ValueError):
valid = False
if not valid:
@@ -43,6 +43,10 @@ async def test_query_and_close_encode_existing_ids_and_use_service_auth(monkeypa
{"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}}))
@@ -71,6 +75,18 @@ async def test_valid_non_success_business_states_are_preserved(monkeypatch, stat
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")