mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-31 12:19:58 +02:00
Elior Nataf LackritzandGitHub
ea5f9cc9fb
Follow-up to #8540, which turned on `PLC0415` (import-outside-top-level) for checkpoint-postgres and checkpoint-sqlite. This does the remaining six packages: checkpoint, checkpoint-conformance, langgraph, prebuilt, cli, sdk-py. Scoped to tests, per @sydney-runkle's call on #8540: library code is exempted with `per-file-ignores`, since it still has deferred imports nobody has reviewed and mixing that in would make this hard to read. ## What changed Function-level imports across 56 test files moved to module level. Nine could not move and carry an explicit `# noqa: PLC0415` with a reason: | File | Why it stays local | |---|---| | `libs/langgraph/tests/test_deprecation.py` (4) | the import has to run inside `pytest.warns` for the warning to be observed | | `libs/langgraph/tests/test_serde_allowlist.py` | try/except guard, skips when langchain_core is absent | | `libs/langgraph/tests/test_delta_channel_benchmark.py` | optional psycopg probe | | `libs/checkpoint/tests/test_conformance_delta.py` (3) | protected by a module-level `pytest.importorskip`; hoisting past the guard turns a skip into a collection error | That last one is the trap: an import moved above `pytest.importorskip` silently defeats the guard. I hit it locally and it turned the skip into a `ModuleNotFoundError` at collection. Every file with an `importorskip` or `except ImportError` was checked by hand for this. ## Verification `make lint` and `make test` in each of the six: | Package | Tests | |---|---| | checkpoint | 156 passed, 17 skipped | | checkpoint-conformance | 1 passed | | langgraph | 1968 passed, 4 skipped | | prebuilt | 284 passed | | cli | 336 passed | | sdk-py | 493 passed | Also confirmed the rule actually fires: a throwaway test file with a function-level import is flagged in all six packages, and the source exemption holds.
271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""Tests for `thread.tool_calls` - typed async tool-call projection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from langgraph_sdk._async.http import HttpClient
|
|
from langgraph_sdk._async.stream import ToolCallHandle
|
|
from langgraph_sdk._async.threads import ThreadsClient
|
|
from streaming._events import (
|
|
lifecycle_completed_event,
|
|
lifecycle_errored_event,
|
|
lifecycle_started_event,
|
|
tool_error_event,
|
|
tool_finished_event,
|
|
tool_output_delta_event,
|
|
tool_started_event,
|
|
)
|
|
from streaming._fake_server import FakeServer
|
|
|
|
|
|
async def test_tool_calls_subscribes_to_tools_channel():
|
|
fake = FakeServer()
|
|
fake.script([lifecycle_completed_event(seq=1)])
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
_ = [call async for call in thread.tool_calls]
|
|
|
|
assert any(
|
|
"tools" in body.get("channels", []) for body in fake.stream_request_bodies
|
|
)
|
|
|
|
|
|
async def test_tool_calls_yields_handle_deltas_and_output():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(
|
|
seq=1,
|
|
tool_call_id="call-1",
|
|
tool_name="search",
|
|
input={"query": "sf weather"},
|
|
),
|
|
tool_output_delta_event(seq=2, tool_call_id="call-1", delta="part "),
|
|
tool_output_delta_event(seq=3, tool_call_id="call-1", delta="two"),
|
|
tool_finished_event(
|
|
seq=4,
|
|
tool_call_id="call-1",
|
|
output={"temperature": 68},
|
|
),
|
|
lifecycle_completed_event(seq=5),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
assert len(calls) == 1
|
|
call = calls[0]
|
|
assert call.tool_call_id == "call-1"
|
|
assert call.name == "search"
|
|
assert call.input == {"query": "sf weather"}
|
|
assert call.namespace == []
|
|
assert call.done is True
|
|
assert [delta async for delta in call.deltas] == ["part ", "two"]
|
|
assert await call.output == {"temperature": 68}
|
|
|
|
|
|
async def test_tool_calls_multiple_concurrent_calls_route_by_id():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, tool_call_id="call-a", tool_name="alpha"),
|
|
tool_started_event(seq=2, tool_call_id="call-b", tool_name="beta"),
|
|
tool_output_delta_event(seq=3, tool_call_id="call-b", delta="b1"),
|
|
tool_output_delta_event(seq=4, tool_call_id="call-a", delta="a1"),
|
|
tool_finished_event(seq=5, tool_call_id="call-a", output="A"),
|
|
tool_finished_event(seq=6, tool_call_id="call-b", output="B"),
|
|
lifecycle_completed_event(seq=7),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
by_id = {call.tool_call_id: call for call in calls}
|
|
assert set(by_id) == {"call-a", "call-b"}
|
|
assert [delta async for delta in by_id["call-a"].deltas] == ["a1"]
|
|
assert [delta async for delta in by_id["call-b"].deltas] == ["b1"]
|
|
assert await by_id["call-a"].output == "A"
|
|
assert await by_id["call-b"].output == "B"
|
|
|
|
|
|
async def test_tool_calls_ignores_nested_namespace_for_root_projection():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, namespace=["child:1"], tool_call_id="nested"),
|
|
tool_finished_event(seq=2, namespace=["child:1"], tool_call_id="nested"),
|
|
lifecycle_completed_event(seq=3),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
assert calls == []
|
|
|
|
|
|
async def test_tool_calls_error_event_fails_output_and_deltas():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, tool_call_id="call-1"),
|
|
tool_output_delta_event(seq=2, tool_call_id="call-1", delta="before"),
|
|
tool_error_event(seq=3, tool_call_id="call-1", message="boom"),
|
|
lifecycle_completed_event(seq=4),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
assert len(calls) == 1
|
|
assert [delta async for delta in calls[0].deltas] == ["before"]
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
await calls[0].output
|
|
|
|
|
|
async def test_tool_calls_run_error_fails_active_handle():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, tool_call_id="call-1"),
|
|
lifecycle_errored_event(seq=2, error="run failed"),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
assert len(calls) == 1
|
|
with pytest.raises(RuntimeError, match="Run errored: run failed"):
|
|
await calls[0].output
|
|
|
|
|
|
async def test_tool_calls_stream_end_fails_active_handle():
|
|
fake = FakeServer()
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, tool_call_id="call-1"),
|
|
lifecycle_completed_event(seq=2),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
calls = [call async for call in thread.tool_calls]
|
|
|
|
assert len(calls) == 1
|
|
with pytest.raises(RuntimeError, match="closed before terminal tool event"):
|
|
await calls[0].output
|
|
|
|
|
|
async def test_tool_calls_explicit_aclose_does_not_block_1s():
|
|
"""Explicitly closing the tool_calls iterator must return in <500ms.
|
|
|
|
The old finally block did `await asyncio.wait_for(asyncio.shield(run_done),
|
|
timeout=1.0)` unconditionally. When the caller explicitly calls aclose() on
|
|
the generator before any lifecycle terminal event arrives, this caused a
|
|
mandatory 1-second stall per iterator close.
|
|
"""
|
|
fake = FakeServer()
|
|
# Script has a started lifecycle and one tool, but NO terminal lifecycle.
|
|
# If the shield-wait is present, aclose() will block for 1s.
|
|
fake.script(
|
|
[
|
|
lifecycle_started_event(seq=0),
|
|
tool_started_event(seq=1, tool_call_id="call-1"),
|
|
]
|
|
)
|
|
asgi = httpx.ASGITransport(app=fake.app)
|
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
|
threads = ThreadsClient(HttpClient(raw))
|
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
|
await thread.run.start(input={})
|
|
# _tool_calls_iter() is an AsyncGenerator; cast so the type checker
|
|
# knows aclose() is available without a bare AsyncIterator protocol.
|
|
|
|
gen: AsyncGenerator = thread.tool_calls._tool_calls_iter()
|
|
_call = await gen.__anext__() # receive the one tool-started handle
|
|
start = time.monotonic()
|
|
await gen.aclose() # explicitly close — must not stall 1s
|
|
elapsed = time.monotonic() - start
|
|
assert elapsed < 0.5, f"tool_calls aclose() took {elapsed:.3f}s (expected <0.5s)"
|
|
|
|
|
|
def test_tool_call_handle_deltas_queue_is_bounded():
|
|
"""ToolCallHandle._deltas must be constructed with a bounded asyncio.Queue.
|
|
|
|
Unbounded queues allow producers to enqueue indefinitely, causing memory
|
|
growth when consumers are slow.
|
|
"""
|
|
|
|
# We need a running loop to create the Future inside ToolCallHandle.__init__.
|
|
async def _make() -> None:
|
|
|
|
handle_default = ToolCallHandle(tool_call_id="tc1", name="foo")
|
|
assert handle_default._deltas.maxsize > 0, (
|
|
"default maxsize must be positive (bounded)"
|
|
)
|
|
|
|
handle_custom = ToolCallHandle(tool_call_id="tc2", name="bar", max_queue_size=8)
|
|
assert handle_custom._deltas.maxsize == 8
|
|
|
|
asyncio.run(_make())
|
|
|
|
|
|
def test_tool_call_handle_deltas_single_consumer_guard():
|
|
"""Accessing `handle.deltas` a second time must raise immediately.
|
|
|
|
`_deltas` is a single-consumer queue; fanning out to multiple consumers
|
|
would cause each consumer to miss events already consumed by the other.
|
|
The property must raise before returning the iterator so the caller
|
|
sees the error even without iterating.
|
|
"""
|
|
|
|
async def _run() -> None:
|
|
|
|
handle = ToolCallHandle(tool_call_id="tc1", name="foo")
|
|
|
|
# First access: fine — returns the iterator.
|
|
_iter_1 = handle.deltas
|
|
|
|
# Second access: must raise immediately (before any iteration).
|
|
with pytest.raises(RuntimeError, match="single consumer"):
|
|
_ = handle.deltas
|
|
|
|
asyncio.run(_run())
|