[CI/Build] Restore streaming executor fixture contract (#8418)

Signed-off-by: princepride <wangzhipeng628@gmail.com>
This commit is contained in:
汪志鹏 authored and GitHub committed 2026-10-02 01:45:56 -07:00
1 parent 53c7340668
commit b742f86136
1 file changed
+5 -4
@@ -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 = {}