m4
d4b53bfb08
Docker / build (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Build / build (push) Has been cancelled
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
"""Real loopback HTTP cancellation, without a provider or business run."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from langchain_core.messages import HumanMessage
|
|
from langgraph.graph import END, START, StateGraph
|
|
|
|
from EvoScientist.llm.gateway_proxy import GatewayProxyChatModel
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_model_cancel_closes_http_sse_reader(monkeypatch):
|
|
monkeypatch.setenv("AI4SCI_EVO_RUNTIME_GRANT_SECRET", "test-only-secret")
|
|
connected = asyncio.Event()
|
|
disconnected = asyncio.Event()
|
|
|
|
async def serve(reader, writer):
|
|
try:
|
|
headers = await reader.readuntil(b"\r\n\r\n")
|
|
length = next(int(line.split(b":", 1)[1]) for line in headers.splitlines()
|
|
if line.lower().startswith(b"content-length:"))
|
|
await reader.readexactly(length)
|
|
writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n"
|
|
b"Transfer-Encoding: chunked\r\n\r\n")
|
|
await writer.drain()
|
|
connected.set()
|
|
assert await reader.read() == b""
|
|
disconnected.set()
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
server = await asyncio.start_server(serve, "127.0.0.1", 0)
|
|
port = server.sockets[0].getsockname()[1]
|
|
model = GatewayProxyChatModel(gateway_url=f"http://127.0.0.1:{port}",
|
|
run_id="test", envelope_signature="test")
|
|
|
|
async def consume(_state):
|
|
async for _ in model.astream([HumanMessage(content="test")]):
|
|
pytest.fail("silent test stream should produce no model output")
|
|
return {}
|
|
|
|
builder = StateGraph(dict)
|
|
builder.add_node("model", consume)
|
|
builder.add_edge(START, "model")
|
|
builder.add_edge("model", END)
|
|
task = asyncio.create_task(builder.compile().ainvoke({}))
|
|
try:
|
|
await asyncio.wait_for(connected.wait(), 3)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
await asyncio.wait_for(disconnected.wait(), 3)
|
|
finally:
|
|
task.cancel()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
server.close()
|
|
await server.wait_closed() |