test(channels): await artifact replies between sequential turns (#7616)

This commit is contained in:
Willem Jiang 2026-10-27 17:51:11 +08:00 committed by GitHub
parent 03afd05917
commit dbd72df405
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 29 additions and 15 deletions

View File

@ -490,3 +490,10 @@ Multi-instance harness tests use `support.shell.require_script_bash()` for shell
smoke checks. Keep config content and secret stability checks portable; isolate
POSIX owner-only permission assertions so Windows skips only those checks.
Nginx path expectations must include quoted-string escaping for native paths.
## Channel artifact turn sequencing
`test_channels.py` models consecutive artifact-producing turns by awaiting an
async outbound callback before sending the next inbound message. Keep the real
worker-pool concurrency, preserve per-turn artifact assertions, and stop the
manager in `finally` so a failure cannot leave tasks using a closed test loop.

View File

@ -7797,22 +7797,29 @@ class TestHandleChatWithArtifacts:
manager._client = mock_client
outbound_received = []
bus.subscribe_outbound(lambda msg: outbound_received.append(msg))
outbound_queue: asyncio.Queue[OutboundMessage] = asyncio.Queue()
async def record_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
await outbound_queue.put(msg)
bus.subscribe_outbound(record_outbound)
await manager.start()
# Send two messages with the same topic_id (same thread)
for text in ["make report", "add chart"]:
msg = InboundMessage(
channel_name="test",
chat_id="c1",
user_id="u1",
text=text,
topic_id="topic-dup",
)
await bus.publish_inbound(msg)
await _wait_for(lambda: len(outbound_received) >= 2)
await manager.stop()
try:
# Model completed turns on one thread; concurrent worker replies
# can arrive in either order after off-loop artifact preparation.
for text in ["make report", "add chart"]:
msg = InboundMessage(
channel_name="test",
chat_id="c1",
user_id="u1",
text=text,
topic_id="topic-dup",
)
await bus.publish_inbound(msg)
await asyncio.wait_for(outbound_queue.get(), timeout=5.0)
finally:
await manager.stop()
assert len(outbound_received) == 2