File size: 3,852 Bytes
f15fb1d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from __future__ import annotations
import asyncio
import time
from openai_codex.async_client import AsyncCodexClient
from openai_codex.generated.v2_all import (
TurnCompletedNotification,
)
from openai_codex.models import Notification, UnknownNotification
def test_async_client_allows_concurrent_transport_calls() -> None:
"""Async wrappers should offload sync calls so concurrent awaits can overlap."""
async def scenario() -> int:
"""Run two blocking sync calls and report peak overlap."""
client = AsyncCodexClient()
active = 0
max_active = 0
def fake_model_list(include_hidden: bool = False) -> bool:
"""Simulate a blocking sync transport call."""
nonlocal active, max_active
active += 1
max_active = max(max_active, active)
time.sleep(0.05)
active -= 1
return include_hidden
client._sync.model_list = fake_model_list # type: ignore[method-assign]
await asyncio.gather(client.model_list(), client.model_list())
return max_active
assert asyncio.run(scenario()) == 2
def test_async_client_turn_notification_methods_delegate_to_sync_client() -> None:
"""Async turn routing methods should preserve sync-client registration semantics."""
async def scenario() -> tuple[list[tuple[str, str]], Notification, str]:
"""Record the sync-client calls made by async turn notification wrappers."""
client = AsyncCodexClient()
event = Notification(
method="unknown/direct",
payload=UnknownNotification(params={"turnId": "turn-1"}),
)
completed = TurnCompletedNotification.model_validate(
{
"threadId": "thread-1",
"turn": {"id": "turn-1", "items": [], "status": "completed"},
}
)
calls: list[tuple[str, str]] = []
def fake_register(turn_id: str) -> None:
"""Record turn registration through the wrapped sync client."""
calls.append(("register", turn_id))
def fake_unregister(turn_id: str) -> None:
"""Record turn unregistration through the wrapped sync client."""
calls.append(("unregister", turn_id))
def fake_next(turn_id: str) -> Notification:
"""Return one routed notification through the wrapped sync client."""
calls.append(("next", turn_id))
return event
def fake_wait(turn_id: str) -> TurnCompletedNotification:
"""Return one completion through the wrapped sync client."""
calls.append(("wait", turn_id))
return completed
client._sync.register_turn_notifications = fake_register # type: ignore[method-assign]
client._sync.unregister_turn_notifications = fake_unregister # type: ignore[method-assign]
client._sync.next_turn_notification = fake_next # type: ignore[method-assign]
client._sync.wait_for_turn_completed = fake_wait # type: ignore[method-assign]
client.register_turn_notifications("turn-1")
next_event = await client.next_turn_notification("turn-1")
completed_event = await client.wait_for_turn_completed("turn-1")
client.unregister_turn_notifications("turn-1")
return calls, next_event, completed_event.turn.id
calls, next_event, completed_turn_id = asyncio.run(scenario())
assert (
calls,
next_event,
completed_turn_id,
) == (
[
("register", "turn-1"),
("next", "turn-1"),
("wait", "turn-1"),
("unregister", "turn-1"),
],
Notification(
method="unknown/direct",
payload=UnknownNotification(params={"turnId": "turn-1"}),
),
"turn-1",
)
|