mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-26 17:42:24 +02:00
Elior Nataf LackritzandGitHub
ea5f9cc9fb
Follow-up to #8540, which turned on `PLC0415` (import-outside-top-level) for checkpoint-postgres and checkpoint-sqlite. This does the remaining six packages: checkpoint, checkpoint-conformance, langgraph, prebuilt, cli, sdk-py. Scoped to tests, per @sydney-runkle's call on #8540: library code is exempted with `per-file-ignores`, since it still has deferred imports nobody has reviewed and mixing that in would make this hard to read. ## What changed Function-level imports across 56 test files moved to module level. Nine could not move and carry an explicit `# noqa: PLC0415` with a reason: | File | Why it stays local | |---|---| | `libs/langgraph/tests/test_deprecation.py` (4) | the import has to run inside `pytest.warns` for the warning to be observed | | `libs/langgraph/tests/test_serde_allowlist.py` | try/except guard, skips when langchain_core is absent | | `libs/langgraph/tests/test_delta_channel_benchmark.py` | optional psycopg probe | | `libs/checkpoint/tests/test_conformance_delta.py` (3) | protected by a module-level `pytest.importorskip`; hoisting past the guard turns a skip into a collection error | That last one is the trap: an import moved above `pytest.importorskip` silently defeats the guard. I hit it locally and it turned the skip into a `ModuleNotFoundError` at collection. Every file with an `importorskip` or `except ImportError` was checked by hand for this. ## Verification `make lint` and `make test` in each of the six: | Package | Tests | |---|---| | checkpoint | 156 passed, 17 skipped | | checkpoint-conformance | 1 passed | | langgraph | 1968 passed, 4 skipped | | prebuilt | 284 passed | | cli | 336 passed | | sdk-py | 493 passed | Also confirmed the rule actually fires: a throwaway test file with a function-level import is flagged in all six packages, and the source exemption holds.
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Factory-graph execution regression test.
|
|
|
|
Unlike the other integration graphs (all pre-compiled), `factory_agent` is a
|
|
graph *factory*, so executing a run against it drives the server's graph-factory
|
|
code path. That path regressed in langgraph 1.2.3: a leaked `__pregel_runtime`
|
|
(an SDK `_ExecutionRuntime`) survived `ensure_config`'s configurable-merge into
|
|
`astream`, which then raised
|
|
`AttributeError: '_ExecutionRuntime' object has no attribute 'control'`. A
|
|
successful `runs.wait` here proves the factory path executes end to end.
|
|
"""
|
|
|
|
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):
|
|
|
|
return RunsClient(HttpClient(raw))
|
|
|
|
|
|
def _sync_runs(raw):
|
|
|
|
return SyncRunsClient(SyncHttpClient(raw))
|
|
|
|
|
|
async def test_factory_graph_executes_async(async_threads) -> None:
|
|
"""A run against a factory graph completes and echoes the input."""
|
|
threads, raw = async_threads
|
|
runs = _async_runs(raw)
|
|
thread = await threads.create(
|
|
metadata={"suite": "integration", "label": "factory-async"}
|
|
)
|
|
tid = thread["thread_id"]
|
|
try:
|
|
result = await runs.wait(tid, FACTORY_ASSISTANT_ID, input={"text": "hi"})
|
|
assert isinstance(result, dict), result
|
|
assert result.get("text") == "hi echoed"
|
|
assert result.get("access_context") == "threads.create_run"
|
|
assert result.get("is_for_execution") is True
|
|
finally:
|
|
await threads.delete(tid)
|
|
|
|
|
|
def test_factory_graph_executes_sync(sync_threads) -> None:
|
|
threads, raw = sync_threads
|
|
runs = _sync_runs(raw)
|
|
thread = threads.create(metadata={"suite": "integration", "label": "factory-sync"})
|
|
tid = thread["thread_id"]
|
|
try:
|
|
result = runs.wait(tid, FACTORY_ASSISTANT_ID, input={"text": "hi"})
|
|
assert isinstance(result, dict), result
|
|
assert result.get("text") == "hi echoed"
|
|
assert result.get("access_context") == "threads.create_run"
|
|
assert result.get("is_for_execution") is True
|
|
finally:
|
|
threads.delete(tid)
|