WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: enforce PLC0415 in tests for the remaining packages (#8547)

Browse Source 
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.
This commit is contained in:
Elior Nataf Lackritz
2026-08-07 09:40:18 -04:00
committed by GitHub
parent 36a505ac65
commit ea5f9cc9fb
62 changed files with 274 additions and 537 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint-conformance/pyproject.toml
+5
View File
@@ -58,9 +58,14 @@ lint.select = [
"UP", # pyupgrade
"B", # flake8-bugbear
"I", # isort
"PLC0415", # import-outside-top-level
"RUF100", # unused noqa directive
]
lint.ignore = ["E501", "B008"]
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
lint.per-file-ignores = { "langgraph/**" = ["PLC0415"] }
target-version = "py310"
[tool.uv.sources]
libs/checkpoint/pyproject.toml
+5
View File
@@ -59,10 +59,15 @@ lint.select = [
"UP", # pyupgrade
"B", # flake8-bugbear
"I", # isort
"PLC0415", # import-outside-top-level
"RUF100", # unused noqa directive
"UP", # pyupgrade
]
lint.ignore = ["E501", "B008"]
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
lint.per-file-ignores = { "langgraph/**" = ["PLC0415"] }
target-version = "py310"
[tool.ty.rules]
libs/checkpoint/tests/test_conformance_delta.py
+7 -3
View File
@@ -12,10 +12,14 @@ conformance = pytest.importorskip(
@pytest.mark.asyncio
async def test_delta_channel_conformance():
from langgraph.checkpoint.conformance import validate
from langgraph.checkpoint.conformance.initializer import checkpointer_test
# Imported inside the test: the module-level importorskip above is what
# makes these safe, so they cannot move to the top of the file.
from langgraph.checkpoint.conformance import validate # noqa: PLC0415
from langgraph.checkpoint.conformance.initializer import ( # noqa: PLC0415
checkpointer_test,
)
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.memory import InMemorySaver # noqa: PLC0415
@checkpointer_test(name="InMemorySaver")
async def mem_saver():
libs/checkpoint/tests/test_encrypted.py
-2
View File
@@ -307,8 +307,6 @@ class TestWithMsgpackAllowlistEncrypted:
def loads_typed(self, data: tuple[str, bytes]) -> None:
return None
from langgraph.checkpoint.serde.base import CipherProtocol
class DummyCipher(CipherProtocol):
def encrypt(self, plaintext: bytes) -> tuple[str, bytes]:
return "dummy", plaintext
libs/checkpoint/tests/test_jsonplus.py
+4 -9
View File
@@ -1,9 +1,12 @@
import dataclasses
import json
import logging
import os
import pathlib
import pickle
import re
import sys
import tempfile
import uuid
from collections import deque
from datetime import date, datetime, time, timezone
@@ -18,7 +21,7 @@ import ormsgpack
import pandas as pd
import pytest
from langchain_core.documents.base import Document
from langchain_core.messages import HumanMessage
from langchain_core.messages import AIMessage, HumanMessage
from pydantic import BaseModel, SecretStr
from pydantic.v1 import BaseModel as BaseModelV1
from pydantic.v1 import SecretStr as SecretStrV1
@@ -341,7 +344,6 @@ def test_lc2_json_safe_type_revives_without_allowlist() -> None:
constructor dicts. Resuming those threads must reconstruct proper BaseMessage objects
rather than returning raw dicts that cause MESSAGE_COERCION_FAILURE in add_messages.
"""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer() # default: _allowed_json_modules=None
@@ -410,7 +412,6 @@ def test_lc2_json_method_field_is_ignored() -> None:
to that method: the result is whatever ``AIMessage(*args, **kwargs)`` would
produce, which proves the default constructor ran instead of ``parse_raw``.
"""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer()
load = {
@@ -436,7 +437,6 @@ def test_lc2_json_method_field_is_ignored_for_allowlisted_types() -> None:
method dispatch as a side effect. Revival is restricted to the default
constructor regardless of how the class reached the revival path.
"""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer(
allowed_json_modules=[("langchain_core.messages.ai", "AIMessage")]
@@ -455,7 +455,6 @@ def test_lc2_json_method_field_is_ignored_for_allowlisted_types() -> None:
def test_lc2_json_safe_type_init_still_works() -> None:
"""SAFE-type lc=2 revival without a `method` field still constructs the class."""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer()
load = {
@@ -479,7 +478,6 @@ def test_lc2_json_legacy_pydantic_method_list_falls_back_to_default() -> None:
this shape continue to revive correctly as long as the default constructor
accepts the serialized kwargs.
"""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer()
load = {
@@ -551,9 +549,6 @@ def test_lc2_json_safe_type_pickle_payload_does_not_execute() -> None:
With method dispatch removed from `_revive_lc2`, the gadget bytes are never
passed to `parse_raw` and therefore never reach `pickle.loads`.
"""
import os
import pickle
import tempfile
marker = tempfile.NamedTemporaryFile(
prefix="lc2_block_proof_", suffix=".out", delete=False
libs/checkpoint/tests/test_memory.py
+1 -1
View File
@@ -1,3 +1,4 @@
import asyncio
import logging
from typing import Any
@@ -523,7 +524,6 @@ class TestBaseFallbackGetChannelWrites:
`threading.local()` guard would let whichever task set it first
short-circuit the other to `writes=[]`.
"""
import asyncio
saver, thread_id, ns = self._build_saver_with_chain()
libs/cli/pyproject.toml
+5
View File
@@ -72,10 +72,15 @@ lint.select = [
"UP", # pyupgrade
"B", # flake8-bugbear
"I", # isort
"PLC0415", # import-outside-top-level
"RUF100", # unused noqa directive
"UP", # pyupgrade
]
lint.ignore = ["E501", "B008"]
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
lint.per-file-ignores = { "langgraph_cli/**" = ["PLC0415"], "generate_schema.py" = ["PLC0415"] }
target-version = "py310"
[tool.ty.rules]
libs/cli/tests/unit_tests/test_archive.py
+1 -6
View File
@@ -11,6 +11,7 @@ from langgraph_cli.archive import (
_tar_filter,
create_archive,
)
from langgraph_cli.config import LocalDeps
# ---------------------------------------------------------------------------
# _tar_filter
@@ -198,7 +199,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
def test_yields_archive_with_config(self, mock_deps, tmp_path):
from langgraph_cli.config import LocalDeps
config_file = self._make_project(tmp_path)
mock_deps.return_value = LocalDeps(
@@ -218,7 +218,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
def test_excludes_pycache(self, mock_deps, tmp_path):
from langgraph_cli.config import LocalDeps
config_file = self._make_project(tmp_path)
mock_deps.return_value = LocalDeps(
@@ -232,7 +231,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
def test_cleans_up_tmp_dir_on_normal_exit(self, mock_deps, tmp_path):
from langgraph_cli.config import LocalDeps
config_file = self._make_project(tmp_path)
mock_deps.return_value = LocalDeps(
@@ -247,7 +245,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
def test_cleans_up_tmp_dir_on_exception(self, mock_deps, tmp_path):
from langgraph_cli.config import LocalDeps
config_file = self._make_project(tmp_path)
mock_deps.return_value = LocalDeps(
@@ -264,7 +261,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
@patch("langgraph_cli.archive._MAX_SIZE", 10)
def test_raises_on_oversized_archive(self, mock_deps, tmp_path):
from langgraph_cli.config import LocalDeps
config_file = self._make_project(tmp_path)
mock_deps.return_value = LocalDeps(
@@ -278,7 +274,6 @@ class TestCreateArchive:
@patch("langgraph_cli.archive._assemble_local_deps")
def test_handles_extra_contexts(self, mock_deps, tmp_path):
"""Monorepo case: project + sibling dependency directory."""
from langgraph_cli.config import LocalDeps
project = tmp_path / "myproject"
project.mkdir()
libs/cli/tests/unit_tests/test_deploy_helpers.py
-3
View File
@@ -347,7 +347,6 @@ class TestCallHostBackendWithOptionalTenant:
def test_workspace_prompt_blocked_by_no_input(self, monkeypatch):
"""With _no_input=True, 403 requiring workspace should raise ClickException."""
import langgraph_cli.deploy as deploy_mod
monkeypatch.setattr(deploy_mod, "_no_input", True)
@@ -515,7 +514,6 @@ class TestEmitterTextMode:
class TestCreateHostBackendClientNoInput:
def test_raises_when_no_api_key_and_no_input(self, monkeypatch, tmp_path):
import langgraph_cli.deploy as deploy_mod
monkeypatch.setattr(deploy_mod, "_no_input", True)
monkeypatch.delenv("LANGSMITH_API_KEY", raising=False)
@@ -530,7 +528,6 @@ class TestCreateHostBackendClientNoInput:
)
def test_succeeds_with_api_key_in_env(self, monkeypatch, tmp_path):
import langgraph_cli.deploy as deploy_mod
monkeypatch.setattr(deploy_mod, "_no_input", True)
monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
libs/langgraph/pyproject.toml
+5 -1
View File
@@ -89,8 +89,12 @@ langgraph-sdk = { path = "../sdk-py", editable = true }
langgraph-cli = { path = "../cli", editable = true }
[tool.ruff]
lint.select = [ "E", "F", "I", "RUF100", "TID251", "UP" ]
lint.select = [ "E", "F", "I", "PLC0415", "RUF100", "TID251", "UP" ]
lint.ignore = [ "E501" ]
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
lint.per-file-ignores = { "langgraph/**" = ["PLC0415"] }
line-length = 88
indent-width = 4
extend-include = ["*.ipynb"]
libs/langgraph/tests/memory_assert.py
+1 -2
View File
@@ -1,5 +1,6 @@
import os
import tempfile
import time
from collections import defaultdict
from functools import partial
from typing import Any
@@ -73,8 +74,6 @@ class MemorySaverAssertImmutable(InMemorySaver):
new_versions: ChannelVersions,
) -> None:
if self.put_sleep:
import time
time.sleep(self.put_sleep)
# assert checkpoint hasn't been modified since last written
thread_id = config["configurable"]["thread_id"]
libs/langgraph/tests/test_channels.py
+3 -8
View File
@@ -2,14 +2,16 @@ import operator
from collections.abc import Sequence
from typing import Annotated
import orjson
import pytest
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.serde.types import _DeltaSnapshot
from typing_extensions import NotRequired, TypedDict
from langgraph._internal._constants import OVERWRITE
from langgraph._internal._typing import MISSING
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.binop import BinaryOperatorAggregate, _get_overwrite
from langgraph.channels.delta import DeltaChannel
from langgraph.channels.last_value import LastValue
from langgraph.channels.topic import Topic
@@ -194,10 +196,6 @@ def test_overwrite_dataclass_form_survives_json_roundtrip() -> None:
...}`) is indistinguishable from a literal channel value, and downstream
reducers raise `MESSAGE_COERCION_FAILURE` (or similar) on read.
"""
import orjson
from langgraph._internal._constants import OVERWRITE
from langgraph.channels.binop import _get_overwrite
ow = Overwrite(value=[HumanMessage(content="new", id="h2")])
erased = orjson.loads(orjson.dumps(ow, default=lambda o: o.model_dump()))
@@ -213,8 +211,6 @@ def test_overwrite_sentinel_dict_still_recognised() -> None:
"""The pre-existing `{"__overwrite__": value}` dict form continues to be
recognised. This is the canonical sentinel emitted by producers that do
not have an `Overwrite` dataclass available."""
from langgraph._internal._constants import OVERWRITE
from langgraph.channels.binop import _get_overwrite
is_overwrite, value = _get_overwrite({OVERWRITE: ["b"]})
assert is_overwrite
@@ -224,7 +220,6 @@ def test_overwrite_sentinel_dict_still_recognised() -> None:
def test_overwrite_non_matching_dict_not_recognised() -> None:
"""Dicts that resemble the erased shape but do not carry the
`__overwrite__` discriminator must not be misclassified as overwrites."""
from langgraph.channels.binop import _get_overwrite
assert _get_overwrite({"value": ["b"]}) == (False, None)
assert _get_overwrite({"type": "human", "value": "hi"}) == (False, None)
libs/langgraph/tests/test_delta_channel_benchmark.py
+1 -1
View File
@@ -220,7 +220,7 @@ def _checkpointers() -> list[tuple[str, Any]]:
result: list[tuple[str, Any]] = [("InMemory", None)]
if _POSTGRES_AVAILABLE:
try:
import psycopg
import psycopg # noqa: PLC0415
psycopg.connect(_POSTGRES_URI).close()
result.append(("Postgres", "postgres"))
libs/langgraph/tests/test_delta_channel_update_state.py
+1 -1
View File
@@ -27,6 +27,7 @@ from typing_extensions import TypedDict
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import _messages_delta_reducer
from langgraph.types import StateUpdate
pytestmark = pytest.mark.anyio
@@ -277,7 +278,6 @@ def test_bulk_update_state_multi_task_per_superstep_delta_channel() -> None:
different `StateUpdate`s targeting the same node — otherwise both share
the deterministic interrupt-derived id and collide in the saver.
"""
from langgraph.types import StateUpdate
saver = InMemorySaver()
graph = _build_graph(saver)
libs/langgraph/tests/test_deprecation.py
+4 -4
View File
@@ -88,13 +88,13 @@ def test_constants_deprecation() -> None:
LangGraphDeprecatedSinceV10,
match="Importing Send from langgraph.constants is deprecated. Please use 'from langgraph.types import Send' instead.",
):
from langgraph.constants import Send # noqa: F401
from langgraph.constants import Send # noqa: PLC0415, F401
with pytest.warns(
LangGraphDeprecatedSinceV10,
match="Importing Interrupt from langgraph.constants is deprecated. Please use 'from langgraph.types import Interrupt' instead.",
):
from langgraph.constants import Interrupt # noqa: F401
from langgraph.constants import Interrupt # noqa: PLC0415, F401
def test_pregel_types_deprecation() -> None:
@@ -102,7 +102,7 @@ def test_pregel_types_deprecation() -> None:
LangGraphDeprecatedSinceV10,
match="Importing from langgraph.pregel.types is deprecated. Please use 'from langgraph.types import ...' instead.",
):
from langgraph.pregel.types import StateSnapshot # noqa: F401
from langgraph.pregel.types import StateSnapshot # noqa: PLC0415, F401
def test_config_schema_deprecation() -> None:
@@ -195,7 +195,7 @@ def test_deprecated_import() -> None:
LangGraphDeprecatedSinceV10,
match="Importing PREVIOUS from langgraph.constants is deprecated. This constant is now private and should not be used directly.",
):
from langgraph.constants import PREVIOUS # noqa: F401
from langgraph.constants import PREVIOUS # noqa: PLC0415, F401
@pytest.mark.filterwarnings(
libs/langgraph/tests/test_graph_callbacks.py
+1 -1
View File
@@ -13,6 +13,7 @@ from langgraph.callbacks import (
GraphCallbackHandler,
GraphInterruptEvent,
GraphResumeEvent,
_GraphCallbackManager,
)
from langgraph.graph import START, StateGraph
from langgraph.types import Command, Interrupt, interrupt
@@ -286,7 +287,6 @@ def test_non_graph_handler_via_add_handler_does_not_crash() -> None:
GraphCallbackHandler. They must be silently accepted — graph lifecycle
events will simply not be dispatched to them.
"""
from langgraph.callbacks import _GraphCallbackManager
manager = _GraphCallbackManager()
plain_handler = _LangChainCustomEventHandler()
libs/langgraph/tests/test_large_cases.py
+16 -45
View File
@@ -2,11 +2,26 @@ import json
import operator
import re
import time
from copy import deepcopy
from dataclasses import replace
from typing import Annotated, Any, Literal, cast
import pytest
from langchain_core.messages import AIMessage, AnyMessage, ToolCall
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
ToolCall,
ToolMessage,
)
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig, RunnableMap, RunnablePick
from langchain_core.tools import tool
from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
@@ -484,9 +499,6 @@ def test_conditional_state_graph(
snapshot: SnapshotAssertion,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool
class AgentState(TypedDict, total=False):
input: Annotated[str, UntrackedValue]
@@ -1261,8 +1273,6 @@ def test_conditional_state_graph(
def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None:
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
@tool()
def search_api(query: str) -> str:
@@ -1626,17 +1636,6 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None:
def test_state_graph_packets(
sync_checkpointer: BaseCheckpointSaver, mocker: MockerFixture
) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
ToolCall,
ToolMessage,
)
from langchain_core.tools import tool
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
@@ -2381,15 +2380,6 @@ def test_message_graph(
deterministic_uuids: MockerFixture,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from copy import deepcopy
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool
class FakeFunctionChatModel(FakeMessagesListChatModel):
def bind_functions(self, functions: list):
@@ -3099,20 +3089,6 @@ def test_root_graph(
deterministic_uuids: MockerFixture,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from copy import deepcopy
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
ToolMessage,
)
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool
class FakeFunctionChatModel(FakeMessagesListChatModel):
def bind_functions(self, functions: list):
@@ -5837,7 +5813,6 @@ def test_send_to_nested_graphs(sync_checkpointer: BaseCheckpointSaver) -> None:
def test_send_react_interrupt(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
ai_message = AIMessage(
"",
@@ -6228,7 +6203,6 @@ def test_send_react_interrupt(
def test_send_react_interrupt_control(
sync_checkpointer: BaseCheckpointSaver, snapshot: SnapshotAssertion
) -> None:
from langchain_core.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
ai_message = AIMessage(
"",
@@ -6455,9 +6429,6 @@ def test_send_react_interrupt_control(
def test_weather_subgraph(
sync_checkpointer: BaseCheckpointSaver, snapshot: SnapshotAssertion
) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
# setup subgraph
libs/langgraph/tests/test_large_cases_async.py
+16 -29
View File
@@ -9,8 +9,22 @@ from typing import (
)
import pytest
from langchain_core.messages import AnyMessage, ToolCall
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
ToolCall,
ToolMessage,
)
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig, RunnablePick
from langchain_core.tools import tool
from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.prebuilt.chat_agent_executor import create_react_agent
@@ -22,6 +36,7 @@ from langgraph._internal._constants import PULL, PUSH
from langgraph.channels.last_value import LastValue
from langgraph.channels.untracked_value import UntrackedValue
from langgraph.constants import END, START
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from langgraph.graph.state import StateGraph
from langgraph.pregel import NodeBuilder, Pregel
@@ -479,10 +494,6 @@ async def test_fork_always_re_runs_nodes(
async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool
class AgentState(TypedDict):
input: Annotated[str, UntrackedValue]
@@ -1017,8 +1028,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
async def test_prebuilt_tool_chat() -> None:
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
model = FakeChatModel(
messages=[
@@ -1358,16 +1367,6 @@ async def test_prebuilt_tool_chat() -> None:
async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
ToolMessage,
)
from langchain_core.tools import tool
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
@@ -2072,11 +2071,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
async def test_message_graph(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
class FakeFunctionChatModel(FakeMessagesListChatModel):
def bind_functions(self, functions: list):
@@ -3537,13 +3531,6 @@ async def test_send_to_nested_graphs(async_checkpointer: BaseCheckpointSaver) ->
async def test_weather_subgraph(
async_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, ToolCall
from langchain_core.tools import tool
from langgraph.graph import MessagesState
# setup subgraph
libs/langgraph/tests/test_pregel.py
+18 -27
View File
@@ -4,25 +4,39 @@ import gc
import json
import logging
import operator
import random
import threading
import time
import uuid
from collections import Counter, deque
from collections import Counter, defaultdict, deque
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from random import randrange
from typing import Annotated, Any, Literal, get_type_hints
from unittest.mock import patch
import pytest
from langchain_core.language_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
RemoveMessage,
)
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.runnables import (
RunnableConfig,
RunnableLambda,
RunnablePassthrough,
)
from langchain_core.runnables.graph import Edge
from langchain_core.tools import tool
from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import (
@@ -56,8 +70,9 @@ from langgraph.pregel import (
NodeBuilder,
Pregel,
)
from langgraph.pregel._loop import SyncPregelLoop
from langgraph.pregel._loop import PregelLoop, SyncPregelLoop
from langgraph.pregel._runner import PregelRunner
from langgraph.runtime import RunControl
from langgraph.types import (
CachePolicy,
Command,
@@ -125,7 +140,6 @@ def test_graph_validation() -> None:
def test_request_drain_allows_inflight_call_scheduling(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from langgraph.runtime import RunControl
@task
def child(x: int) -> int:
@@ -1769,9 +1783,6 @@ def test_conditional_state_graph_with_list_edge_inputs(snapshot: SnapshotAsserti
def test_state_graph_w_config_inherited_state_keys(snapshot: SnapshotAssertion) -> None:
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool
class BaseState(TypedDict):
input: str
@@ -3769,12 +3780,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
previous checkpoint config for each step in the run.
"""
# set up test
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, AnyMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
# graph state
class BaseState(TypedDict):
@@ -3940,7 +3945,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
def test_remove_message_via_state_update(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
workflow = StateGraph(state_schema=Annotated[list[AnyMessage], add_messages]) # type: ignore[arg-type]
workflow.add_node(
@@ -3973,7 +3977,6 @@ def test_remove_message_via_state_update(
def test_remove_message_from_node():
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
workflow = StateGraph(state_schema=Annotated[list[AnyMessage], add_messages]) # type: ignore[arg-type]
workflow.add_node(
@@ -3999,7 +4002,6 @@ def test_remove_message_from_node():
def test_xray_lance(snapshot: SnapshotAssertion):
from langchain_core.messages import AnyMessage, HumanMessage
class Analyst(BaseModel):
affiliation: str = Field(
@@ -4483,7 +4485,6 @@ def test_debug_subgraphs(
def test_debug_nested_subgraphs(
sync_checkpointer: BaseCheckpointSaver, durability: Durability
):
from collections import defaultdict
class State(TypedDict):
messages: Annotated[list[str], operator.add]
@@ -4743,8 +4744,6 @@ def test_runnable_passthrough_node_graph() -> None:
def test_parent_command(
sync_checkpointer: BaseCheckpointSaver, subgraph_persist: bool
) -> None:
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
@tool(return_direct=True)
def get_user_name() -> Command:
@@ -5164,7 +5163,6 @@ def test_command_with_static_breakpoints(
def test_multistep_plan(sync_checkpointer: BaseCheckpointSaver):
from langchain_core.messages import AnyMessage
class State(TypedDict, total=False):
plan: list[str | list[str]]
@@ -5910,9 +5908,6 @@ def test_no_redundant_put_writes_for_cached_task(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
"""Cached @tasks on resume must not trigger redundant put_writes."""
from unittest.mock import patch
from langgraph.pregel._loop import PregelLoop
@task
def setup(x: int) -> int:
@@ -6975,7 +6970,6 @@ def test_configurable_propagates_to_stream_metadata() -> None:
def test_stream_mode_messages_command() -> None:
from langchain_core.messages import HumanMessage
def my_node(state):
return {"messages": HumanMessage(content="foo")}
@@ -7243,7 +7237,6 @@ def test_get_stream_writer() -> None:
def test_stream_messages_dedupe_inputs() -> None:
from langchain_core.messages import AIMessage
def call_model(state):
return {"messages": AIMessage("hi", id="1")}
@@ -7281,7 +7274,6 @@ def test_stream_messages_dedupe_inputs() -> None:
def test_stream_messages_dedupe_state(sync_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.messages import AIMessage
to_emit = [AIMessage("bye", id="1"), AIMessage("bye again", id="2")]
@@ -8253,7 +8245,6 @@ def test_get_graph_loop(snapshot: SnapshotAssertion) -> None:
def test_get_graph_self_loop(snapshot: SnapshotAssertion) -> None:
import random
subgraph_builder = StateGraph(MessagesState)
subgraph_builder.add_node("agent", lambda x: x)
libs/langgraph/tests/test_pregel_async.py
+18 -29
View File
@@ -7,7 +7,7 @@ import operator
import random
import sys
import uuid
from collections import Counter, deque
from collections import Counter, defaultdict, deque
from dataclasses import replace
from time import perf_counter
from typing import (
@@ -21,8 +21,20 @@ from uuid import UUID
import pytest
from langchain_core.language_models import GenericFakeChatModel
from langchain_core.messages import HumanMessage
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
ToolCall,
ToolMessage,
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnablePassthrough
from langchain_core.tools import tool
from langchain_core.utils.aiter import aclosing
from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
from langgraph.cache.base import BaseCache
@@ -45,6 +57,7 @@ from typing_extensions import NotRequired, TypedDict
from langgraph._internal._constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL
from langgraph._internal._queue import AsyncQueue
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.delta import DeltaChannel
from langgraph.channels.last_value import LastValue
from langgraph.channels.topic import Topic
from langgraph.errors import (
@@ -55,10 +68,11 @@ from langgraph.errors import (
)
from langgraph.func import entrypoint, task
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import MessagesState, add_messages
from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
from langgraph.pregel import NodeBuilder, Pregel
from langgraph.pregel._loop import AsyncPregelLoop
from langgraph.pregel._loop import AsyncPregelLoop, PregelLoop
from langgraph.pregel._runner import PregelRunner
from langgraph.runtime import RunControl
from langgraph.types import (
CachePolicy,
Command,
@@ -222,7 +236,6 @@ async def test_checkpoint_errors() -> None:
async def test_request_drain_allows_inflight_acall_scheduling(
async_checkpointer: BaseCheckpointSaver,
) -> None:
from langgraph.runtime import RunControl
@task
async def child(x: int) -> int:
@@ -2868,7 +2881,6 @@ async def test_send_dedupe_on_resume(
async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
ai_message = AIMessage(
"",
@@ -3259,7 +3271,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
async def test_send_react_interrupt_control(
async_checkpointer: BaseCheckpointSaver, snapshot: SnapshotAssertion
) -> None:
from langchain_core.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
ai_message = AIMessage(
"",
@@ -5538,12 +5549,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
previous checkpoint config for each step in the run.
"""
# set up test
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, AnyMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
# graph state
class BaseState(TypedDict):
@@ -5944,7 +5949,6 @@ async def test_debug_subgraphs(
async def test_debug_nested_subgraphs(
async_checkpointer: BaseCheckpointSaver, durability: Durability
) -> None:
from collections import defaultdict
class State(TypedDict):
messages: Annotated[list[str], operator.add]
@@ -6061,8 +6065,6 @@ async def test_debug_nested_subgraphs(
async def test_parent_command(
async_checkpointer: BaseCheckpointSaver, subgraph_persist: bool
) -> None:
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
@tool(return_direct=True)
def get_user_name() -> Command:
@@ -6130,10 +6132,6 @@ async def test_parent_command(
async def test_delta_channel_durability_exit_stores_snapshot_async() -> None:
"""DeltaChannel must reload from an async durability='exit' checkpoint."""
from langchain_core.messages import AIMessage
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import _messages_delta_reducer
class State(TypedDict):
messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
@@ -6420,7 +6418,6 @@ async def test_command_with_static_breakpoints(
async def test_multistep_plan(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.messages import AnyMessage
class State(TypedDict, total=False):
plan: list[str | list[str]]
@@ -6758,7 +6755,6 @@ async def test_multiple_interrupts_functional(
async_checkpointer: BaseCheckpointSaver,
) -> None:
"""Test multiple interrupts with functional API."""
from langgraph.func import entrypoint, task
counter = 0
@@ -7674,7 +7670,6 @@ async def test_configurable_propagates_to_stream_metadata() -> None:
async def test_stream_mode_messages_command() -> None:
from langchain_core.messages import HumanMessage
async def my_node(state):
return {"messages": HumanMessage(content="foo")}
@@ -7723,7 +7718,6 @@ async def test_stream_mode_messages_command() -> None:
async def test_stream_messages_dedupe_inputs() -> None:
from langchain_core.messages import AIMessage
async def call_model(state):
return {"messages": AIMessage("hi", id="1")}
@@ -7763,7 +7757,6 @@ async def test_stream_messages_dedupe_inputs() -> None:
async def test_stream_messages_dedupe_state(
async_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.messages import AIMessage
to_emit = [AIMessage("bye", id="1"), AIMessage("bye again", id="2")]
@@ -8142,9 +8135,6 @@ async def test_no_redundant_put_writes_for_cached_task(
async_checkpointer: BaseCheckpointSaver,
) -> None:
"""Cached @tasks on resume must not trigger redundant put_writes."""
from unittest.mock import patch
from langgraph.pregel._loop import PregelLoop
@task
async def setup(x: int) -> int:
@@ -8646,7 +8636,6 @@ async def test_batch_update_as_input(
async def test_draw_invalid():
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
libs/langgraph/tests/test_pydantic.py
+3 -4
View File
@@ -4,10 +4,13 @@ import ipaddress
import pathlib
import re
import sys
import typing
import uuid
from enum import Enum
from typing import Annotated, Literal, Optional
import pydantic
import typing_extensions
from langgraph.checkpoint.base import BaseCheckpointSaver
from pydantic import (
BaseModel,
@@ -32,10 +35,6 @@ from tests.any_str import AnyStr
def test_is_supported_by_pydantic() -> None:
"""Test if types are supported by pydantic."""
import typing
import pydantic
import typing_extensions
class TypedDictExtensions(typing_extensions.TypedDict):
x: int
libs/langgraph/tests/test_remote_graph.py
+3 -5
View File
@@ -10,12 +10,14 @@ from langchain_core.messages import AnyMessage, BaseMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.graph import Edge as DrawableEdge
from langchain_core.runnables.graph import Node as DrawableNode
from langgraph.checkpoint.memory import InMemorySaver
from langgraph_sdk.client import get_client, get_sync_client
from langgraph_sdk.schema import StreamPart
from pydantic import BaseModel
from typing_extensions import TypedDict
from langgraph.errors import GraphInterrupt
from langgraph.graph import StateGraph, add_messages
from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
from langgraph.pregel import Pregel
from langgraph.pregel.remote import RemoteGraph
from langgraph.types import Interrupt, StateSnapshot
@@ -1097,10 +1099,6 @@ def test_stream_context_base_model():
)
@pytest.mark.anyio
async def test_langgraph_cloud_integration():
from langgraph.checkpoint.memory import InMemorySaver
from langgraph_sdk.client import get_client, get_sync_client
from langgraph.graph import END, START, MessagesState, StateGraph
# create RemotePregel instance
client = get_client(url="http://localhost:8123")
libs/langgraph/tests/test_remote_graph_v3.py
+1 -1
View File
@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from langgraph.pregel import remote as remote_mod
from langgraph.pregel._remote_run_stream import (
_AsyncRemoteGraphRunStream,
_ChannelProjection,
@@ -577,7 +578,6 @@ def test_stream_events_v3_strips_checkpoint_keys_from_configurable():
def test_stream_events_v3_merges_tracing_headers_when_distributed_tracing(
monkeypatch,
):
from langgraph.pregel import remote as remote_mod
sync_client = MagicMock()
sync_client.threads.stream.return_value = MagicMock()
libs/langgraph/tests/test_retry.py
+3 -5
View File
@@ -11,7 +11,9 @@ from typing import Annotated, Any
from unittest.mock import Mock, patch
from uuid import uuid4
import httpx
import pytest
import requests
from langchain_core.callbacks import AsyncCallbackManagerForLLMRun, BaseCallbackHandler
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage, HumanMessage
@@ -63,6 +65,7 @@ from langgraph.types import (
RetryPolicy,
Send,
TimeoutPolicy,
interrupt,
)
NEEDS_CONTEXTVARS = pytest.mark.skipif(
@@ -171,8 +174,6 @@ def test_checkpoint_ns_for_parent_command() -> None:
def test_should_retry_default_retry_on():
"""Test the default retry_on function."""
import httpx
import requests
# Create a RetryPolicy with default_retry_on
policy = RetryPolicy()
@@ -2198,7 +2199,6 @@ def test_graph_error_handler_does_not_swallow_interrupt_concurrent():
"""When a graph error handler is configured and a node calls interrupt()
concurrently with other nodes, the interrupt must still be raised — not
silently swallowed."""
from langgraph.types import interrupt
class State(TypedDict):
foo: str
@@ -2587,8 +2587,6 @@ async def test_set_node_defaults_timeout():
.compile()
)
from langgraph.errors import NodeTimeoutError
with pytest.raises(NodeTimeoutError):
await graph.ainvoke({"foo": ""})
libs/langgraph/tests/test_runtime.py
+2 -3
View File
@@ -6,9 +6,11 @@ from typing import Any
import pytest
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel, ValidationError
from typing_extensions import TypedDict
from langgraph._internal._constants import CONFIG_KEY_RUNTIME
from langgraph.errors import GraphDrained
from langgraph.graph import END, START, StateGraph
from langgraph.runtime import (
@@ -1177,9 +1179,6 @@ def test_foreign_object_in_runtime_slot_is_coerced() -> None:
`merge` when no per-run `context` is provided. `store` is resolved
separately, so it is not read off the foreign object in the coercion.
"""
from langgraph.store.memory import InMemoryStore
from langgraph._internal._constants import CONFIG_KEY_RUNTIME
store = InMemoryStore()
graph_level_context = {"source": "graph-level"}
libs/langgraph/tests/test_serde_allowlist.py
+1 -1
View File
@@ -79,7 +79,7 @@ class DummyChannel:
def test_curated_core_allowlist_includes_messages() -> None:
try:
from langchain_core.messages import BaseMessage
from langchain_core.messages import BaseMessage # noqa: PLC0415
except Exception:
pytest.skip("langchain_core not available")
allowlist = curated_core_allowlist()
libs/langgraph/tests/test_stream_data_transformers.py
+2 -3
View File
@@ -13,8 +13,10 @@ import operator
import time
from typing import Annotated, Any
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
from langgraph.config import get_stream_writer
from langgraph.constants import END, START
from langgraph.graph import StateGraph
from langgraph.stream._mux import StreamMux
@@ -488,7 +490,6 @@ class _State(TypedDict):
def _my_node(state: _State) -> dict[str, Any]:
from langgraph.config import get_stream_writer
writer = get_stream_writer()
writer({"status": "working", "node": "my_node"})
@@ -606,7 +607,6 @@ def test_stream_events_v3_all_transformers_interleaved() -> None:
def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
"""All transformers with a checkpointer — run.checkpoints populated."""
from langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(_State, input_schema=_State)
builder.add_node("my_node", _my_node)
@@ -645,7 +645,6 @@ def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
def test_stream_events_v3_checkpoints_projection_opt_in() -> None:
"""run.checkpoints surfaces checkpoint data when opted in with a checkpointer."""
from langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(_State, input_schema=_State)
builder.add_node("my_node", _my_node)
libs/langgraph/tests/test_stream_messages_transformer.py
+4 -14
View File
@@ -3,8 +3,10 @@ legacy v1 chunk filtering, and end-to-end via stream_events(version="v3") / astr
from __future__ import annotations
import asyncio
import time
from typing import Any
from uuid import uuid4
import pytest
from langchain_core.language_models import GenericFakeChatModel
@@ -13,11 +15,13 @@ from langchain_core.language_models.chat_model_stream import (
ChatModelStream,
)
from langchain_core.messages import AIMessage, AIMessageChunk, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
from langchain_core.runnables import RunnableConfig
from typing_extensions import TypedDict
from langgraph.constants import END, START
from langgraph.graph import MessagesState, StateGraph
from langgraph.pregel._messages import StreamMessagesHandlerV2
from langgraph.stream._mux import StreamMux
from langgraph.stream.run_stream import GraphRunStream
from langgraph.stream.stream_channel import StreamChannel
@@ -607,7 +611,6 @@ class TestEndToEnd:
@pytest.mark.anyio
async def test_nested_async_iteration_yields_text_deltas(self) -> None:
"""Inner stream.text drives the shared graph pump via the async pump binding."""
import asyncio
model = GenericFakeChatModel(messages=iter(["hello world"]))
@@ -870,11 +873,6 @@ class TestDirectMessagesModeStaysV1:
class TestStreamMessagesHandlerV2Unit:
def test_on_llm_new_token_is_noop(self) -> None:
"""v2 handler must not emit v1 chunks even when on_llm_new_token fires."""
from uuid import uuid4
from langchain_core.outputs import ChatGenerationChunk
from langgraph.pregel._messages import StreamMessagesHandlerV2
emitted: list[Any] = []
handler = StreamMessagesHandlerV2(emitted.append, subgraphs=False)
@@ -890,9 +888,6 @@ class TestStreamMessagesHandlerV2Unit:
assert emitted == []
def test_on_chain_end_does_not_emit_tool_messages(self) -> None:
from uuid import uuid4
from langgraph.pregel._messages import StreamMessagesHandlerV2
emitted: list[Any] = []
handler = StreamMessagesHandlerV2(emitted.append, subgraphs=False)
@@ -909,11 +904,6 @@ class TestStreamMessagesHandlerV2Unit:
def test_on_llm_end_dedupes_when_final_message_id_differs(self) -> None:
"""A streamed v2 message should not be emitted again from the final
AIMessage fallback when its final id does not match `message-start`."""
from uuid import uuid4
from langchain_core.outputs import ChatGeneration, LLMResult
from langgraph.pregel._messages import StreamMessagesHandlerV2
emitted: list[Any] = []
handler = StreamMessagesHandlerV2(emitted.append, subgraphs=False)
libs/langgraph/tests/test_utils.py
+4 -10
View File
@@ -2,6 +2,7 @@ import functools
import sys
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from typing import (
Annotated,
Any,
@@ -17,7 +18,10 @@ import langsmith
import pytest
from langchain_core.callbacks import BaseCallbackHandler, CallbackManager
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.config import var_child_runnable_config
from langchain_core.tracers import LangChainTracer
from langsmith import get_current_run_tree # type: ignore
from pydantic import BaseModel, Field
from typing_extensions import NotRequired, Required, TypedDict
from langgraph._internal._config import (
@@ -118,7 +122,6 @@ def rt_graph() -> CompiledStateGraph:
node_run_id: int
def node(_: State):
from langsmith import get_current_run_tree # type: ignore
return {"node_run_id": get_current_run_tree().id} # type: ignore
@@ -243,10 +246,6 @@ def test_is_required():
def test_enhanced_type_hints() -> None:
from dataclasses import dataclass
from typing import Annotated
from pydantic import BaseModel, Field
class MyTypedDict(TypedDict):
val_1: str
@@ -510,7 +509,6 @@ def test_ensure_config_explicit_configurable_replaces_ambient() -> None:
# An explicit checkpoint coordinate (here a new thread_id) starts a fresh
# lineage and drops the ambient run context (e.g. a parent task's
# checkpoint_ns), so a child graph does not inherit it.
from langchain_core.runnables.config import var_child_runnable_config
token = var_child_runnable_config.set(
{"configurable": {"checkpoint_ns": "p:parent-task", "checkpoint_id": "cid"}}
@@ -527,7 +525,6 @@ def test_ensure_config_explicit_configurable_replaces_ambient() -> None:
def test_ensure_config_ambient_inherited_when_no_explicit_configurable() -> None:
# With no explicit configurable, the ambient run context is inherited
# unchanged (stateless subgraph / interrupt-resume pattern).
from langchain_core.runnables.config import var_child_runnable_config
token = var_child_runnable_config.set(
{"configurable": {"checkpoint_ns": "p:parent-task"}}
@@ -543,7 +540,6 @@ def test_ensure_config_explicit_configurables_still_merge_over_ambient() -> None
# A new thread_id drops the ambient, but explicit configs still shallow-merge
# among themselves, so a with_config(...) value (ls_agent_type) survives
# alongside an invoke-time thread_id.
from langchain_core.runnables.config import var_child_runnable_config
token = var_child_runnable_config.set(
{"configurable": {"checkpoint_ns": "p:parent-task"}}
@@ -564,7 +560,6 @@ def test_ensure_config_non_coordinate_config_keeps_ambient_checkpoint_ns() -> No
# A nested subagent is invoked with a non-coordinate configurable key
# (ls_agent_type) and no thread_id; it must keep the inherited checkpoint_ns
# so it stays a discoverable child of the parent run (deepagents `task` tool).
from langchain_core.runnables.config import var_child_runnable_config
token = var_child_runnable_config.set(
{"configurable": {"thread_id": "parent", "checkpoint_ns": "p:parent-task"}}
@@ -582,7 +577,6 @@ def test_ensure_config_same_thread_id_still_clears_ambient() -> None:
# A child that reuses the parent's thread_id is still addressing its own root
# namespace on that thread, so the parent task's checkpoint_ns must not leak
# in; otherwise the child writes state that get_state cannot read back.
from langchain_core.runnables.config import var_child_runnable_config
token = var_child_runnable_config.set(
{"configurable": {"thread_id": "shared", "checkpoint_ns": "p:parent-task"}}
libs/prebuilt/pyproject.toml
+5 -1
View File
@@ -75,8 +75,12 @@ addopts = "--strict-markers --strict-config --durations=5 -vv"
asyncio_mode = "auto"
[tool.ruff]
lint.select = [ "E", "F", "I", "RUF100", "TID251", "UP" ]
lint.select = [ "E", "F", "I", "PLC0415", "RUF100", "TID251", "UP" ]
lint.ignore = [ "E501" ]
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
lint.per-file-ignores = { "langgraph/**" = ["PLC0415"] }
target-version = "py310"
[tool.ty.rules]
libs/prebuilt/tests/memory_assert.py
+1 -2
View File
@@ -1,5 +1,6 @@
import os
import tempfile
import time
from collections import defaultdict
from functools import partial
@@ -38,8 +39,6 @@ class MemorySaverAssertImmutable(InMemorySaver):
new_versions: ChannelVersions,
) -> None:
if self.put_sleep:
import time
time.sleep(self.put_sleep)
# assert checkpoint hasn't been modified since last written
thread_id = config["configurable"]["thread_id"]
libs/prebuilt/tests/test_injected_state_not_required.py
+3 -4
View File
@@ -9,16 +9,19 @@ handle missing fields by injecting None instead of raising KeyError.
import sys
from typing import Annotated
from unittest.mock import Mock
import pytest
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph.message import add_messages
from langgraph.runtime import Runtime
from pydantic import BaseModel, Field
from typing_extensions import NotRequired
from langgraph.prebuilt import InjectedState, ToolNode, create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState
from langgraph.prebuilt.tool_node import ToolRuntime
from .model import FakeToolCallingModel
@@ -50,9 +53,6 @@ def _create_mock_runtime(
store=None,
):
"""Create a mock Runtime for testing ToolNode directly."""
from unittest.mock import Mock
from langgraph.runtime import Runtime
mock_runtime = Mock(spec=Runtime)
mock_runtime.context = {}
@@ -61,7 +61,6 @@ def _create_mock_runtime(
def _create_config_with_runtime(store=None, state=None):
"""Create a RunnableConfig with mocked runtime for direct ToolNode testing."""
from langgraph.prebuilt.tool_node import ToolRuntime
tool_runtime = ToolRuntime(
state=state or {},
libs/prebuilt/tests/test_on_tool_call.py
+1 -1
View File
@@ -1,5 +1,6 @@
"""Unit tests for tool call interceptor in ToolNode."""
import functools
from collections.abc import Callable
from unittest.mock import Mock
@@ -1331,7 +1332,6 @@ def _config_with_channel_read(
learn channel names. The stub matches the shape: partial whose second and
third positional args are `channels` and `managed` mappings.
"""
import functools
channels_stub = {k: None for k in channel_values}
managed_stub: dict[str, object] = {}
libs/prebuilt/tests/test_tool_node.py
+4 -10
View File
@@ -2,6 +2,7 @@ import contextlib
import dataclasses
import json
import sys
import warnings
from functools import partial
from typing import (
Annotated,
@@ -23,10 +24,12 @@ from langchain_core.messages import (
from langchain_core.runnables.config import RunnableConfig
from langchain_core.tools import BaseTool, InjectedToolArg, ToolException
from langchain_core.tools import tool as dec_tool
from langchain_core.tools.base import InjectedToolCallId
from langgraph.config import get_stream_writer
from langgraph.errors import GraphBubbleUp, GraphInterrupt
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages
from langgraph.runtime import ExecutionInfo, ServerInfo
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.types import Command, Send
@@ -41,6 +44,7 @@ from langgraph.prebuilt import (
)
from langgraph.prebuilt.tool_node import (
TOOL_CALL_ERROR_TEMPLATE,
ToolCallRequest,
ToolInvocationError,
ToolRuntime,
tools_condition,
@@ -59,7 +63,6 @@ def _create_mock_runtime(store: BaseStore | None = None) -> Mock:
which is injected by RunnableCallable from config["configurable"]["__pregel_runtime"].
When testing ToolNode directly (outside a graph), we need to provide this manually.
"""
from langgraph.runtime import ExecutionInfo
mock_runtime = Mock()
mock_runtime.store = store
@@ -625,7 +628,6 @@ def test_tool_node_node_interrupt() -> None:
@pytest.mark.parametrize("input_type", ["dict", "tool_calls"])
async def test_tool_node_command(input_type: str) -> None:
from langchain_core.tools.base import InjectedToolCallId
@dec_tool
def transfer_to_bob(tool_call_id: Annotated[str, InjectedToolCallId]):
@@ -934,7 +936,6 @@ async def test_tool_node_command(input_type: str) -> None:
async def test_tool_node_command_list_input() -> None:
from langchain_core.tools.base import InjectedToolCallId
@dec_tool
def transfer_to_bob(tool_call_id: Annotated[str, InjectedToolCallId]):
@@ -1194,7 +1195,6 @@ async def test_tool_node_command_list_input() -> None:
def test_tool_node_parent_command_with_send() -> None:
from langchain_core.tools.base import InjectedToolCallId
@dec_tool
def transfer_to_alice(tool_call_id: Annotated[str, InjectedToolCallId]):
@@ -1282,7 +1282,6 @@ def test_tool_node_parent_command_with_send() -> None:
async def test_tool_node_command_remove_all_messages() -> None:
from langchain_core.tools.base import InjectedToolCallId
@dec_tool
def remove_all_messages_tool(tool_call_id: Annotated[str, InjectedToolCallId]):
@@ -1621,9 +1620,6 @@ def test_tool_node_stream_writer() -> None:
def test_tool_call_request_setattr_deprecation_warning():
"""Test that ToolCallRequest raises a deprecation warning on direct attribute modification."""
import warnings
from langgraph.prebuilt.tool_node import ToolCallRequest
# Create a mock ToolCall
tool_call = {"name": "test", "args": {"a": 1}, "id": "call_1", "type": "tool_call"}
@@ -2031,7 +2027,6 @@ def test_tool_runtime_defaults_tools_to_empty_list() -> None:
def test_tool_runtime_forwards_execution_info_server_info_and_tools() -> None:
"""Test that execution_info, server_info, and tools are forwarded from Runtime to ToolRuntime."""
from langgraph.runtime import ExecutionInfo, ServerInfo
exec_info = ExecutionInfo(
thread_id="t-1",
@@ -2088,7 +2083,6 @@ async def test_tool_runtime_forwards_execution_info_server_info_and_tools_async(
None
):
"""Test that execution_info, server_info, and tools are forwarded in async path."""
from langgraph.runtime import ExecutionInfo, ServerInfo
exec_info = ExecutionInfo(
thread_id="t-2",
libs/sdk-py/pyproject.toml
+5 -1
View File
@@ -80,6 +80,7 @@ select = [
"SIM", # flake8-simplify (code simplification)
"RUF", # ruff-specific rules
"S101", # flake8-bandit: use of assert
"PLC0415", # import-outside-top-level
]
ignore = [
"E501", # line too long (handled by formatter)
@@ -87,7 +88,10 @@ ignore = [
"B904", # raise without from inside except (sometimes intentional)
"SIM102", # nested if statements (sometimes clearer)
]
per-file-ignores = { "tests/**" = ["S101", "B017"], "integration/**" = ["S101", "T20", "B017", "ARG001", "ARG002"] }
# PLC0415 (import-outside-top-level) is enforced in tests only. Library code
# still has deferred imports that have not been reviewed, so it stays exempt
# for now.
per-file-ignores = { "tests/**" = ["S101", "B017"], "integration/**" = ["S101", "T20", "B017", "ARG001", "ARG002", "PLC0415"], "langgraph_sdk/**" = ["PLC0415"] }
[tool.ty.src]
# The `integration/` graphs run inside the docker image (with `deepagents`
libs/sdk-py/tests/integration/conftest.py
+5 -4
View File
@@ -18,6 +18,11 @@ from collections.abc import AsyncIterator, Iterator
import httpx
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.threads import SyncThreadsClient
BASE_URL = os.environ.get("LANGGRAPH_INTEGRATION_URL", "http://localhost:2024")
ASSISTANT_ID = "agent"
TOOLS_ASSISTANT_ID = "tools_agent"
@@ -47,8 +52,6 @@ def _require_running_api() -> None:
@pytest.fixture
async def async_threads() -> AsyncIterator[tuple[object, httpx.AsyncClient]]:
"""Build an async ThreadsClient. Yields `(threads, raw_httpx)` so tests can close raw."""
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
raw = httpx.AsyncClient(base_url=BASE_URL, timeout=30.0)
try:
@@ -60,8 +63,6 @@ async def async_threads() -> AsyncIterator[tuple[object, httpx.AsyncClient]]:
@pytest.fixture
def sync_threads() -> Iterator[tuple[object, httpx.Client]]:
"""Build a sync ThreadsClient. Yields `(threads, raw_httpx)` so tests can close raw."""
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.threads import SyncThreadsClient
raw = httpx.Client(base_url=BASE_URL, timeout=30.0)
try:
libs/sdk-py/tests/integration/test_assistants.py
+5 -4
View File
@@ -9,21 +9,22 @@ from __future__ import annotations
import pytest
from langgraph_sdk._async.assistants import AssistantsClient
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._sync.assistants import SyncAssistantsClient
from langgraph_sdk._sync.http import SyncHttpClient
from .conftest import ASSISTANT_ID
pytestmark = pytest.mark.integration
def _async_assistants(raw):
from langgraph_sdk._async.assistants import AssistantsClient
from langgraph_sdk._async.http import HttpClient
return AssistantsClient(HttpClient(raw))
def _sync_assistants(raw):
from langgraph_sdk._sync.assistants import SyncAssistantsClient
from langgraph_sdk._sync.http import SyncHttpClient
return SyncAssistantsClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_cancel.py
+5 -4
View File
@@ -10,6 +10,11 @@ from typing import Any
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
from .conftest import ASSISTANT_ID
pytestmark = pytest.mark.integration
@@ -29,8 +34,6 @@ async def _cancel_after_first_event(
async def test_cancel_async(async_threads) -> None:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
threads, raw = async_threads
runs_client = RunsClient(HttpClient(raw))
@@ -88,8 +91,6 @@ def _cancel_after_first_event_sync(
def test_cancel_sync(sync_threads) -> None:
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
threads, raw = sync_threads
runs_client = SyncRunsClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_crons.py
+5 -4
View File
@@ -10,21 +10,22 @@ from __future__ import annotations
import pytest
from langgraph_sdk._async.cron import CronClient
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._sync.cron import SyncCronClient
from langgraph_sdk._sync.http import SyncHttpClient
from .conftest import ASSISTANT_ID
pytestmark = pytest.mark.integration
def _async_crons(raw):
from langgraph_sdk._async.cron import CronClient
from langgraph_sdk._async.http import HttpClient
return CronClient(HttpClient(raw))
def _sync_crons(raw):
from langgraph_sdk._sync.cron import SyncCronClient
from langgraph_sdk._sync.http import SyncHttpClient
return SyncCronClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_factory_graph.py
+5 -4
View File
@@ -13,21 +13,22 @@ from __future__ import annotations
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
from .conftest import FACTORY_ASSISTANT_ID
pytestmark = pytest.mark.integration
def _async_runs(raw):
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
return RunsClient(HttpClient(raw))
def _sync_runs(raw):
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
return SyncRunsClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_runs.py
+5 -4
View File
@@ -11,21 +11,22 @@ from __future__ import annotations
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
from .conftest import ASSISTANT_ID
pytestmark = pytest.mark.integration
def _async_runs(raw):
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.runs import RunsClient
return RunsClient(HttpClient(raw))
def _sync_runs(raw):
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.runs import SyncRunsClient
return SyncRunsClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_store.py
+5 -4
View File
@@ -10,19 +10,20 @@ import uuid
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.store import StoreClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.store import SyncStoreClient
pytestmark = pytest.mark.integration
def _async_store(raw):
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.store import StoreClient
return StoreClient(HttpClient(raw))
def _sync_store(raw):
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.store import SyncStoreClient
return SyncStoreClient(SyncHttpClient(raw))
libs/sdk-py/tests/integration/test_websocket.py
+5 -4
View File
@@ -4,6 +4,11 @@ from __future__ import annotations
import pytest
from langgraph_sdk.stream.transport import (
ProtocolWebSocketTransport,
SyncProtocolWebSocketTransport,
)
from .conftest import ASSISTANT_ID, EXPECTED_TERMINAL_ITEMS
pytestmark = pytest.mark.integration
@@ -14,8 +19,6 @@ async def test_websocket_async(async_threads) -> None:
async with threads.stream(
assistant_id=ASSISTANT_ID, transport="websocket"
) as thread:
from langgraph_sdk.stream.transport import ProtocolWebSocketTransport
assert isinstance(thread._transport, ProtocolWebSocketTransport)
await thread.run.start(input={"messages": [], "value": "init", "items": []})
@@ -34,8 +37,6 @@ async def test_websocket_async(async_threads) -> None:
def test_websocket_sync(sync_threads) -> None:
threads, _ = sync_threads
with threads.stream(assistant_id=ASSISTANT_ID, transport="websocket") as thread:
from langgraph_sdk.stream.transport import SyncProtocolWebSocketTransport
assert isinstance(thread._transport, SyncProtocolWebSocketTransport)
thread.run.start(input={"messages": [], "value": "init", "items": []})
libs/sdk-py/tests/streaming/test_controller.py
+9 -15
View File
@@ -3,14 +3,21 @@
from __future__ import annotations
import asyncio
import asyncio as _asyncio
import logging
from collections.abc import AsyncIterator
from typing import Any
from unittest.mock import AsyncMock
import httpx
import pytest
from langgraph_sdk.stream.controller import StreamController, _SeenEventIds
from langgraph_sdk.stream.transport.http import EventStreamHandle
from langgraph_sdk.stream.controller import (
StreamController,
_close_after,
_SeenEventIds,
)
from langgraph_sdk.stream.transport.http import EventStreamHandle, ProtocolSseTransport
# ---------------------------------------------------------------------------
# Task 3.1: bounded subscription queues
@@ -20,9 +27,6 @@ from langgraph_sdk.stream.transport.http import EventStreamHandle
@pytest.mark.asyncio
async def test_subscription_queue_bounded_by_max_queue_size():
"""`StreamController` must create per-subscription queues bounded by `max_queue_size`."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
transport = ProtocolSseTransport(
client=httpx.AsyncClient(base_url="http://test"),
@@ -36,9 +40,6 @@ async def test_subscription_queue_bounded_by_max_queue_size():
@pytest.mark.asyncio
async def test_subscription_queue_default_max_queue_size_is_1024():
"""`StreamController` default `max_queue_size` is 1024."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
transport = ProtocolSseTransport(
client=httpx.AsyncClient(base_url="http://test"),
@@ -114,12 +115,6 @@ def test_seen_event_ids_iter_returns_keys():
async def test_close_awaits_pending_rotation_closes():
"""When a rotation is mid-flight, controller.close() must await the old
stream close before returning."""
import asyncio as _asyncio
import httpx
from langgraph_sdk.stream.controller import _close_after
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
rotation_close_done = _asyncio.Event()
@@ -269,7 +264,6 @@ async def test_reconnect_accepts_backoff_kwargs():
@pytest.mark.anyio
async def test_transport_drop_exception_logged_with_type(monkeypatch, caplog):
"""Bare `pass` discarded exception types; the drop should at least log."""
import logging
monkeypatch.setattr("asyncio.sleep", AsyncMock())
libs/sdk-py/tests/streaming/test_decoders.py
+2 -1
View File
@@ -8,6 +8,8 @@ from __future__ import annotations
from typing import Any
import pytest
from langgraph_sdk.stream.decoders import (
DataDecoder,
ExtensionsDecoder,
@@ -454,7 +456,6 @@ def test_extensions_decoder_ignores_non_dict_data():
def test_extensions_decoder_rejects_empty_name():
import pytest
with pytest.raises(ValueError):
ExtensionsDecoder(name="")
libs/sdk-py/tests/streaming/test_extensions_projection.py
+1 -2
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import httpx
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.stream import ScopedStreamHandle
from langgraph_sdk._async.threads import ThreadsClient
from streaming._events import custom_event, lifecycle_completed_event
from streaming._fake_server import FakeServer
@@ -46,8 +47,6 @@ async def test_extension_projection_supports_namespace_scope_on_subgraph_handle(
)
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.stream import ScopedStreamHandle
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
await thread.run.start(input={})
libs/sdk-py/tests/streaming/test_lifecycle_watcher.py
+2 -2
View File
@@ -7,9 +7,11 @@ import contextlib
from typing import Any
import httpx
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
from langgraph_sdk.stream.transport import EventStreamHandle, ProtocolSseTransport
from streaming._events import (
input_requested_event,
lifecycle_completed_event,
@@ -153,7 +155,6 @@ async def test_lifecycle_clean_eof_resolves_run_done_with_errored():
"""If the lifecycle SSE stream ends cleanly (server closes without a
terminal `completed` or `errored` event), `_run_done` must resolve with
an errored terminal so awaiters don't hang."""
import pytest
fake = FakeServer()
# Emit a non-terminal lifecycle event, then close cleanly without
@@ -179,7 +180,6 @@ async def test_lifecycle_mid_iteration_error_resolves_run_done_with_error(
"""If the transport reports an error via `handle.done` after iteration
exits without a terminal lifecycle event, `_run_done` propagates the
transport error rather than the generic clean-EOF message."""
from langgraph_sdk.stream.transport import EventStreamHandle, ProtocolSseTransport
def synthetic_handle() -> EventStreamHandle:
loop = asyncio.get_running_loop()
libs/sdk-py/tests/streaming/test_scoped_handles.py
+4 -10
View File
@@ -2,12 +2,16 @@
from __future__ import annotations
from unittest.mock import MagicMock
import httpx
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.stream import ScopedStreamHandle
from langgraph_sdk._async.threads import ThreadsClient
from streaming._events import (
lifecycle_completed_event,
lifecycle_errored_event,
lifecycle_started_event,
message_finish_event,
message_start_event,
@@ -430,9 +434,6 @@ async def test_grandchild_events_dispatched_to_correct_sibling_not_first_match()
def test_scoped_handle_inboxes_bounded_by_max_queue_size():
"""ScopedStreamHandle with max_queue_size=N creates queues with maxsize=N."""
from unittest.mock import MagicMock
from langgraph_sdk._async.stream import ScopedStreamHandle
fake_thread = MagicMock()
handle = ScopedStreamHandle(
@@ -484,7 +485,6 @@ async def test_force_complete_uses_failed_when_run_errored():
"""If the lifecycle signals an errored run, scoped children that are still
'started' when the subgraphs projection's finally block runs must be
force-finished as 'failed', not 'completed'."""
from streaming._events import lifecycle_errored_event
fake = FakeServer()
fake.script(
@@ -538,9 +538,6 @@ async def test_force_complete_uses_completed_when_run_completed():
def test_close_inboxes_does_not_enqueue_on_uniterated_inboxes():
"""_close_inboxes must not push a sentinel on inboxes that had no consumer."""
from unittest.mock import MagicMock
from langgraph_sdk._async.stream import ScopedStreamHandle
fake_thread = MagicMock()
handle = ScopedStreamHandle(
@@ -559,9 +556,6 @@ def test_close_inboxes_does_not_enqueue_on_uniterated_inboxes():
def test_close_inboxes_enqueues_sentinel_on_iterated_inboxes():
"""_close_inboxes must push a None sentinel only on inboxes that had a consumer,
so projection iterators see the EOF signal."""
from unittest.mock import MagicMock
from langgraph_sdk._async.stream import ScopedStreamHandle
fake_thread = MagicMock()
handle = ScopedStreamHandle(
libs/sdk-py/tests/streaming/test_shared_stream.py
+3 -11
View File
@@ -3,14 +3,15 @@ from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator
from typing import Any, cast
from unittest.mock import MagicMock
import httpx
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
from langgraph_sdk.stream.controller import StreamController
from langgraph_sdk.stream.transport.http import EventStreamHandle
from streaming._events import lifecycle_event, values_event
from langgraph_sdk.stream.transport.http import EventStreamHandle, ProtocolSseTransport
from streaming._events import lifecycle_completed_event, lifecycle_event, values_event
from streaming._fake_server import FakeServer, _StreamScript
@@ -165,7 +166,6 @@ async def test_values_projection_registers_via_delegation_not_controller_directl
directly — the subscription count seen through the thread wrapper equals
the count inside the controller at the moment the subscription is live.
"""
from streaming._events import lifecycle_completed_event
fake = FakeServer()
fake.script([lifecycle_completed_event(seq=0)])
@@ -255,10 +255,6 @@ async def test_shared_stream_reconnects_with_since_after_transport_drop():
handle2, _ = _make_handle([values_event(seq=2, values={"counter": 2})])
handles = [handle1, handle2]
from unittest.mock import MagicMock
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
transport = MagicMock(spec=ProtocolSseTransport)
def _open(params: dict[str, Any]) -> EventStreamHandle:
@@ -303,10 +299,6 @@ async def test_shared_stream_reconnect_dedupes_replayed_overlap():
)
handles = [handle1, handle2]
from unittest.mock import MagicMock
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
transport = MagicMock(spec=ProtocolSseTransport)
transport.open_event_stream.side_effect = lambda _params: handles.pop(0)
libs/sdk-py/tests/streaming/test_sync_extensions_projection.py
+1 -2
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import httpx
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.stream import SyncScopedStreamHandle
from langgraph_sdk._sync.threads import SyncThreadsClient
from streaming._events import custom_event, lifecycle_completed_event
from streaming._sync_fake_server import SyncFakeServer
@@ -41,8 +42,6 @@ def test_sync_extension_projection_supports_namespace_scope_on_subgraph_handle()
]
)
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
from langgraph_sdk._sync.stream import SyncScopedStreamHandle
threads = SyncThreadsClient(SyncHttpClient(raw))
with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
thread.run.start(input={})
libs/sdk-py/tests/streaming/test_sync_projections.py
+3 -7
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
import time
from collections.abc import Generator
from typing import Any, cast
import httpx
@@ -10,6 +12,7 @@ from langchain_core.language_models.chat_model_stream import ChatModelStream
from langchain_protocol import Event
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.stream import SyncToolCallHandle
from langgraph_sdk._sync.threads import SyncThreadsClient
from streaming._events import (
lifecycle_completed_event,
@@ -459,11 +462,6 @@ def test_sync_tool_calls_explicit_close_does_not_block_1s():
tool_started_event(seq=1, tool_call_id="call-1"),
]
)
import time
from collections.abc import Generator
from typing import cast
from langgraph_sdk._sync.stream import SyncToolCallHandle
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
threads = SyncThreadsClient(SyncHttpClient(raw))
@@ -528,7 +526,6 @@ def test_sync_tool_call_handle_deltas_queue_is_bounded():
Unbounded queues allow producers to enqueue indefinitely, causing memory
growth when consumers are slow.
"""
from langgraph_sdk._sync.stream import SyncToolCallHandle
handle_default = SyncToolCallHandle(tool_call_id="tc1", name="foo")
assert handle_default._deltas.maxsize > 0, (
@@ -552,7 +549,6 @@ def test_sync_tool_call_handle_deltas_single_consumer_guard():
The property must raise before returning the iterator so the caller
sees the error even without iterating.
"""
from langgraph_sdk._sync.stream import SyncToolCallHandle
handle = SyncToolCallHandle(tool_call_id="tc1", name="foo")
libs/sdk-py/tests/streaming/test_sync_scoped_handles.py
+2 -2
View File
@@ -6,12 +6,14 @@ import threading
from concurrent.futures import ThreadPoolExecutor, wait
import httpx
from langchain_protocol import Event
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.stream import SyncScopedStreamHandle
from langgraph_sdk._sync.threads import SyncThreadsClient
from streaming._events import (
lifecycle_completed_event,
lifecycle_errored_event,
lifecycle_started_event,
message_finish_event,
message_start_event,
@@ -356,7 +358,6 @@ def test_sync_register_descendant_forwards_buffered_events_in_order():
"""_register_descendant must drain already-buffered events whose namespace
matches the new grandchild, push them into the grandchild, and preserve
the original arrival order in the parent inbox."""
from langchain_protocol import Event
parent = SyncScopedStreamHandle(
thread=None, # ty: ignore[invalid-argument-type]
@@ -673,7 +674,6 @@ def test_sync_force_complete_uses_failed_when_run_errored():
"""If the lifecycle signals an errored run, scoped children that are still
'started' when the subgraphs iterator's finally block runs must be
force-finished as 'failed', not 'completed'."""
from streaming._events import lifecycle_errored_event
fake = SyncFakeServer()
fake.script(
libs/sdk-py/tests/streaming/test_sync_thread_stream.py
+22 -87
View File
@@ -2,21 +2,43 @@
from __future__ import annotations
import queue
import re
import threading
import time
import uuid
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from typing import Any
import httpx
import orjson
import pytest
import langgraph_sdk.stream.sync_controller as _ctrl_mod
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.threads import SyncThreadsClient
from langgraph_sdk.stream.sync_controller import SyncStreamController
from langgraph_sdk.stream.transport.sync_http import (
SyncEventStreamHandle,
SyncProtocolSseTransport,
)
from streaming._events import (
checkpoints_event,
custom_event,
lifecycle_completed_event,
lifecycle_event,
lifecycle_started_event,
message_finish_event,
message_start_event,
message_text_delta_event,
message_text_finish_event,
tasks_start_event,
tool_finished_event,
tool_started_event,
updates_event,
values_event,
)
from streaming._sync_fake_server import SyncFakeServer, SyncStreamScript
# ---------------------------------------------------------------------------
@@ -71,14 +93,10 @@ def test_sync_subscribe_before_run_start_waits_on_gate():
def test_sync_reconnect_uses_backoff_between_attempts(monkeypatch):
"""_reconnect_shared_stream sleeps between retry attempts with exp+jitter
backoff, mirroring the async reconnect behavior."""
import langgraph_sdk.stream.sync_controller as _ctrl_mod
sleeps: list[float] = []
monkeypatch.setattr(_ctrl_mod.time, "sleep", lambda d: sleeps.append(d))
from langgraph_sdk.stream.sync_controller import SyncStreamController
from langgraph_sdk.stream.transport.sync_http import SyncProtocolSseTransport
class _FailingTransport(SyncProtocolSseTransport):
"""Transport that always raises on open_event_stream."""
@@ -113,15 +131,6 @@ def test_sync_rotation_does_not_lose_buffered_events():
"""When the shared stream rotates, old-stream events already in the queue
are not dropped. _drain_and_close dispatches remaining events from the
old handle to subscribers before closing it."""
import queue
from typing import Any
from langgraph_sdk.stream.sync_controller import SyncStreamController
from langgraph_sdk.stream.transport.sync_http import (
SyncEventStreamHandle,
SyncProtocolSseTransport,
)
from streaming._events import values_event
event_a = values_event(seq=1, counter=1)
@@ -188,8 +197,6 @@ def test_sync_rotation_does_not_lose_buffered_events():
def test_sync_concurrent_commands_do_not_share_command_id():
"""50 concurrent threads calling _send_command must each get a unique id."""
from concurrent.futures import ThreadPoolExecutor
from typing import Any
captured_ids: list[int] = []
ids_lock = threading.Lock()
@@ -246,7 +253,6 @@ def test_sync_events_returns_fresh_iterator_each_access():
"""Two accesses of `thread.events` yield independent subscriptions,
mirroring the async semantics where each access opens a new subscriber."""
fake = SyncFakeServer()
from streaming._events import values_event
event_1 = values_event(seq=1, counter=1)
fake.script_sequence(
@@ -280,7 +286,6 @@ def test_close_unblocks_active_subscription_before_lifecycle_join():
"""close() must send None to active subscriptions BEFORE joining the
lifecycle watcher thread, so callers wake quickly even if the watcher
thread blocks for up to 1s."""
import queue
# Gate that keeps the lifecycle watcher thread alive for 0.4s.
lifecycle_block = threading.Event()
@@ -293,8 +298,6 @@ def test_close_unblocks_active_subscription_before_lifecycle_join():
def _handle(self, request: httpx.Request) -> httpx.Response:
path = request.url.path
if path.endswith("/stream/events"):
import orjson
body = orjson.loads(request.content)
channels = body.get("channels", [])
if "lifecycle" in channels:
@@ -417,7 +420,6 @@ def test_sync_threads_stream_mints_uuid4_when_thread_id_none():
def test_sync_run_start_sends_command():
from streaming._events import lifecycle_completed_event
fake = SyncFakeServer()
fake.script([lifecycle_completed_event(seq=1)])
@@ -432,7 +434,6 @@ def test_sync_run_start_sends_command():
def test_sync_events_iterates_raw_events():
from streaming._events import values_event
fake = SyncFakeServer()
fake.script([values_event(seq=1, counter=1)])
@@ -446,7 +447,6 @@ def test_sync_events_iterates_raw_events():
def test_sync_lifecycle_watcher_reconnects_with_since_after_transport_drop():
from streaming._events import lifecycle_completed_event, lifecycle_event
fake = SyncFakeServer()
fake.set_state({"ok": True})
@@ -481,7 +481,6 @@ def test_sync_threads_stream_accepts_websocket_transport_option():
def test_sync_threads_stream_rejects_unknown_transport_option():
import pytest
with httpx.Client(base_url="http://test") as raw:
threads = SyncThreadsClient(SyncHttpClient(raw))
@@ -494,17 +493,6 @@ def test_sync_threads_stream_rejects_unknown_transport_option():
def test_v3_streaming_sync_surface_smoke():
from streaming._events import (
custom_event,
lifecycle_completed_event,
message_finish_event,
message_start_event,
message_text_delta_event,
message_text_finish_event,
tool_finished_event,
tool_started_event,
values_event,
)
fake = SyncFakeServer()
fake.set_state({"final": True})
@@ -610,11 +598,6 @@ def test_v3_streaming_sync_surface_smoke():
def test_interleave_projections_single_channel_values():
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
values_event,
)
fake = SyncFakeServer()
fake.set_state({"counter": 0})
@@ -639,13 +622,6 @@ def test_interleave_projections_single_channel_values():
def test_interleave_projections_values_and_messages_arrival_order():
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
message_finish_event,
message_start_event,
values_event,
)
fake = SyncFakeServer()
fake.set_state({"counter": 0})
@@ -672,12 +648,6 @@ def test_interleave_projections_values_and_messages_arrival_order():
def test_interleave_projections_mixes_builtin_and_extension():
from streaming._events import (
custom_event,
lifecycle_completed_event,
lifecycle_started_event,
values_event,
)
fake = SyncFakeServer()
fake.set_state({"counter": 0})
@@ -701,12 +671,6 @@ def test_interleave_projections_mixes_builtin_and_extension():
def test_interleave_projections_tool_calls_uses_public_name():
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
tool_finished_event,
tool_started_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -735,10 +699,6 @@ def test_interleave_projections_tool_calls_uses_public_name():
def test_interleave_projections_subgraphs_discovers_child():
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -761,11 +721,6 @@ def test_interleave_projections_subgraphs_discovers_child():
def test_interleave_projections_inflight_tool_call_failed_on_break():
"""A tool handle held past an early break is failed in teardown, never left hanging."""
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
tool_started_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -794,10 +749,6 @@ def test_interleave_projections_inflight_tool_call_failed_on_break():
def test_interleave_projections_inflight_subgraph_finished_on_terminal():
"""A discovered subgraph child with no terminal tasks-result is force-completed."""
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -829,10 +780,6 @@ def test_interleave_projections_rejects_reserved_channel(channel):
would subscribe to a channel that never matches and yield nothing. Fail
closed. (`updates`/`checkpoints`/`tasks` are supported and tested below.)
"""
from streaming._events import (
lifecycle_completed_event,
lifecycle_started_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -849,13 +796,6 @@ def test_interleave_projections_rejects_reserved_channel(channel):
def test_interleave_projections_data_channels_yield_payloads():
"""`updates`/`checkpoints`/`tasks` yield their raw `params.data` payloads."""
from streaming._events import (
checkpoints_event,
lifecycle_completed_event,
lifecycle_started_event,
tasks_start_event,
updates_event,
)
fake = SyncFakeServer()
fake.set_state({})
@@ -882,11 +822,6 @@ def test_interleave_projections_data_channels_yield_payloads():
def test_interleave_projections_data_channel_scoped_to_root_namespace():
"""A child-namespace checkpoint must not leak into a root interleave."""
from streaming._events import (
checkpoints_event,
lifecycle_completed_event,
lifecycle_started_event,
)
fake = SyncFakeServer()
fake.set_state({"counter": 0})
libs/sdk-py/tests/streaming/test_sync_transport_ws.py
+2 -3
View File
@@ -6,8 +6,10 @@ import httpx
import orjson
import pytest
from langgraph_sdk.stream.sync_controller import SyncStreamController
from langgraph_sdk.stream.transport.sync_ws import SyncProtocolWebSocketTransport
from streaming._events import values_event
from streaming._sync_fake_server import SyncFakeServer
class _FakeSyncWebSocket:
@@ -102,7 +104,6 @@ def test_sync_websocket_records_post_ready_error():
def test_sync_websocket_send_command_uses_http_commands_endpoint():
from streaming._sync_fake_server import SyncFakeServer
fake = SyncFakeServer()
with httpx.Client(transport=fake.transport, base_url="http://test") as client:
@@ -124,7 +125,6 @@ def test_sync_websocket_open_event_stream_raises_when_closed():
def test_sync_websocket_transport_feeds_sync_stream_controller():
from langgraph_sdk.stream.sync_controller import SyncStreamController
socket = _FakeSyncWebSocket(
[
@@ -164,7 +164,6 @@ def test_sync_websocket_transport_feeds_sync_stream_controller():
def test_sync_websocket_controller_reconnects_with_since_after_drop():
from langgraph_sdk.stream.sync_controller import SyncStreamController
first_socket = _FakeSyncWebSocket(
[values_event(seq=1, values={"counter": 1})],
libs/sdk-py/tests/streaming/test_thread_stream.py
+7 -66
View File
@@ -4,10 +4,14 @@ import asyncio
import contextlib
import re
import uuid
from typing import Any
from typing import Any, cast
import httpx
import pytest
from langchain_protocol import Event
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.stream import AsyncThreadStream
@@ -24,6 +28,8 @@ from streaming._events import (
lifecycle_started_event,
message_finish_event,
message_start_event,
message_text_delta_event,
message_text_finish_event,
tasks_start_event,
tool_finished_event,
tool_started_event,
@@ -225,9 +231,6 @@ async def test_aenter_constructs_transport_with_thread_id():
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
stream = threads.stream(thread_id="t-1", assistant_id="agent")
async with stream:
@@ -237,9 +240,6 @@ async def test_aenter_constructs_transport_with_thread_id():
async def test_aenter_selects_websocket_transport():
async with httpx.AsyncClient(base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
stream = threads.stream(
thread_id="t-1", assistant_id="agent", transport="websocket"
@@ -252,9 +252,6 @@ async def test_aexit_closes_transport():
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
stream = threads.stream(thread_id="t-1", assistant_id="agent")
async with stream:
@@ -268,9 +265,6 @@ async def test_run_start_sends_command_with_assistant_id():
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
result = await thread.run.start(input={"x": 1})
@@ -286,9 +280,6 @@ async def test_command_ids_are_monotonic():
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
await thread.run.start(input={"x": 1})
@@ -300,9 +291,6 @@ async def test_run_start_forwards_config_and_metadata():
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
await thread.run.start(
@@ -316,7 +304,6 @@ async def test_run_start_forwards_config_and_metadata():
async def test_run_start_raises_outside_context_manager():
import pytest
async with httpx.AsyncClient(base_url="http://test") as raw:
stream = AsyncThreadStream(
@@ -327,9 +314,6 @@ async def test_run_start_raises_outside_context_manager():
async def test_run_start_raises_on_error_envelope():
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
async def commands(_request):
return JSONResponse(
@@ -346,11 +330,6 @@ async def test_run_start_raises_on_error_envelope():
)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
import pytest
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
with pytest.raises(RuntimeError, match="invalid_argument"):
@@ -367,9 +346,6 @@ async def test_events_yields_raw_events_after_run_start():
)
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
await thread.run.start(input={})
@@ -383,9 +359,6 @@ async def test_events_subscribes_to_all_channels():
fake.script([])
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
await thread.run.start(input={})
@@ -405,17 +378,11 @@ async def test_events_subscribes_to_all_channels():
async def test_events_terminates_on_aexit():
import asyncio
import pytest
fake = FakeServer()
fake.script([lifecycle_event(seq=i) for i in range(5)])
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
stream = threads.stream(thread_id="t-1", assistant_id="agent")
async with stream as thread:
@@ -462,9 +429,6 @@ async def test_events_property_returns_fresh_iterator_each_access():
fake.script([])
transport = httpx.ASGITransport(app=fake.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
threads = ThreadsClient(HttpClient(raw))
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
first_iter = thread.events
@@ -549,7 +513,6 @@ async def test_unregister_subscription_removes_from_registry():
async def test_await_run_start_gate_honors_timeout():
"""Gate must raise asyncio.TimeoutError if run.start never completes
within the configured timeout."""
import asyncio
async with httpx.AsyncClient(base_url="http://test") as raw:
threads = ThreadsClient(HttpClient(raw))
@@ -568,7 +531,6 @@ async def test_await_run_start_gate_honors_timeout():
async def test_await_run_start_gate_returns_when_gate_resolves_in_time():
"""With a generous timeout and a gate that resolves promptly, the
gate returns without raising."""
import asyncio
async with httpx.AsyncClient(base_url="http://test") as raw:
threads = ThreadsClient(HttpClient(raw))
@@ -583,7 +545,6 @@ async def test_await_run_start_gate_returns_when_gate_resolves_in_time():
async def test_run_start_timeout_constructor_kwarg_forwarded_to_gate():
"""`run_start_timeout` constructor kwarg is stored and consulted by
`_reconcile_stream` via `_await_run_start_gate`."""
import asyncio
async with httpx.AsyncClient(base_url="http://test") as raw:
stream = AsyncThreadStream(
@@ -608,7 +569,6 @@ async def test_subscribe_waits_for_run_start_to_commit():
their SSE. Without it, a fast subscribe would 404 against a thread the
server hasn't created yet.
"""
import asyncio
fake = FakeServer()
fake.script([])
@@ -699,7 +659,6 @@ async def test_run_respond_snapshots_interrupts_under_lock():
`respond()` blocks until the lock is released — proving it serializes
with the terminal-clear path that takes the same lock.
"""
import asyncio
fake = FakeServer()
asgi = httpx.ASGITransport(app=fake.app)
@@ -737,7 +696,6 @@ async def test_terminal_lifecycle_clear_acquires_interrupts_lock():
"""Terminal lifecycle event clears `interrupts` under the same lock
that `respond()` uses, preventing TOCTOU between snapshot and
dispatch."""
import asyncio
fake = FakeServer()
# No scripted events; we exercise `_apply_lifecycle_event` directly.
@@ -754,10 +712,6 @@ async def test_terminal_lifecycle_clear_acquires_interrupts_lock():
# clearing interrupts.
await thread._interrupts_lock.acquire()
try:
from typing import cast
from langchain_protocol import Event
terminal_event = cast(
Event,
{
@@ -909,19 +863,6 @@ async def test_threads_stream_rejects_unknown_transport_option():
async def test_v3_streaming_async_surface_smoke():
import asyncio
from streaming._events import (
custom_event,
lifecycle_completed_event,
message_finish_event,
message_start_event,
message_text_delta_event,
message_text_finish_event,
tool_finished_event,
tool_started_event,
values_event,
)
fake = FakeServer()
fake.set_state({"final": True})
libs/sdk-py/tests/streaming/test_tool_calls_projection.py
+3 -5
View File
@@ -2,12 +2,15 @@
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,
@@ -214,7 +217,6 @@ async def test_tool_calls_explicit_aclose_does_not_block_1s():
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.
from collections.abc import AsyncGenerator
gen: AsyncGenerator = thread.tool_calls._tool_calls_iter()
_call = await gen.__anext__() # receive the one tool-started handle
@@ -230,11 +232,9 @@ def test_tool_call_handle_deltas_queue_is_bounded():
Unbounded queues allow producers to enqueue indefinitely, causing memory
growth when consumers are slow.
"""
import asyncio
# We need a running loop to create the Future inside ToolCallHandle.__init__.
async def _make() -> None:
from langgraph_sdk._async.stream import ToolCallHandle
handle_default = ToolCallHandle(tool_call_id="tc1", name="foo")
assert handle_default._deltas.maxsize > 0, (
@@ -255,10 +255,8 @@ def test_tool_call_handle_deltas_single_consumer_guard():
The property must raise before returning the iterator so the caller
sees the error even without iterating.
"""
import asyncio
async def _run() -> None:
from langgraph_sdk._async.stream import ToolCallHandle
handle = ToolCallHandle(tool_call_id="tc1", name="foo")
libs/sdk-py/tests/streaming/test_transport_http.py
+10 -42
View File
@@ -6,8 +6,17 @@ import contextlib
import httpx
import orjson
import pytest
from starlette.applications import Starlette
from starlette.responses import JSONResponse, Response
from starlette.routing import Route
from langgraph_sdk.stream.transport.http import EventStreamHandle, ProtocolSseTransport
from langgraph_sdk.stream.transport.http import (
EventStreamHandle,
ProtocolSseTransport,
_build_event_stream_body,
)
from streaming._events import lifecycle_event, values_event
from streaming._fake_server import FakeServer
async def test_event_stream_handle_constructs_with_open_state():
@@ -35,7 +44,6 @@ async def test_event_stream_handle_constructs_with_open_state():
async def test_send_command_posts_json_and_returns_response():
from streaming._fake_server import FakeServer
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
@@ -53,9 +61,6 @@ async def test_send_command_posts_json_and_returns_response():
async def test_send_command_returns_none_on_202():
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Route
received: list[dict] = []
@@ -75,7 +80,6 @@ async def test_send_command_returns_none_on_202():
async def test_send_command_raises_when_closed():
from streaming._fake_server import FakeServer
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
@@ -87,9 +91,6 @@ async def test_send_command_raises_when_closed():
async def test_send_command_raises_http_error_on_4xx():
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
async def commands(_request):
return JSONResponse({"error": "bad request"}, status_code=400)
@@ -105,8 +106,6 @@ async def test_send_command_raises_http_error_on_4xx():
async def test_open_event_stream_yields_scripted_events():
from streaming._events import lifecycle_event, values_event
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script(
@@ -129,7 +128,6 @@ async def test_open_event_stream_yields_scripted_events():
async def test_open_event_stream_passes_since_in_body():
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script([])
@@ -145,8 +143,6 @@ async def test_open_event_stream_passes_since_in_body():
async def test_open_event_stream_close_cancels_in_flight_iteration():
from streaming._events import lifecycle_event
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script(
@@ -219,9 +215,6 @@ async def test_mid_stream_error_after_ready_surfaces_on_done():
"""If the SSE response body iteration raises after headers/ready, the
error must be exposed on handle.done so callers can distinguish a clean
end from a transport failure."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
def handler(_request: httpx.Request) -> httpx.Response:
async def body():
@@ -250,9 +243,6 @@ async def test_mid_stream_error_after_ready_surfaces_on_done():
@pytest.mark.anyio
async def test_clean_stream_end_done_resolves_with_none():
"""A stream that ends without error must resolve `done` with None."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
def handler(_request: httpx.Request) -> httpx.Response:
async def body():
@@ -280,9 +270,6 @@ async def test_clean_stream_end_done_resolves_with_none():
async def test_send_command_empty_200_body_raises_runtime_error_not_decoder_error():
"""A 200 response with empty body must raise RuntimeError matching the
'did not return a valid response' contract, not orjson.JSONDecodeError."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
def handler(_request: httpx.Request) -> httpx.Response:
return httpx.Response(200, content=b"")
@@ -305,9 +292,6 @@ async def test_send_command_empty_200_body_raises_runtime_error_not_decoder_erro
async def test_cancel_event_prevents_post_cancel_flush():
"""When the consumer cancels the handle mid-stream, the pump's decoder
flush MUST NOT emit additional events after the cancel point."""
import httpx
from langgraph_sdk.stream.transport.http import ProtocolSseTransport
received: list = []
@@ -342,9 +326,6 @@ async def test_cancel_event_prevents_post_cancel_flush():
@pytest.mark.anyio
async def test_open_event_stream_ready_rejects_on_5xx():
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
async def stream_events(_request):
return JSONResponse({"error": "boom"}, status_code=500)
@@ -371,14 +352,12 @@ async def test_open_event_stream_ready_rejects_on_5xx():
def test_build_event_stream_body_minimal_channels_only():
from langgraph_sdk.stream.transport.http import _build_event_stream_body
body = _build_event_stream_body({"channels": ["values"]})
assert body == {"channels": ["values"]}
def test_build_event_stream_body_includes_all_optional_fields():
from langgraph_sdk.stream.transport.http import _build_event_stream_body
body = _build_event_stream_body(
{
@@ -397,14 +376,12 @@ def test_build_event_stream_body_includes_all_optional_fields():
def test_build_event_stream_body_omits_since_when_not_int():
from langgraph_sdk.stream.transport.http import _build_event_stream_body
body = _build_event_stream_body({"channels": ["values"], "since": None})
assert "since" not in body
async def test_open_event_stream_raises_when_closed():
from streaming._fake_server import FakeServer
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
@@ -416,8 +393,6 @@ async def test_open_event_stream_raises_when_closed():
async def test_transport_close_cancels_open_event_streams():
from streaming._events import lifecycle_event
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script([lifecycle_event(seq=i) for i in range(5)], delay=0.05)
@@ -439,7 +414,6 @@ async def test_transport_close_cancels_open_event_streams():
async def test_default_headers_forwarded_to_send_command():
"""Headers passed at construction are sent on every command request."""
from streaming._fake_server import FakeServer
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
@@ -457,7 +431,6 @@ async def test_default_headers_forwarded_to_send_command():
async def test_default_headers_forwarded_to_open_event_stream():
"""Headers passed at construction are sent on every SSE stream request."""
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script([])
@@ -479,7 +452,6 @@ async def test_default_headers_forwarded_to_open_event_stream():
async def test_default_headers_cannot_override_sse_fixed_headers():
"""Caller-supplied default headers must not override content-type or accept."""
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.script([])
@@ -506,7 +478,6 @@ async def test_default_headers_cannot_override_sse_fixed_headers():
async def test_fake_server_state_endpoint():
"""State endpoint returns the set state and increments the counter."""
from streaming._fake_server import FakeServer
fake = FakeServer()
fake.set_state({"foo": "bar"}, next=["node_a"])
@@ -527,7 +498,6 @@ async def test_fake_server_state_endpoint():
def test_values_event_builder_shape():
"""values_event produces the expected shape with params.data as the snapshot."""
from streaming._events import values_event
evt = values_event(seq=1, values={"foo": 1})
assert evt["event_id"] == "evt-1"
@@ -537,13 +507,11 @@ def test_values_event_builder_shape():
async def test_open_event_stream_done_records_post_ready_error():
from streaming._events import values_event
event_data = values_event(seq=1)
class _FailAfterOneStream(httpx.AsyncByteStream):
async def __aiter__(self):
import orjson
payload = orjson.dumps(event_data).decode()
yield f"id: {event_data.get('event_id', '')}\n".encode()
libs/sdk-py/tests/streaming/test_transport_ws.py
+2 -4
View File
@@ -10,8 +10,10 @@ import pytest
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
from websockets.frames import Close
from langgraph_sdk.stream.controller import StreamController
from langgraph_sdk.stream.transport.ws import ProtocolWebSocketTransport
from streaming._events import values_event
from streaming._fake_server import FakeServer
class _FakeAsyncWebSocket:
@@ -334,7 +336,6 @@ async def test_websocket_done_records_post_ready_error():
async def test_websocket_send_command_uses_http_commands_endpoint():
from streaming._fake_server import FakeServer
fake = FakeServer()
transport = httpx.ASGITransport(app=fake.app)
@@ -357,7 +358,6 @@ async def test_websocket_open_event_stream_raises_when_closed():
async def test_websocket_transport_feeds_async_stream_controller():
from langgraph_sdk.stream.controller import StreamController
socket = _FakeAsyncWebSocket(
[
@@ -418,7 +418,6 @@ async def test_ws_transport_default_max_queue_size_is_1024():
async def test_websocket_controller_reconnects_with_since_after_drop():
from langgraph_sdk.stream.controller import StreamController
first_socket = _FakeAsyncWebSocket(
[values_event(seq=1, values={"counter": 1})],
@@ -465,7 +464,6 @@ async def test_websocket_controller_reconnects_with_since_after_drop():
async def test_async_close_sends_normal_close_frame():
"""`handle.close()` sends a WebSocket close frame with code 1000 explicitly."""
import asyncio
# Use an event to distinguish an explicit close(code=1000) call from
# the implicit one in __aexit__ when the task is cancelled.
libs/sdk-py/tests/test_client_stream.py
+2 -2
View File
@@ -8,7 +8,9 @@ import httpx
import pytest
from typing_extensions import assert_type
from langgraph_sdk._async.runs import _wrap_stream_v2
from langgraph_sdk._shared.utilities import _sse_to_v2_dict
from langgraph_sdk._sync.runs import _wrap_stream_v2_sync
from langgraph_sdk.client import HttpClient, SyncHttpClient
from langgraph_sdk.schema import (
CheckpointPayload,
@@ -380,7 +382,6 @@ def test_sse_to_v2_dict_values_with_interrupts() -> None:
@pytest.mark.asyncio
async def test_async_stream_v2_client_side_conversion() -> None:
from langgraph_sdk._async.runs import _wrap_stream_v2
async def mock_stream() -> Any:
yield StreamPart(event="metadata", data={"run_id": "r1"})
@@ -415,7 +416,6 @@ async def test_async_stream_v2_client_side_conversion() -> None:
def test_sync_stream_v2_client_side_conversion() -> None:
from langgraph_sdk._sync.runs import _wrap_stream_v2_sync
def mock_stream() -> Any:
yield StreamPart(event="metadata", data={"run_id": "r1"})
libs/sdk-py/tests/test_langsmith_tracing.py
+2 -5
View File
@@ -7,6 +7,8 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from langgraph_sdk._async.runs import RunsClient
from langgraph_sdk._sync.runs import SyncRunsClient
from langgraph_sdk.schema import LangSmithTracing
@@ -24,7 +26,6 @@ class TestLangSmithTracingPayload:
@pytest.mark.asyncio
async def test_async_create_includes_langsmith_tracer(self, tracing_config):
"""Test that async create sends langsmith_tracer in payload."""
from langgraph_sdk._async.runs import RunsClient
captured: dict[str, Any] = {}
@@ -50,7 +51,6 @@ class TestLangSmithTracingPayload:
def test_sync_create_includes_langsmith_tracer(self, tracing_config):
"""Test that sync create sends langsmith_tracer in payload."""
from langgraph_sdk._sync.runs import SyncRunsClient
captured: dict[str, Any] = {}
@@ -76,7 +76,6 @@ class TestLangSmithTracingPayload:
def test_sync_wait_includes_langsmith_tracer(self, tracing_config):
"""Test that sync wait sends langsmith_tracer in payload."""
from langgraph_sdk._sync.runs import SyncRunsClient
captured: dict[str, Any] = {}
@@ -102,7 +101,6 @@ class TestLangSmithTracingPayload:
def test_create_without_langsmith_tracing_excludes_key(self):
"""Test that langsmith_tracer is not in payload when not provided."""
from langgraph_sdk._sync.runs import SyncRunsClient
captured: dict[str, Any] = {}
@@ -123,7 +121,6 @@ class TestLangSmithTracingPayload:
def test_langsmith_tracing_project_name_only(self):
"""Test that langsmith_tracing works with only project_name."""
from langgraph_sdk._sync.runs import SyncRunsClient
captured: dict[str, Any] = {}
libs/sdk-py/tests/test_path_encoding.py
+2 -1
View File
@@ -8,6 +8,8 @@ URL paths.
from __future__ import annotations
import uuid
import httpx
import pytest
@@ -60,7 +62,6 @@ class TestQuotePathParam:
assert "/" not in encoded
def test_non_string_values_are_coerced_to_str(self) -> None:
import uuid
uid = uuid.UUID("550e8400-e29b-41d4-a716-446655440000")
assert _quote_path_param(uid) == str(uid)
libs/sdk-py/tests/test_serde.py
+1 -1
View File
@@ -1,3 +1,4 @@
from dataclasses import dataclass
from typing import Any
import orjson
@@ -36,7 +37,6 @@ async def test_serde_pydantic():
async def test_serde_dataclass():
from dataclasses import dataclass
@dataclass
class TestDataClass: