test_async_middleware_with_can_jump_to_graph_snapshot() — langchain Function Reference
Architecture documentation for the test_async_middleware_with_can_jump_to_graph_snapshot() function in test_decorators.py from the langchain codebase.
Entity Profile
Dependency Diagram
graph TD 5c25d42d_8717_6afd_6929_7361aee18580["test_async_middleware_with_can_jump_to_graph_snapshot()"] 85a104a3_11cf_6539_d0c0_ba69a17a41d4["test_decorators.py"] 5c25d42d_8717_6afd_6929_7361aee18580 -->|defined in| 85a104a3_11cf_6539_d0c0_ba69a17a41d4 5db83528_8e08_d7fe_e2fc_f806b038e621["after_model()"] 5c25d42d_8717_6afd_6929_7361aee18580 -->|calls| 5db83528_8e08_d7fe_e2fc_f806b038e621 28fa7487_7074_9d90_d3ef_bb64640336d2["before_model()"] 5c25d42d_8717_6afd_6929_7361aee18580 -->|calls| 28fa7487_7074_9d90_d3ef_bb64640336d2 style 5c25d42d_8717_6afd_6929_7361aee18580 fill:#6366f1,stroke:#818cf8,color:#fff
Relationship Graph
Source Code
libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_decorators.py lines 556–623
def test_async_middleware_with_can_jump_to_graph_snapshot(snapshot: SnapshotAssertion) -> None:
"""Test async middleware with can_jump_to graph snapshot.
Test that async middleware with `can_jump_to` creates correct graph structure with
conditional edges.
"""
# Test 1: Async before_model with can_jump_to
@before_model(can_jump_to=["end"])
async def async_before_with_jump(
state: AgentState[Any], *_args: Any, **_kwargs: Any
) -> dict[str, Any] | None:
if len(state["messages"]) > 5:
return {"jump_to": "end"}
return None
agent_async_before = create_agent(
model=FakeToolCallingModel(), middleware=[async_before_with_jump]
)
assert agent_async_before.get_graph().draw_mermaid() == snapshot
# Test 2: Async after_model with can_jump_to
@after_model(can_jump_to=["model", "end"])
async def async_after_with_jump(
state: AgentState[Any], *_args: Any, **_kwargs: Any
) -> dict[str, Any] | None:
if state["messages"][-1].content == "retry":
return {"jump_to": "model"}
return None
agent_async_after = create_agent(
model=FakeToolCallingModel(), middleware=[async_after_with_jump]
)
assert agent_async_after.get_graph().draw_mermaid() == snapshot
# Test 3: Multiple async middleware with can_jump_to
@before_model(can_jump_to=["end"])
async def async_before_early_exit(*_args: Any, **_kwargs: Any) -> dict[str, Any] | None:
return None
@after_model(can_jump_to=["model"])
async def async_after_retry(*_args: Any, **_kwargs: Any) -> dict[str, Any] | None:
return None
agent_multiple_async = create_agent(
model=FakeToolCallingModel(),
middleware=[async_before_early_exit, async_after_retry],
)
assert agent_multiple_async.get_graph().draw_mermaid() == snapshot
# Test 4: Mixed sync and async middleware with can_jump_to
@before_model(can_jump_to=["end"])
def sync_before_with_jump(*_args: Any, **_kwargs: Any) -> dict[str, Any] | None:
return None
@after_model(can_jump_to=["model", "end"])
async def async_after_with_jumps(*_args: Any, **_kwargs: Any) -> dict[str, Any] | None:
return None
agent_mixed = create_agent(
model=FakeToolCallingModel(),
middleware=[sync_before_with_jump, async_after_with_jumps],
)
assert agent_mixed.get_graph().draw_mermaid() == snapshot
Domain
Subdomains
Source
Frequently Asked Questions
What does test_async_middleware_with_can_jump_to_graph_snapshot() do?
test_async_middleware_with_can_jump_to_graph_snapshot() is a function in the langchain codebase, defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_decorators.py.
Where is test_async_middleware_with_can_jump_to_graph_snapshot() defined?
test_async_middleware_with_can_jump_to_graph_snapshot() is defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_decorators.py at line 556.
What does test_async_middleware_with_can_jump_to_graph_snapshot() call?
test_async_middleware_with_can_jump_to_graph_snapshot() calls 2 function(s): after_model, before_model.
Analyze Your Own Codebase
Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.
Try Supermodel Free