From b742f86136d28423903a1213adde2a62a11067a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=AA=E5=BF=97=E9=B9=8F?= Date: Fri, 2 Oct 2026 16:45:56 +0800 Subject: [PATCH] [CI/Build] Restore streaming executor fixture contract (#8418) Signed-off-by: princepride --- .../test_diffusion_streaming_output.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/diffusion/streaming_interactive_generation/test_diffusion_streaming_output.py b/tests/diffusion/streaming_interactive_generation/test_diffusion_streaming_output.py index 68d17d49f..e9514abc9 100644 --- a/tests/diffusion/streaming_interactive_generation/test_diffusion_streaming_output.py +++ b/tests/diffusion/streaming_interactive_generation/test_diffusion_streaming_output.py @@ -77,6 +77,7 @@ class _PipelineBackedEngine: def __init__(self, pipeline: _StepStreamingPipeline | _FailingStreamingPipeline) -> None: self.pipeline = pipeline self.executor = SimpleNamespace( + is_dead=False, register_failure_callback=MagicMock(), check_health=MagicMock(), ) @@ -127,9 +128,9 @@ def _streaming_diffusion_output( def _streaming_payload_and_metadata(output: DiffusionOutput) -> tuple[dict, dict]: envelope = output.output if isinstance(output.output, dict) else {} - payload = envelope.get("payload") if isinstance(envelope.get("payload"), dict) else {} - metadata = envelope.get("metadata") if isinstance(envelope.get("metadata"), dict) else {} - return payload, metadata + payload = envelope.get("payload") + metadata = envelope.get("metadata") + return payload if isinstance(payload, dict) else {}, metadata if isinstance(metadata, dict) else {} class TestPipelineStreamingOutputToStageDiffusionClient: @@ -230,7 +231,7 @@ class TestPipelineStreamingOutputToStageDiffusionClient: client._response_socket.bind(response_address) client._encoder = OmniMsgpackEncoder() client._decoder = OmniMsgpackDecoder() - client._output_queue: asyncio.Queue[OmniRequestOutput] = asyncio.Queue() + client._output_queue = asyncio.Queue[OmniRequestOutput]() client._rpc_results = {} client._pending_rpcs = set() client._tasks = {}