mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-12 04:37:51 +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.
160 lines
4.8 KiB
Python
160 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Annotated, Any, Literal, NewType, Optional, Union
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
from typing_extensions import NotRequired, Required, TypedDict
|
|
|
|
from langgraph._internal._serde import (
|
|
collect_allowlist_from_schemas,
|
|
curated_core_allowlist,
|
|
)
|
|
|
|
|
|
class Color(Enum):
|
|
RED = "red"
|
|
BLUE = "blue"
|
|
|
|
|
|
@dataclass
|
|
class InnerDataclass:
|
|
value: int
|
|
|
|
|
|
class InnerModel(BaseModel):
|
|
name: str
|
|
|
|
|
|
@dataclass
|
|
class Node:
|
|
value: int
|
|
child: Node | None = None
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
class MissingType:
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class MissingRefDataclass:
|
|
payload: MissingType
|
|
|
|
|
|
class Payload(TypedDict):
|
|
item: InnerDataclass
|
|
maybe: NotRequired[InnerModel]
|
|
required: Required[str]
|
|
|
|
|
|
@dataclass
|
|
class NestedDataclass:
|
|
inner: InnerDataclass
|
|
items: list[InnerModel]
|
|
mapping: dict[str, InnerDataclass]
|
|
optional: InnerModel | None
|
|
union: InnerDataclass | InnerModel
|
|
queue: deque[InnerDataclass]
|
|
frozen: frozenset[InnerModel]
|
|
|
|
|
|
AnnotatedList = Annotated[list[InnerDataclass], "meta"]
|
|
UserId = NewType("UserId", int)
|
|
|
|
|
|
class DummyChannel:
|
|
@property
|
|
def ValueType(self) -> type[InnerDataclass]:
|
|
return InnerDataclass
|
|
|
|
@property
|
|
def UpdateType(self) -> type[InnerModel]:
|
|
return InnerModel
|
|
|
|
|
|
def test_curated_core_allowlist_includes_messages() -> None:
|
|
try:
|
|
from langchain_core.messages import BaseMessage # noqa: PLC0415
|
|
except Exception:
|
|
pytest.skip("langchain_core not available")
|
|
allowlist = curated_core_allowlist()
|
|
assert (BaseMessage.__module__, BaseMessage.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_basic_models() -> None:
|
|
allowlist = collect_allowlist_from_schemas(
|
|
schemas=[InnerDataclass, InnerModel, Color]
|
|
)
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
assert (Color.__module__, Color.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_nested_containers() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[NestedDataclass])
|
|
assert (NestedDataclass.__module__, NestedDataclass.__name__) in allowlist
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_annotated_and_union() -> None:
|
|
allowlist = collect_allowlist_from_schemas(
|
|
schemas=[AnnotatedList, InnerModel | None, InnerDataclass | None]
|
|
)
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_literal_and_any() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[Any, Literal["a"]])
|
|
assert allowlist == set()
|
|
|
|
|
|
def test_collect_allowlist_typeddict_fields_only() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[Payload])
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
assert (Payload.__module__, Payload.__name__) not in allowlist
|
|
|
|
|
|
def test_collect_allowlist_forward_refs() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[Node])
|
|
assert (Node.__module__, Node.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_missing_forward_ref() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[MissingRefDataclass])
|
|
assert allowlist == {(MissingRefDataclass.__module__, MissingRefDataclass.__name__)}
|
|
|
|
|
|
def test_collect_allowlist_newtype_supertype() -> None:
|
|
allowlist = collect_allowlist_from_schemas(schemas=[UserId])
|
|
assert allowlist == set()
|
|
|
|
|
|
def test_collect_allowlist_channels() -> None:
|
|
channels = {"a": DummyChannel(), "b": DummyChannel()}
|
|
allowlist = collect_allowlist_from_schemas(channels=channels)
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_pep604_union() -> None:
|
|
schema = InnerDataclass | InnerModel
|
|
allowlist = collect_allowlist_from_schemas(schemas=[schema])
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|
|
|
|
|
|
def test_collect_allowlist_typing_union_optional() -> None:
|
|
typing_optional = Optional[InnerDataclass] # noqa: UP045
|
|
typing_union = Union[InnerDataclass, InnerModel] # noqa: UP007
|
|
allowlist = collect_allowlist_from_schemas(schemas=[typing_optional, typing_union])
|
|
assert (InnerDataclass.__module__, InnerDataclass.__name__) in allowlist
|
|
assert (InnerModel.__module__, InnerModel.__name__) in allowlist
|