wehub-resource-sync
a7d6d88f6f
CI / changes (push) Has been cancelled
CI / cd libs/checkpoint (push) Has been cancelled
CI / cd libs/checkpoint-conformance (push) Has been cancelled
CI / cd libs/checkpoint-postgres (push) Has been cancelled
CI / cd libs/checkpoint-sqlite (push) Has been cancelled
CI / cd libs/cli (push) Has been cancelled
CI / cd libs/prebuilt (push) Has been cancelled
CI / cd libs/sdk-py (push) Has been cancelled
CI / cd libs/langgraph (push) Has been cancelled
CI / Check SDK methods matching (push) Has been cancelled
CI / Check CLI schema hasn't changed #3.13 (push) Has been cancelled
CI / CLI integration test (push) Has been cancelled
CI / sdk-py integration test (push) Has been cancelled
CI / CI Success (push) Has been cancelled
baseline / benchmark (push) Has been cancelled
Deploy Redirects to GitHub Pages / deploy (push) Has been cancelled
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""Concurrent `threads.stream()` against the integration API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from .conftest import ASSISTANT_ID, EXPECTED_TERMINAL_ITEMS
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
async def _drive_one_async(threads: Any) -> dict[str, Any]:
|
|
async with threads.stream(assistant_id=ASSISTANT_ID) as thread:
|
|
await thread.run.start(input={"messages": [], "value": "init", "items": []})
|
|
|
|
async for _ in thread.values:
|
|
if thread.interrupted:
|
|
break
|
|
if thread.interrupted:
|
|
await thread.run.respond("yes")
|
|
|
|
final = await thread.output
|
|
return {"thread_id": thread.thread_id, "items": final.get("items")}
|
|
|
|
|
|
async def test_concurrent_streams_async(async_threads) -> None:
|
|
threads, _ = async_threads
|
|
a, b = await asyncio.gather(_drive_one_async(threads), _drive_one_async(threads))
|
|
assert a["items"] == EXPECTED_TERMINAL_ITEMS
|
|
assert b["items"] == EXPECTED_TERMINAL_ITEMS
|
|
assert a["thread_id"] != b["thread_id"], (
|
|
f"concurrent streams collided on thread_id {a['thread_id']!r}"
|
|
)
|
|
|
|
|
|
def _drive_one_sync(
|
|
threads: Any, label: str, results: dict[str, dict[str, Any]]
|
|
) -> None:
|
|
with threads.stream(assistant_id=ASSISTANT_ID) as thread:
|
|
thread.run.start(input={"messages": [], "value": "init", "items": []})
|
|
|
|
for _ in thread.values:
|
|
if thread.interrupted:
|
|
break
|
|
if thread.interrupted:
|
|
thread.run.respond("yes")
|
|
|
|
final = thread.output
|
|
results[label] = {"thread_id": thread.thread_id, "items": final.get("items")}
|
|
|
|
|
|
def test_concurrent_streams_sync(sync_threads) -> None:
|
|
threads, _ = sync_threads
|
|
results: dict[str, dict[str, Any]] = {}
|
|
workers = [
|
|
threading.Thread(
|
|
target=_drive_one_sync,
|
|
args=(threads, label, results),
|
|
daemon=True,
|
|
name=f"sync-stream-{label}",
|
|
)
|
|
for label in ("A", "B")
|
|
]
|
|
for w in workers:
|
|
w.start()
|
|
for w in workers:
|
|
w.join(timeout=60)
|
|
assert not w.is_alive(), f"worker {w.name} did not finish within 60s"
|
|
|
|
a = results.get("A")
|
|
b = results.get("B")
|
|
assert a is not None and a["items"] == EXPECTED_TERMINAL_ITEMS
|
|
assert b is not None and b["items"] == EXPECTED_TERMINAL_ITEMS
|
|
assert a["thread_id"] != b["thread_id"], (
|
|
f"concurrent streams collided on thread_id {a['thread_id']!r}"
|
|
)
|