Skip to content

Commit d31c5ea

Browse files
author
Jianke LIN
committed
test: exercise dispatcher shutdown drain mode
1 parent 1498a1d commit d31c5ea

1 file changed

Lines changed: 13 additions & 4 deletions

File tree

tests/shared/test_jsonrpc_dispatcher.py

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1038,10 +1038,14 @@ async def test_shutdown_error_response_write_is_bounded_when_the_transport_is_we
10381038
):
10391039
"""Cancelling the task group hosting run() completes even when the shutdown error write wedges:
10401040
only `_SHUTDOWN_WRITE_TIMEOUT` releases the join (SDK-defined). A 0-buffer stream nobody reads
1041-
expresses the wedge: run() closes its write stream only after the join, so the send stays parked."""
1041+
expresses the wedge: drain-mode run() closes its write stream only after the join, so the send stays parked."""
10421042
c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
10431043
s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](0)
1044-
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
1044+
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
1045+
c2s_recv,
1046+
s2c_send,
1047+
close_write_stream_on_read_close=False,
1048+
)
10451049
handler_started = anyio.Event()
10461050

10471051
async def park(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
@@ -1072,10 +1076,15 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) ->
10721076
@pytest.mark.anyio
10731077
async def test_shutdown_answers_in_flight_request_with_connection_closed():
10741078
"""Read-stream EOF answers a still-running request with CONNECTION_CLOSED (SDK-defined):
1075-
run() keeps the write stream open until the task-group join, so the shielded teardown write lands."""
1079+
drain-mode run() keeps the write stream open until the task-group join, so the shielded teardown write lands."""
10761080
c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4)
10771081
s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4)
1078-
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
1082+
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
1083+
c2s_recv,
1084+
s2c_send,
1085+
close_write_stream_on_read_close=False,
1086+
read_eof_drain_timeout_seconds=0.05,
1087+
)
10791088
handler_started = anyio.Event()
10801089

10811090
async def park(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:

0 commit comments

Comments
 (0)