diff --git a/lib/bindings/python/tests/cancellation/test_cancellation.py b/lib/bindings/python/tests/cancellation/test_cancellation.py index 9171e015b6ac..7dd06a02b007 100644 --- a/lib/bindings/python/tests/cancellation/test_cancellation.py +++ b/lib/bindings/python/tests/cancellation/test_cancellation.py @@ -11,6 +11,7 @@ pytest.mark.gpu_0, pytest.mark.pre_merge, pytest.mark.unit, + pytest.mark.timeout(30), ] @@ -23,6 +24,7 @@ def __init__(self): self.context_is_stopped = False self.context_is_killed = False self.context_metadata: dict[str, str] = {} + self.context_cancelled = asyncio.Event() async def generate(self, request, context): print("################## generate called ######################") @@ -53,6 +55,7 @@ async def _generate_until_context_cancelled(self, request, context): self.context_is_stopped = True self.context_is_killed = context.is_killed() self.context_metadata = dict(context.metadata.items()) + self.context_cancelled.set() raise asyncio.CancelledError if context.is_killed(): @@ -60,6 +63,7 @@ async def _generate_until_context_cancelled(self, request, context): self.context_is_stopped = context.is_stopped() self.context_is_killed = True self.context_metadata = dict(context.metadata.items()) + self.context_cancelled.set() raise asyncio.CancelledError await asyncio.sleep(0.1) @@ -83,6 +87,7 @@ async def _generate_until_context_cancelled_with_metadata(self, request, context self.context_is_stopped = True self.context_is_killed = context.is_killed() self.context_metadata = dict(context.metadata.items()) + self.context_cancelled.set() raise asyncio.CancelledError if context.is_killed(): @@ -90,6 +95,7 @@ async def _generate_until_context_cancelled_with_metadata(self, request, context self.context_is_stopped = context.is_stopped() self.context_is_killed = True self.context_metadata = dict(context.metadata.items()) + self.context_cancelled.set() raise asyncio.CancelledError await asyncio.sleep(0.1) @@ -221,8 +227,7 @@ async def test_client_context_cancel(temp_file_store, server, client): iteration_count += 1 - # Give server a moment to process the cancellation - await asyncio.sleep(0.2) + await asyncio.wait_for(handler.context_cancelled.wait(), timeout=5) # Verify server detected the cancellation assert handler.context_is_stopped @@ -256,7 +261,7 @@ async def test_client_context_cancel_preserves_metadata( iteration_count += 1 - await asyncio.sleep(0.2) + await asyncio.wait_for(handler.context_cancelled.wait(), timeout=5) assert handler.context_is_stopped assert not handler.context_is_killed @@ -350,6 +355,18 @@ async def test_server_raise_cancelled(temp_file_store, server, client): assert not handler.context_is_killed +async def _wait_for_cancelled_request(request, handler): + stream = await request + # Cancellation propagates asynchronously, so responses can arrive before it completes. + try: + async for _ in stream: + pass + except ValueError as error: + if str(error) != "Cancelled: CancelledError": + raise + await handler.context_cancelled.wait() + + @pytest.mark.forked @pytest.mark.asyncio @pytest.mark.parametrize("request_plane", ["nats", "tcp"], indirect=True) @@ -358,15 +375,8 @@ async def test_client_context_already_cancelled(temp_file_store, server, client) context = Context() context.stop_generating() # TODO: (DIS-830) The outgoing call should raise if context is cancelled - stream = await client.generate("_generate_until_context_cancelled", context=context) - - async for _ in stream: - raise AssertionError( - "Request should be cancelled before any responses are generated" - ) - - # Give server a moment to update status - await asyncio.sleep(0.2) + request = client.generate("_generate_until_context_cancelled", context=context) + await asyncio.wait_for(_wait_for_cancelled_request(request, handler), timeout=5) # Verify server context cancellation status assert handler.context_is_stopped @@ -384,15 +394,7 @@ async def test_client_context_cancel_before_await_request( request = client.generate("_generate_until_context_cancelled", context=context) context.stop_generating() # TODO: (DIS-830) The outgoing call should raise if context is cancelled - stream = await request - - async for _ in stream: - raise AssertionError( - "Request should be cancelled before any responses are generated" - ) - - # Give server a moment to update status - await asyncio.sleep(0.2) + await asyncio.wait_for(_wait_for_cancelled_request(request, handler), timeout=5) # Verify server context cancellation status assert handler.context_is_stopped