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
25 lines
821 B
Python
25 lines
821 B
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
|
|
from streaming._fake_server import FakeServer
|
|
from streaming.assert_transport_replays import assert_transport_replays
|
|
|
|
|
|
class _Harness:
|
|
def __init__(self, fake: FakeServer, transport: ProtocolSseTransport) -> None:
|
|
self._fake = fake
|
|
self.transport = transport
|
|
|
|
def script_buffered(self, events):
|
|
self._fake.script(events)
|
|
|
|
|
|
async def test_fake_server_replays_buffered_events():
|
|
fake = FakeServer()
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as client:
|
|
transport = ProtocolSseTransport(client=client, thread_id="t-1")
|
|
await assert_transport_replays(_Harness(fake, transport))
|