HAL9000
f51c630cf0
The test-infrastructure patches asyncio.sleep with a 10 ms cap to speed up retry waits. The two slow-executor Behave step definitions used asyncio.sleep(10) as the "slow" coroutine, which was silently capped to 10 ms — the same duration as the 0.01 s executor timeout — creating a race condition that caused the "Executor times out via thread pool path" and "Executor times out via run_coroutine_threadsafe path" scenarios to fail intermittently. Fix: use asyncio._original_sleep (falling back to asyncio.sleep when the patch is absent) with a 0.5 s delay, which is 50× longer than the timeout and guarantees the timeout always fires before the coroutine completes.
1091 lines
41 KiB
Python
1091 lines
41 KiB
Python
import asyncio
|
|
import concurrent.futures
|
|
import threading
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from rx.scheduler.eventloop import AsyncIOScheduler
|
|
|
|
from cleveragents.langgraph.graph import (
|
|
MAX_EXECUTION_HISTORY,
|
|
GraphConfig,
|
|
LangGraph,
|
|
)
|
|
from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveragents.langgraph.state import GraphState, StateManager
|
|
from cleveragents.reactive.stream_router import StreamMessage
|
|
|
|
# Existing steps cover constructor scheduling, execution dispatch, start validation,
|
|
# cycle detection, and node executor state updates.
|
|
|
|
|
|
@given("a langgraph config with layered nodes")
|
|
def step_layered_config(context):
|
|
nodes = {
|
|
"a": NodeConfig(name="a", type=NodeType.FUNCTION),
|
|
"b": NodeConfig(name="b", type=NodeType.FUNCTION),
|
|
"c": NodeConfig(name="c", type=NodeType.FUNCTION),
|
|
}
|
|
edges = [Edge(source="a", target="b"), Edge(source="b", target="c")]
|
|
context.graph = _fresh_graph(
|
|
GraphConfig(name="graph-levels", nodes=nodes, edges=edges)
|
|
)
|
|
|
|
|
|
@when("I compute the topological levels")
|
|
def step_compute_levels(context):
|
|
context.levels = context.graph._topological_levels() # pylint: disable=protected-access
|
|
|
|
|
|
@then("the levels should list each layer")
|
|
def step_assert_levels(context):
|
|
assert context.levels[0] == {"a"}
|
|
assert context.levels[1] == {"b"}
|
|
assert context.levels[2] == {"c"}
|
|
|
|
|
|
@given("a langgraph instance with recorded history")
|
|
def step_history_instance(context):
|
|
config = GraphConfig(name="graph-history")
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.execution_history.extend(["x", "y"])
|
|
|
|
|
|
@when("I request execution history")
|
|
def step_get_history(context):
|
|
context.history_result = context.graph.get_execution_history()
|
|
|
|
|
|
@then("it should return a copy of the recorded history")
|
|
def step_assert_history_copy(context):
|
|
assert context.history_result == ["x", "y"]
|
|
assert context.history_result is not context.graph.execution_history
|
|
|
|
|
|
@given("a langgraph config with explicit start and end nodes")
|
|
def step_explicit_guards(context):
|
|
start = NodeConfig(name="start", type=NodeType.START, metadata={"guard": True})
|
|
end = NodeConfig(name="end", type=NodeType.END, metadata={"guard": True})
|
|
nodes = {
|
|
"start": start,
|
|
"end": end,
|
|
"worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
|
|
}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-guards", nodes=nodes))
|
|
|
|
|
|
@when("I construct the LangGraph with that config")
|
|
def step_construct_with_guards(_context):
|
|
# Construction already performed in the given step
|
|
return
|
|
|
|
|
|
@then("initialization should reuse the provided start and end nodes")
|
|
def step_assert_guard_reuse(context):
|
|
assert context.graph.nodes["start"].config is context.graph.config.nodes["start"]
|
|
assert context.graph.nodes["end"].config is context.graph.config.nodes["end"]
|
|
|
|
|
|
@given("a langgraph instance prepared to observe node streams")
|
|
def step_prepare_node_stream(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-observe", nodes=nodes))
|
|
context.worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
context.graph.logger = MagicMock()
|
|
|
|
|
|
@when("the node stream emits next error and completion events")
|
|
def step_emit_stream_events(context):
|
|
context.worker_stream.on_next("payload")
|
|
context.worker_stream.on_error(ValueError("boom"))
|
|
context.worker_stream.on_completed()
|
|
|
|
|
|
@then("the subscription handlers should be exercised")
|
|
def step_assert_stream_handlers(context):
|
|
context.graph.logger.error.assert_called_once()
|
|
|
|
|
|
@given("a langgraph config with converging branches")
|
|
def step_converging_config(context):
|
|
nodes = {
|
|
"a": NodeConfig(name="a", type=NodeType.FUNCTION),
|
|
"b": NodeConfig(name="b", type=NodeType.FUNCTION),
|
|
"c": NodeConfig(name="c", type=NodeType.FUNCTION),
|
|
}
|
|
edges = [
|
|
Edge(source="start", target="a"),
|
|
Edge(source="start", target="b"),
|
|
Edge(source="a", target="c"),
|
|
Edge(source="b", target="c"),
|
|
Edge(source="c", target="end"),
|
|
]
|
|
context.graph = _fresh_graph(
|
|
GraphConfig(name="graph-parallel", nodes=nodes, edges=edges)
|
|
)
|
|
|
|
|
|
@when("I construct the LangGraph for parallel groups")
|
|
def step_construct_parallel(_context):
|
|
return
|
|
|
|
|
|
@then("parallel groups should reflect staged execution order")
|
|
def step_assert_parallel_groups(context):
|
|
assert context.graph.parallel_groups[0] == {"start"}
|
|
assert context.graph.parallel_groups[1] == {"a", "b"}
|
|
assert context.graph.parallel_groups[2] == {"c"}
|
|
assert context.graph.parallel_groups[3] == {"end"}
|
|
|
|
|
|
@given("a langgraph instance missing required guard nodes")
|
|
def step_missing_guard_nodes(context):
|
|
config = GraphConfig(name="graph-missing-guards")
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.nodes.pop("start", None)
|
|
|
|
|
|
@when("I validate the graph explicitly")
|
|
def step_validate_graph(context):
|
|
context.validation_error = None
|
|
try:
|
|
context.graph._validate_graph() # pylint: disable=protected-access
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.validation_error = exc
|
|
|
|
|
|
@then("a guard validation error should be raised")
|
|
def step_assert_guard_validation(context):
|
|
assert isinstance(context.validation_error, ValueError)
|
|
|
|
|
|
@given("a langgraph instance ready to start")
|
|
def step_ready_to_start(context):
|
|
config = GraphConfig(name="graph-start")
|
|
context.graph = _fresh_graph(config)
|
|
context.start_stream = f"__{context.graph.name}_node_start__"
|
|
context.graph.stream_router.send_message = MagicMock()
|
|
|
|
|
|
@when("I start without input then stop and fetch state")
|
|
def step_start_and_stop(context):
|
|
context.graph.start()
|
|
context.graph.stop()
|
|
context.final_state = context.graph.get_state()
|
|
|
|
|
|
@then("it should use default state and update running flags")
|
|
def step_assert_start_stop(context):
|
|
context.graph.stream_router.send_message.assert_called_with(
|
|
context.start_stream, GraphState()
|
|
)
|
|
assert context.graph.is_running is False
|
|
assert isinstance(context.final_state, GraphState)
|
|
|
|
|
|
def _fresh_graph(config: GraphConfig) -> LangGraph:
|
|
"""Construct a LangGraph with isolated scheduler and router."""
|
|
return LangGraph(config=config, stream_router=None, scheduler=None)
|
|
|
|
|
|
@given("a langgraph config with only start and end nodes")
|
|
def step_minimal_config(context):
|
|
context.graph_config = GraphConfig(name="graph-minimal")
|
|
|
|
|
|
@given("asyncio has no running loop")
|
|
def step_no_running_loop(_context):
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
if loop.is_running():
|
|
loop.stop()
|
|
except RuntimeError:
|
|
# Expected when no loop is running
|
|
pass
|
|
|
|
|
|
@when("I construct a LangGraph without providing a scheduler")
|
|
def step_construct_graph(context):
|
|
context.graph = _fresh_graph(context.graph_config)
|
|
|
|
|
|
@then("it should create an AsyncIOScheduler and initialize streams")
|
|
def step_assert_scheduler_and_streams(context):
|
|
assert isinstance(context.graph.scheduler, AsyncIOScheduler)
|
|
start_stream = f"__{context.graph.name}_node_start__"
|
|
assert start_stream in context.graph.stream_router.streams
|
|
|
|
|
|
@given("a langgraph instance with a fresh state manager")
|
|
def step_graph_with_state_manager(context):
|
|
config = GraphConfig(name="graph-exec")
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = True
|
|
context.start_stream = f"__{context.graph.name}_node_start__"
|
|
context.graph.stream_router.send_message = MagicMock()
|
|
|
|
|
|
@when("I execute the graph with input data")
|
|
def step_execute_graph(context):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
try:
|
|
result = loop.run_until_complete(context.graph.execute({"messages": []}))
|
|
context.execute_result = result
|
|
finally:
|
|
loop.close()
|
|
replacement_loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(replacement_loop)
|
|
context.add_cleanup(replacement_loop.close)
|
|
|
|
|
|
@then("the start stream should receive the state and execution returns a GraphState")
|
|
def step_assert_execute(context):
|
|
context.graph.stream_router.send_message.assert_called_once()
|
|
called_state = context.graph.stream_router.send_message.call_args.args[1]
|
|
assert isinstance(called_state, GraphState)
|
|
assert isinstance(context.execute_result, GraphState)
|
|
|
|
|
|
@given("a langgraph instance with a removed start stream")
|
|
def step_graph_missing_start(context):
|
|
config = GraphConfig(name="graph-missing-start")
|
|
context.graph = _fresh_graph(config)
|
|
start_stream = f"__{context.graph.name}_node_start__"
|
|
context.graph.stream_router.streams.pop(start_stream, None)
|
|
|
|
|
|
@when("I call start on the graph")
|
|
def step_call_start(context):
|
|
context.start_error = None
|
|
try:
|
|
context.graph.start()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.start_error = exc
|
|
|
|
|
|
@then("a start stream error should be raised")
|
|
def step_assert_start_error(context):
|
|
assert isinstance(context.start_error, ValueError)
|
|
|
|
|
|
@given("a langgraph config with parallel execution disabled and a cycle")
|
|
def step_cycle_config(context):
|
|
nodes = {
|
|
"a": NodeConfig(name="a", type=NodeType.FUNCTION),
|
|
}
|
|
edges = [Edge(source="start", target="a"), Edge(source="a", target="start")]
|
|
context.graph_config = GraphConfig(
|
|
name="graph-cycle",
|
|
nodes=nodes,
|
|
edges=edges,
|
|
parallel_execution=False,
|
|
)
|
|
|
|
|
|
@when("I construct the LangGraph")
|
|
def step_construct_cycle_graph(context):
|
|
context.graph = _fresh_graph(context.graph_config)
|
|
|
|
|
|
@then("the graph should report cycles and empty parallel groups")
|
|
def step_assert_cycles(context):
|
|
assert context.graph.has_cycles is True
|
|
assert context.graph.parallel_groups == []
|
|
|
|
|
|
@given("a langgraph instance with a patched state manager")
|
|
def step_graph_with_executor(context):
|
|
nodes = {
|
|
"worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
|
|
}
|
|
config = GraphConfig(name="graph-executor", nodes=nodes)
|
|
graph = _fresh_graph(config)
|
|
graph.is_running = True
|
|
graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
graph.state_manager.update_state = MagicMock()
|
|
graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["m"]})
|
|
context.executor = graph._node_executors["worker"] # pylint: disable=protected-access
|
|
context.graph = graph
|
|
|
|
|
|
@when("I invoke the builtin node executor")
|
|
def step_invoke_executor(context):
|
|
result = context.executor(StreamMessage(content="x", metadata={}))
|
|
context.executor_result = result
|
|
|
|
|
|
@then("it should update state and record execution history")
|
|
def step_assert_executor(context):
|
|
context.graph.state_manager.update_state.assert_called_once()
|
|
# deque does not compare equal to list; convert first
|
|
assert list(context.graph.execution_history) == ["worker"]
|
|
assert isinstance(context.executor_result, StreamMessage)
|
|
|
|
|
|
# ── Node stream on_next wiring tests ──────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance with a mocked node executor")
|
|
def step_prepare_mocked_executor(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-exec-wired", nodes=nodes))
|
|
context.mock_executor = MagicMock(
|
|
return_value=StreamMessage(content={}, metadata={})
|
|
)
|
|
context.graph._node_executors["worker"] = context.mock_executor # pylint: disable=protected-access
|
|
|
|
|
|
@when("a message is emitted on the worker node stream")
|
|
def step_emit_worker_stream_message(context):
|
|
worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
worker_stream.on_next(StreamMessage(content="test-payload", metadata={}))
|
|
|
|
|
|
@then("the registered executor should be called with the message")
|
|
def step_assert_executor_called(context):
|
|
context.mock_executor.assert_called_once()
|
|
call_args = context.mock_executor.call_args
|
|
msg = call_args[0][0]
|
|
assert isinstance(msg, StreamMessage)
|
|
assert msg.content == "test-payload"
|
|
|
|
|
|
@when("a raw value is emitted on the worker node stream")
|
|
def step_emit_raw_value_on_stream(context):
|
|
worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
worker_stream.on_next("raw-string-payload")
|
|
|
|
|
|
@then("the executor should receive a StreamMessage wrapping the raw value")
|
|
def step_assert_wrapped_message(context):
|
|
context.mock_executor.assert_called_once()
|
|
call_args = context.mock_executor.call_args
|
|
msg = call_args[0][0]
|
|
assert isinstance(msg, StreamMessage)
|
|
assert msg.content == "raw-string-payload"
|
|
assert msg.metadata == {}
|
|
|
|
|
|
@given("a langgraph instance with a removed node executor")
|
|
def step_prepare_missing_executor(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-no-exec", nodes=nodes))
|
|
context.graph._node_executors.pop("worker", None) # pylint: disable=protected-access
|
|
context.graph.logger = MagicMock()
|
|
|
|
|
|
@when("a message is emitted on the executorless worker stream")
|
|
def step_emit_executorless_message(context):
|
|
worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
worker_stream.on_next("test")
|
|
|
|
|
|
@then("a warning about missing executor should be logged")
|
|
def step_assert_warning_logged(context):
|
|
context.graph.logger.warning.assert_called_once()
|
|
args, _kwargs = context.graph.logger.warning.call_args
|
|
assert args[0] == "No executor found for node %s"
|
|
assert args[1] == "worker"
|
|
|
|
|
|
@given("a langgraph instance with multiple mocked node executors")
|
|
def step_prepare_multi_executors(context):
|
|
nodes = {
|
|
"alpha": NodeConfig(name="alpha", type=NodeType.FUNCTION),
|
|
"beta": NodeConfig(name="beta", type=NodeType.FUNCTION),
|
|
}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-multi", nodes=nodes))
|
|
context.alpha_executor = MagicMock(
|
|
return_value=StreamMessage(content={}, metadata={})
|
|
)
|
|
context.beta_executor = MagicMock(
|
|
return_value=StreamMessage(content={}, metadata={})
|
|
)
|
|
context.graph._node_executors["alpha"] = context.alpha_executor # pylint: disable=protected-access
|
|
context.graph._node_executors["beta"] = context.beta_executor # pylint: disable=protected-access
|
|
|
|
|
|
@when("messages are emitted on both alpha and beta node streams")
|
|
def step_emit_both_messages(context):
|
|
alpha_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_alpha__"
|
|
]
|
|
beta_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_beta__"
|
|
]
|
|
alpha_stream.on_next(StreamMessage(content="alpha-payload", metadata={}))
|
|
beta_stream.on_next(StreamMessage(content="beta-payload", metadata={}))
|
|
|
|
|
|
@then("each executor should receive its own message independently")
|
|
def step_assert_independent_executors(context):
|
|
context.alpha_executor.assert_called_once()
|
|
context.beta_executor.assert_called_once()
|
|
alpha_msg = context.alpha_executor.call_args[0][0]
|
|
beta_msg = context.beta_executor.call_args[0][0]
|
|
assert isinstance(alpha_msg, StreamMessage)
|
|
assert isinstance(beta_msg, StreamMessage)
|
|
assert alpha_msg.content == "alpha-payload"
|
|
assert beta_msg.content == "beta-payload"
|
|
|
|
|
|
# ── Executor exception handling (M3) ─────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance with a failing node executor")
|
|
def step_prepare_failing_executor(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-fail-exec", nodes=nodes))
|
|
context.graph._node_executors["worker"] = MagicMock( # pylint: disable=protected-access
|
|
side_effect=RuntimeError("executor boom")
|
|
)
|
|
context.graph.logger = MagicMock()
|
|
|
|
|
|
@when("a message is emitted on the failing worker stream")
|
|
def step_emit_failing_worker_message(context):
|
|
worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
context.on_next_error = None
|
|
try:
|
|
worker_stream.on_next(StreamMessage(content="trigger", metadata={}))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.on_next_error = exc
|
|
|
|
|
|
@then("the exception should be caught and logged without propagating")
|
|
def step_assert_exception_caught(context):
|
|
assert context.on_next_error is None, (
|
|
f"on_next raised unexpectedly: {context.on_next_error}"
|
|
)
|
|
context.graph.logger.exception.assert_called_once()
|
|
args, _kwargs = context.graph.logger.exception.call_args
|
|
assert args[0] == "Executor failed for node %s"
|
|
assert args[1] == "worker"
|
|
|
|
|
|
# ── Missing observable branch coverage (m3) ──────────────────────────
|
|
|
|
|
|
@given("a langgraph instance with a removed observable")
|
|
def step_prepare_removed_observable(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-no-obs", nodes=nodes))
|
|
stream_name = f"__{context.graph.name}_node_worker__"
|
|
context.graph.stream_router.observables.pop(stream_name, None)
|
|
context.graph.logger = MagicMock()
|
|
|
|
|
|
@when("node stream subscriptions are set up")
|
|
def step_rerun_subscriptions(context):
|
|
# Note: calling _setup_node_stream_subscriptions a second time creates
|
|
# duplicate subscriptions for start/end nodes. This is expected and
|
|
# harmless in this test context — we only care about the worker branch.
|
|
context.graph._setup_node_stream_subscriptions() # pylint: disable=protected-access
|
|
|
|
|
|
@then("no executor should be called and no error should occur")
|
|
def step_assert_no_error_on_missing_observable(context):
|
|
context.graph.logger.debug.assert_any_call(
|
|
"No observable for stream %s; skipping subscription",
|
|
f"__{context.graph.name}_node_worker__",
|
|
)
|
|
context.graph.logger.error.assert_not_called()
|
|
context.graph.logger.warning.assert_not_called()
|
|
|
|
|
|
# ── Execute raises when start stream is missing (m1) ─────────────────
|
|
|
|
|
|
@given("a langgraph instance with a removed start stream for execute")
|
|
def step_prepare_execute_missing_start(context):
|
|
config = GraphConfig(name="graph-exec-no-start")
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = True
|
|
start_stream = f"__{context.graph.name}_node_start__"
|
|
context.graph.stream_router.streams.pop(start_stream, None)
|
|
|
|
|
|
@when("I execute the graph expecting a start stream error")
|
|
def step_execute_expecting_error(context):
|
|
context.execute_error = None
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
try:
|
|
loop.run_until_complete(context.graph.execute({"messages": []}))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.execute_error = exc
|
|
finally:
|
|
loop.close()
|
|
replacement_loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(replacement_loop)
|
|
context.add_cleanup(replacement_loop.close)
|
|
|
|
|
|
@then("a start stream missing error should be raised from execute")
|
|
def step_assert_execute_start_error(context):
|
|
assert isinstance(context.execute_error, ValueError)
|
|
assert "Start stream not initialized" in str(context.execute_error)
|
|
|
|
|
|
# ── run_coroutine_threadsafe path coverage (M5) ──────────────────────
|
|
|
|
|
|
def _cleanup_bg_loop(context):
|
|
"""Shut down the background event loop and join the thread."""
|
|
bg_loop = getattr(context, "_bg_loop", None)
|
|
bg_thread = getattr(context, "_bg_thread", None)
|
|
try:
|
|
if bg_loop is not None:
|
|
bg_loop.call_soon_threadsafe(bg_loop.stop)
|
|
except RuntimeError:
|
|
# Loop may already be closed; ensure thread join still happens.
|
|
pass
|
|
if bg_thread is not None:
|
|
bg_thread.join(timeout=5.0)
|
|
if bg_loop is not None:
|
|
bg_loop.close()
|
|
|
|
|
|
@given(
|
|
"a langgraph instance with a scheduler whose loop is running in a background thread"
|
|
)
|
|
def step_prepare_running_loop_graph(context):
|
|
bg_loop = asyncio.new_event_loop()
|
|
|
|
def run_loop() -> None:
|
|
asyncio.set_event_loop(bg_loop)
|
|
bg_loop.run_forever()
|
|
|
|
bg_thread = threading.Thread(target=run_loop, daemon=True)
|
|
bg_thread.start()
|
|
context._bg_loop = bg_loop
|
|
context._bg_thread = bg_thread
|
|
# Register cleanup so the background loop is always stopped, even if
|
|
# assertions fail in the @then step.
|
|
context.add_cleanup(_cleanup_bg_loop, context)
|
|
|
|
scheduler = AsyncIOScheduler(bg_loop)
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
config = GraphConfig(name="graph-running-loop", nodes=nodes)
|
|
context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
|
|
context.graph.is_running = True
|
|
context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
context.graph.state_manager.update_state = MagicMock()
|
|
context.graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["bg"]})
|
|
|
|
|
|
@when("a message is emitted on the worker stream via the running loop path")
|
|
def step_emit_via_running_loop(context):
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
with patch(
|
|
"asyncio.run_coroutine_threadsafe", wraps=asyncio.run_coroutine_threadsafe
|
|
) as mock_rcts:
|
|
context.executor_result = executor(
|
|
StreamMessage(content="bg-test", metadata={})
|
|
)
|
|
context._mock_run_coroutine_threadsafe = mock_rcts
|
|
|
|
|
|
@then("the executor should complete successfully via run_coroutine_threadsafe")
|
|
def step_assert_running_loop_executor(context):
|
|
assert isinstance(context.executor_result, StreamMessage)
|
|
assert context.executor_result.metadata["node"] == "worker"
|
|
context.graph.state_manager.update_state.assert_called_once()
|
|
# Verify that run_coroutine_threadsafe was actually used (not the
|
|
# thread-pool fallback path). The mock retains call history after
|
|
# the patch context manager exits in the @when step.
|
|
context._mock_run_coroutine_threadsafe.assert_called_once()
|
|
|
|
|
|
# ── Execution history cap test (m5) ──────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance with execution history at capacity")
|
|
def step_prepare_full_history(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
config = GraphConfig(name="graph-history-cap", nodes=nodes)
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = True
|
|
context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
context.graph.state_manager.update_state = MagicMock()
|
|
context.graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["m"]})
|
|
# Pre-fill to capacity
|
|
for i in range(MAX_EXECUTION_HISTORY):
|
|
context.graph.execution_history.append(f"entry-{i}")
|
|
context.oldest_entry = "entry-0"
|
|
|
|
|
|
@when("one more executor invocation is recorded")
|
|
def step_invoke_one_more(context):
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
executor(StreamMessage(content="x", metadata={}))
|
|
|
|
|
|
@then(
|
|
"the history length should remain at the maximum and the oldest entry should be dropped"
|
|
)
|
|
def step_assert_history_cap(context):
|
|
assert len(context.graph.execution_history) == MAX_EXECUTION_HISTORY
|
|
history_list = list(context.graph.execution_history)
|
|
assert context.oldest_entry not in history_list
|
|
assert history_list[-1] == "worker"
|
|
|
|
|
|
# ── Concurrent state update lock test (m8) ───────────────────────────
|
|
|
|
|
|
@given("a state manager configured for concurrent access")
|
|
def step_prepare_concurrent_state_manager(context):
|
|
context.state_manager = StateManager(initial_state=GraphState())
|
|
context.num_threads = 20
|
|
context.updates_per_thread = 10
|
|
|
|
|
|
@when("multiple threads perform state updates simultaneously")
|
|
def step_concurrent_updates(context):
|
|
barrier = threading.Barrier(context.num_threads)
|
|
|
|
def updater() -> None:
|
|
barrier.wait(timeout=10.0)
|
|
for _ in range(context.updates_per_thread):
|
|
context.state_manager.update_state(
|
|
{"metadata": {"t": threading.current_thread().name}}
|
|
)
|
|
|
|
threads = [threading.Thread(target=updater) for _ in range(context.num_threads)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=30.0)
|
|
assert all(not t.is_alive() for t in threads), "Some threads did not complete"
|
|
|
|
|
|
@then("the final execution count should equal the total number of updates")
|
|
def step_assert_concurrent_count(context):
|
|
expected = context.num_threads * context.updates_per_thread
|
|
actual = context.state_manager.get_state().execution_count
|
|
assert actual == expected, f"Expected {expected}, got {actual}"
|
|
|
|
|
|
# ── TimeoutError branch coverage (M5) ────────────────────────────────
|
|
|
|
|
|
@given(
|
|
"a langgraph instance with a slow executor on a background loop and a short timeout"
|
|
)
|
|
def step_prepare_slow_executor_bg_loop(context):
|
|
bg_loop = asyncio.new_event_loop()
|
|
|
|
def run_loop() -> None:
|
|
asyncio.set_event_loop(bg_loop)
|
|
bg_loop.run_forever()
|
|
|
|
bg_thread = threading.Thread(target=run_loop, daemon=True)
|
|
bg_thread.start()
|
|
context._bg_loop = bg_loop
|
|
context._bg_thread = bg_thread
|
|
context.add_cleanup(_cleanup_bg_loop, context)
|
|
|
|
scheduler = AsyncIOScheduler(bg_loop)
|
|
# Use a very short timeout (0.01s) so the test completes quickly.
|
|
nodes = {
|
|
"worker": NodeConfig(name="worker", type=NodeType.FUNCTION, timeout=0.01),
|
|
}
|
|
config = GraphConfig(name="graph-timeout-rcts", nodes=nodes)
|
|
context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
|
|
context.graph.is_running = True
|
|
context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
context.graph.state_manager.update_state = MagicMock()
|
|
|
|
async def slow_execute(state):
|
|
# Use the original (un-patched) asyncio.sleep so the test-infrastructure
|
|
# 10 ms sleep cap does not race with the 0.01 s executor timeout.
|
|
_real_sleep = getattr(asyncio, "_original_sleep", asyncio.sleep)
|
|
await _real_sleep(0.5)
|
|
return {"messages": ["never"]}
|
|
|
|
context.graph.nodes["worker"].execute = slow_execute
|
|
|
|
|
|
@when("the slow executor is invoked via the running loop path")
|
|
def step_invoke_slow_executor_rcts(context):
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
context.timeout_error = None
|
|
try:
|
|
executor(StreamMessage(content="slow", metadata={}))
|
|
except TimeoutError as exc:
|
|
context.timeout_error = exc
|
|
|
|
|
|
@then("a TimeoutError should be raised with the node name and timeout duration")
|
|
def step_assert_timeout_rcts(context):
|
|
assert context.timeout_error is not None, "Expected TimeoutError was not raised"
|
|
msg = str(context.timeout_error)
|
|
assert "worker" in msg
|
|
assert "0.01" in msg
|
|
|
|
|
|
@given("a langgraph instance with a slow executor and a short timeout")
|
|
def step_prepare_slow_executor_tp(context):
|
|
nodes = {
|
|
"worker": NodeConfig(name="worker", type=NodeType.FUNCTION, timeout=0.01),
|
|
}
|
|
config = GraphConfig(name="graph-timeout-tp", nodes=nodes)
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = True
|
|
context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
context.graph.state_manager.update_state = MagicMock()
|
|
|
|
async def slow_execute(state):
|
|
# Use the original (un-patched) asyncio.sleep so the test-infrastructure
|
|
# 10 ms sleep cap does not race with the 0.01 s executor timeout.
|
|
_real_sleep = getattr(asyncio, "_original_sleep", asyncio.sleep)
|
|
await _real_sleep(0.5)
|
|
return {"messages": ["never"]}
|
|
|
|
context.graph.nodes["worker"].execute = slow_execute
|
|
|
|
|
|
@when("the slow executor is invoked via the thread pool path")
|
|
def step_invoke_slow_executor_tp(context):
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
context.timeout_error = None
|
|
try:
|
|
executor(StreamMessage(content="slow", metadata={}))
|
|
except TimeoutError as exc:
|
|
context.timeout_error = exc
|
|
|
|
|
|
@then("a TimeoutError should be raised from the thread pool path with the node name")
|
|
def step_assert_timeout_tp(context):
|
|
assert context.timeout_error is not None, "Expected TimeoutError was not raised"
|
|
msg = str(context.timeout_error)
|
|
assert "worker" in msg
|
|
assert "0.01" in msg
|
|
|
|
|
|
# ── Restart-after-stop test (m7) ─────────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance that has been started and stopped")
|
|
def step_prepare_restart_graph(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
config = GraphConfig(name="graph-restart", nodes=nodes)
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.stream_router.send_message = MagicMock()
|
|
context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
|
|
context.graph.state_manager.update_state = MagicMock()
|
|
context.graph.nodes["worker"].execute = AsyncMock(
|
|
return_value={"messages": ["restarted"]}
|
|
)
|
|
context.graph.start()
|
|
context.graph.stop()
|
|
assert context.graph.is_running is False
|
|
|
|
|
|
@when("I restart the graph and invoke an executor")
|
|
def step_restart_and_invoke(context):
|
|
context.graph.start()
|
|
# Ensure the pool is cleaned up even if assertions fail later.
|
|
context.add_cleanup(context.graph.stop)
|
|
# Invoke the executor directly to verify the pool is recreated after
|
|
# restart. This does NOT verify the full stream→executor subscription
|
|
# path; that is covered by the on_next wiring scenarios above.
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
context.restart_result = executor(
|
|
StreamMessage(content="after-restart", metadata={})
|
|
)
|
|
|
|
|
|
@then("the executor should complete successfully after restart")
|
|
def step_assert_restart_success(context):
|
|
assert isinstance(context.restart_result, StreamMessage)
|
|
assert context.restart_result.metadata["node"] == "worker"
|
|
assert context.graph.is_running is True
|
|
# stop() is handled by context.add_cleanup registered in the @when step.
|
|
|
|
|
|
# ── is_running guard test (m4) ───────────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance that is not running")
|
|
def step_prepare_not_running_graph(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
config = GraphConfig(name="graph-not-running", nodes=nodes)
|
|
context.graph = _fresh_graph(config)
|
|
# Ensure the graph is explicitly not running.
|
|
context.graph.is_running = False
|
|
|
|
|
|
@when("I invoke the node executor directly")
|
|
def step_invoke_executor_not_running(context):
|
|
executor = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
context.not_running_error = None
|
|
try:
|
|
executor(StreamMessage(content="should-fail", metadata={}))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.not_running_error = exc
|
|
|
|
|
|
@then("a RuntimeError should be raised indicating the graph is not running")
|
|
def step_assert_not_running_error(context):
|
|
assert isinstance(context.not_running_error, RuntimeError)
|
|
assert "not running" in str(context.not_running_error)
|
|
assert "worker" in str(context.not_running_error)
|
|
|
|
|
|
# ── StateManager.replace_state() direct test (m6) ───────────────────
|
|
|
|
|
|
@given("a state manager with an initial state")
|
|
def step_prepare_state_manager_for_replace(context):
|
|
context.state_manager = StateManager(initial_state=GraphState())
|
|
context.stream_emissions = [] # list[GraphState]
|
|
context.state_manager.state_stream.subscribe(
|
|
on_next=lambda s: context.stream_emissions.append(s)
|
|
)
|
|
|
|
|
|
@when("I call replace_state with a new state")
|
|
def step_call_replace_state(context):
|
|
context.new_state = GraphState(metadata={"replaced": True}, current_node="new-node")
|
|
context.state_manager.replace_state(context.new_state)
|
|
|
|
|
|
@then("get_state should return the new state")
|
|
def step_assert_replace_state_get(context):
|
|
retrieved = context.state_manager.get_state()
|
|
assert retrieved.metadata == {"replaced": True}
|
|
assert retrieved.current_node == "new-node"
|
|
|
|
|
|
@then("the state stream should have received a notification")
|
|
def step_assert_replace_state_stream(context):
|
|
# BehaviorSubject emits the initial value on subscribe, plus the
|
|
# replace_state emission, so we expect at least 2 emissions.
|
|
assert len(context.stream_emissions) >= 2
|
|
|
|
|
|
@then("the emitted state should be a deep copy not the same object")
|
|
def step_assert_replace_state_deep_copy(context):
|
|
# The last emission should be a deep copy, not the same object as
|
|
# the internal state or the new_state we passed in.
|
|
emitted = context.stream_emissions[-1]
|
|
assert emitted is not context.new_state
|
|
assert emitted.metadata == {"replaced": True}
|
|
|
|
|
|
@then("the emitted state should be distinct from the internal state")
|
|
def step_assert_emitted_distinct_from_internal(context):
|
|
# After the M1 fix (deep-copy on store), the emitted copy must also
|
|
# be a distinct object from the internal state held by the manager.
|
|
emitted = context.stream_emissions[-1]
|
|
# Direct access intentional: identity check requires the actual
|
|
# internal object, not a get_state() copy.
|
|
assert emitted is not context.state_manager.state
|
|
|
|
|
|
# ── CancelledError path coverage (M3) ───────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance with a cancellable executor task")
|
|
def step_prepare_cancellable_executor(context):
|
|
nodes = {
|
|
"worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
|
|
}
|
|
config = GraphConfig(name="graph-cancel", nodes=nodes)
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = True
|
|
|
|
|
|
@when("the executor future is cancelled before completion")
|
|
def step_cancel_executor_future(context):
|
|
context.cancel_error = None
|
|
# Patch the executor pool's submit to return a future that raises
|
|
# CancelledError on result(). This simulates graph shutdown
|
|
# (stop() calls shutdown(wait=True, cancel_futures=True)) cancelling
|
|
# a queued-but-not-yet-started task.
|
|
cancelled_future: concurrent.futures.Future[StreamMessage] = (
|
|
concurrent.futures.Future()
|
|
)
|
|
# cancel() on a fresh Future is sufficient to put it in CANCELLED state.
|
|
cancelled_future.cancel()
|
|
|
|
executor_fn = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
with patch.object(
|
|
context.graph._executor_pool, # pylint: disable=protected-access
|
|
"submit",
|
|
return_value=cancelled_future,
|
|
):
|
|
try:
|
|
executor_fn(StreamMessage(content="cancel-me", metadata={}))
|
|
except RuntimeError as exc:
|
|
context.cancel_error = exc
|
|
|
|
|
|
@then("a RuntimeError should be raised indicating graph stopping")
|
|
def step_assert_cancelled_error(context):
|
|
assert context.cancel_error is not None, "Expected RuntimeError was not raised"
|
|
msg = str(context.cancel_error)
|
|
assert "cancelled" in msg and "stopping" in msg, f"Unexpected error message: {msg}"
|
|
|
|
|
|
# ── __del__ safety net test (m3) ─────────────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance that was never stopped")
|
|
def step_prepare_unstopped_graph(context):
|
|
config = GraphConfig(name="graph-del-test")
|
|
context.graph = _fresh_graph(config)
|
|
# Ensure the graph has a live executor pool (never stopped).
|
|
assert context.graph._executor_pool is not None # pylint: disable=protected-access
|
|
|
|
|
|
@when("__del__ is called on the graph")
|
|
def step_call_del(context):
|
|
context.del_error = None
|
|
try:
|
|
context.graph.__del__()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.del_error = exc
|
|
|
|
|
|
# The @then step "no exception should propagate from __del__" is already
|
|
# defined in bridge_coverage_steps.py and is reused here. It checks
|
|
# ``context.del_error is None``.
|
|
|
|
|
|
# ── execute() is_running guard test (n1) ─────────────────────────────
|
|
|
|
|
|
@given("a langgraph instance that is not running for execute")
|
|
def step_prepare_not_running_for_execute(context):
|
|
config = GraphConfig(name="graph-exec-not-running")
|
|
context.graph = _fresh_graph(config)
|
|
context.graph.is_running = False
|
|
|
|
|
|
@when("I call execute on the non-running graph")
|
|
def step_call_execute_not_running(context):
|
|
context.execute_not_running_error = None
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
try:
|
|
loop.run_until_complete(context.graph.execute({"messages": []}))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.execute_not_running_error = exc
|
|
finally:
|
|
loop.close()
|
|
replacement_loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(replacement_loop)
|
|
context.add_cleanup(replacement_loop.close)
|
|
|
|
|
|
@then("a RuntimeError should be raised indicating the graph is not running for execute")
|
|
def step_assert_execute_not_running_error(context):
|
|
assert isinstance(context.execute_not_running_error, RuntimeError)
|
|
assert "not running" in str(context.execute_not_running_error)
|
|
|
|
|
|
# ── TimeoutError through stream path (M2 cycle 6) ───────────────────
|
|
|
|
|
|
@given("a langgraph instance with an executor that raises TimeoutError")
|
|
def step_prepare_timeout_executor_for_stream(context):
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="graph-timeout-stream", nodes=nodes))
|
|
context.graph._node_executors["worker"] = MagicMock( # pylint: disable=protected-access
|
|
side_effect=TimeoutError("timed out")
|
|
)
|
|
context.graph.logger = MagicMock()
|
|
|
|
|
|
@when("a message is emitted on the timeout worker stream")
|
|
def step_emit_timeout_worker_stream(context):
|
|
worker_stream = context.graph.stream_router.streams[
|
|
f"__{context.graph.name}_node_worker__"
|
|
]
|
|
context.stream_error = None
|
|
try:
|
|
worker_stream.on_next(StreamMessage(content="trigger-timeout", metadata={}))
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.stream_error = exc
|
|
|
|
|
|
@then("the timeout should be logged at warning level")
|
|
def step_assert_timeout_warning_logged(context):
|
|
assert context.stream_error is None, (
|
|
f"on_next raised unexpectedly: {context.stream_error}"
|
|
)
|
|
context.graph.logger.warning.assert_called_once()
|
|
args, _kwargs = context.graph.logger.warning.call_args
|
|
assert args[0] == "Executor timed out for node %s"
|
|
assert args[1] == "worker"
|
|
|
|
|
|
@then("logger exception should not be called for timeout")
|
|
def step_assert_no_exception_log_for_timeout(context):
|
|
context.graph.logger.exception.assert_not_called()
|
|
|
|
|
|
# ── CancelledError in run_coroutine_threadsafe path (M1 cycle 6) ────
|
|
|
|
|
|
@given("a langgraph instance with a scheduler loop and a cancellable coroutine future")
|
|
def step_prepare_rcts_cancellable(context):
|
|
bg_loop = asyncio.new_event_loop()
|
|
|
|
def run_loop() -> None:
|
|
asyncio.set_event_loop(bg_loop)
|
|
bg_loop.run_forever()
|
|
|
|
bg_thread = threading.Thread(target=run_loop, daemon=True)
|
|
bg_thread.start()
|
|
context._bg_loop = bg_loop
|
|
context._bg_thread = bg_thread
|
|
context.add_cleanup(_cleanup_bg_loop, context)
|
|
|
|
scheduler = AsyncIOScheduler(bg_loop)
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
config = GraphConfig(name="graph-cancel-rcts", nodes=nodes)
|
|
context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
|
|
context.graph.is_running = True
|
|
|
|
|
|
@when("the coroutine future is cancelled before completion")
|
|
def step_cancel_coroutine_future(context):
|
|
context.cancel_rcts_error = None
|
|
cancelled_future: concurrent.futures.Future[StreamMessage] = (
|
|
concurrent.futures.Future()
|
|
)
|
|
# cancel() on a fresh Future is sufficient to put it in CANCELLED state.
|
|
cancelled_future.cancel()
|
|
|
|
executor_fn = context.graph._node_executors["worker"] # pylint: disable=protected-access
|
|
with patch(
|
|
"asyncio.run_coroutine_threadsafe",
|
|
return_value=cancelled_future,
|
|
):
|
|
try:
|
|
executor_fn(StreamMessage(content="cancel-rcts", metadata={}))
|
|
except RuntimeError as exc:
|
|
context.cancel_rcts_error = exc
|
|
|
|
|
|
@then("a RuntimeError should be raised indicating graph stopping from coroutine path")
|
|
def step_assert_rcts_cancelled_error(context):
|
|
assert context.cancel_rcts_error is not None, "Expected RuntimeError was not raised"
|
|
msg = str(context.cancel_rcts_error)
|
|
assert "cancelled" in msg and "stopping" in msg, f"Unexpected error message: {msg}"
|