mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-26 09:32:25 +02:00
Nuno CamposandGitHub
eeadeb282e
## Summary - ensure both async and sync HTTP clients flush the SSE decoder after streaming - add regression tests covering trailing SSE events without a terminating blank line ## Testing - make format - make lint - make test ------ https://chatgpt.com/codex/tasks/task_e_68c9727ca9f8832d9f207323c5e02a72
82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from langgraph_sdk.client import HttpClient, SyncHttpClient
|
|
from langgraph_sdk.schema import StreamPart
|
|
from langgraph_sdk.sse import BytesLike, BytesLineDecoder, SSEDecoder
|
|
|
|
with open(Path(__file__).parent / "fixtures" / "response.txt", "rb") as f:
|
|
RESPONSE_PAYLOAD = f.read()
|
|
|
|
|
|
def iter_lines_raw(payload: list[bytes]) -> Iterator[BytesLike]:
|
|
decoder = BytesLineDecoder()
|
|
for part in payload:
|
|
yield from decoder.decode(part)
|
|
yield from decoder.flush()
|
|
|
|
|
|
def test_stream_see():
|
|
for groups in (
|
|
[RESPONSE_PAYLOAD],
|
|
RESPONSE_PAYLOAD.splitlines(keepends=True),
|
|
):
|
|
parts: list[StreamPart] = []
|
|
|
|
decoder = SSEDecoder()
|
|
for line in iter_lines_raw(groups):
|
|
sse = decoder.decode(line=line.rstrip(b"\n"))
|
|
if sse is not None:
|
|
parts.append(sse)
|
|
if sse := decoder.decode(b""):
|
|
parts.append(sse)
|
|
|
|
assert decoder.decode(b"") is None
|
|
assert len(parts) == 79
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_client_stream_flushes_trailing_event():
|
|
payload = b'event: foo\ndata: {"bar": 1}\n'
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["accept"] == "text/event-stream"
|
|
assert request.headers["cache-control"] == "no-store"
|
|
return httpx.Response(
|
|
200,
|
|
headers={"Content-Type": "text/event-stream"},
|
|
content=payload,
|
|
)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
async with httpx.AsyncClient(
|
|
transport=transport, base_url="https://example.com"
|
|
) as client:
|
|
http_client = HttpClient(client)
|
|
parts = [part async for part in http_client.stream("/stream", "GET")]
|
|
|
|
assert parts == [StreamPart(event="foo", data={"bar": 1})]
|
|
|
|
|
|
def test_sync_http_client_stream_flushes_trailing_event():
|
|
payload = b'event: foo\ndata: {"bar": 1}\n'
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.headers["accept"] == "text/event-stream"
|
|
assert request.headers["cache-control"] == "no-store"
|
|
return httpx.Response(
|
|
200,
|
|
headers={"Content-Type": "text/event-stream"},
|
|
content=payload,
|
|
)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
with httpx.Client(transport=transport, base_url="https://example.com") as client:
|
|
http_client = SyncHttpClient(client)
|
|
parts = list(http_client.stream("/stream", "GET"))
|
|
|
|
assert parts == [StreamPart(event="foo", data={"bar": 1})]
|