From 711d81bc38b1bc845f0bfc1c7895cb0e8ae07ea6 Mon Sep 17 00:00:00 2001 From: Eugene Yurtsev Date: Tue, 5 Aug 2025 13:57:06 -0400 Subject: [PATCH] Test with multiple interrupts --- libs/langgraph/tests/test_interruption.py | 71 +++++++++++++++-------- 1 file changed, 47 insertions(+), 24 deletions(-) diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index 9fb25a140..3a013b0ff 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -1,10 +1,19 @@ +<<<<<<< HEAD import pytest from langgraph.checkpoint.base import BaseCheckpointSaver from typing_extensions import TypedDict, Annotated import operator +======= +import operator + +import pytest +from typing_extensions import Annotated, TypedDict + +from langgraph.checkpoint.base import BaseCheckpointSaver +>>>>>>> 0ce981a0 (Test with multiple interrupts) from langgraph.graph import END, START, StateGraph -from langgraph.types import Send, interrupt, Command, Durability +from langgraph.types import Command, Durability, Send, interrupt pytestmark = pytest.mark.anyio @@ -95,51 +104,65 @@ async def test_interruption_without_state_updates_async( def test_interrupt_with_send_payloads(sync_checkpointer: BaseCheckpointSaver) -> None: """Test interruption in map node with Send payloads and human-in-the-loop resume.""" - + + # Global counter to track node executions + node_counter = {"entry": 0, "map_node": 0} + class State(TypedDict): items: list[str] processed: Annotated[list[str], operator.add] - + def entry_node(state: State): - return {"items": ["item1", "item2"]} - + node_counter["entry"] += 1 + return {} # No state updates in entry node + def send_to_map(state: State): return [Send("map_node", {"item": item}) for item in state["items"]] - + def map_node(state: State): - value = interrupt({"processing": state["item"]}) - return {"processed": [f"processed_{value}"]} - + node_counter["map_node"] += 1 + if "dangerous" in state["item"]: + value = interrupt({"processing": state["item"]}) + return {"processed": [f"processed_{value}"]} + else: + return {"processed": [f"processed_{state['item']}_auto"]} + builder = StateGraph(State) builder.add_node("entry", entry_node) builder.add_node("map_node", map_node) builder.add_edge(START, "entry") builder.add_conditional_edges("entry", send_to_map, ["map_node"]) builder.add_edge("map_node", END) - + graph = builder.compile(checkpointer=sync_checkpointer) - + config = {"configurable": {"thread_id": "test_interrupt_send"}} - + # Run until interrupts - result = graph.invoke({"items": [], "processed": []}, config=config) - - # Verify we have interrupts + result = graph.invoke({"items": ["item1", "dangerous_item"]}, config=config) + + # Verify we have interrupts (only one for dangerous_item) interrupts = result.get("__interrupt__", []) - assert len(interrupts) == 2 - assert all(i.resumable for i in interrupts) - + assert len(interrupts) == 1 + assert "dangerous_item" in interrupts[0].value["processing"] + # Resume with mapping of interrupt IDs to values resume_map = { - i.interrupt_id: f"human_input_{i.value['processing']}" - for i in interrupts + i.interrupt_id: f"human_input_{i.value['processing']}" for i in interrupts } - + final_result = graph.invoke(Command(resume=resume_map), config=config) - + # Verify final result contains processed items assert "processed" in final_result processed_items = final_result["processed"] assert len(processed_items) == 2 - assert "processed_human_input_item1" in processed_items - assert "processed_human_input_item2" in processed_items + assert "processed_item1_auto" in processed_items # item1 processed automatically + assert ( + "processed_human_input_dangerous_item" in processed_items + ) # dangerous_item processed after interrupt + + # Verify node execution counts + assert node_counter["entry"] == 1 # Entry node runs once + # Map node runs twice initially (item1 completes, dangerous_item interrupts), then once on resume + assert node_counter["map_node"] == 3