163 lines
6.1 KiB
Python
163 lines
6.1 KiB
Python
"""Step definitions for TDD Issue #10398.
|
|
|
|
Verifies that LangGraph._setup_node_stream_subscriptions stores the
|
|
Disposable returned by observable.subscribe() and that stop() disposes
|
|
all stored subscriptions, preventing resource leaks.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.langgraph.graph import GraphConfig, LangGraph
|
|
from cleveragents.langgraph.nodes import NodeConfig, NodeType
|
|
|
|
|
|
def _fresh_graph(config: GraphConfig) -> LangGraph:
|
|
"""Construct a LangGraph with isolated scheduler and router."""
|
|
return LangGraph(config=config, stream_router=None, scheduler=None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the LangGraph module is available")
|
|
def step_langgraph_module_available(_context: object) -> None:
|
|
"""Verify the LangGraph class can be imported (import already at top)."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: LangGraph initialises an empty subscriptions list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I construct a LangGraph with default configuration")
|
|
def step_construct_default_graph(context: object) -> None:
|
|
context.graph = _fresh_graph(GraphConfig(name="tdd-10398-default"))
|
|
|
|
|
|
@then("the graph should have an empty _subscriptions list")
|
|
def step_assert_subscriptions_attr_exists(context: object) -> None:
|
|
# The attribute must exist and be a list (it will be populated by
|
|
# _setup_node_stream_subscriptions, so it won't be empty after __init__,
|
|
# but the attribute itself must be a list).
|
|
assert hasattr(context.graph, "_subscriptions"), (
|
|
"LangGraph must have a _subscriptions attribute"
|
|
)
|
|
assert isinstance(context.graph._subscriptions, list), (
|
|
"_subscriptions must be a list"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: _setup_node_stream_subscriptions stores Disposables
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a LangGraph instance with one worker node")
|
|
def step_graph_with_worker_node(context: object) -> None:
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="tdd-10398-worker", nodes=nodes))
|
|
|
|
|
|
@when("I inspect the subscriptions after construction")
|
|
def step_inspect_subscriptions(_context: object) -> None:
|
|
"""No action needed — subscriptions are populated during __init__."""
|
|
|
|
|
|
@then("the _subscriptions list should contain at least one entry per node")
|
|
def step_assert_subscriptions_populated(context: object) -> None:
|
|
# The graph has start, end, and worker nodes — each should have a
|
|
# subscription stored.
|
|
subs = context.graph._subscriptions
|
|
assert len(subs) >= len(context.graph.nodes), (
|
|
f"Expected at least {len(context.graph.nodes)} subscriptions, got {len(subs)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: stop() disposes all stored subscriptions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a LangGraph instance with tracked disposable subscriptions")
|
|
def step_graph_with_tracked_disposables(context: object) -> None:
|
|
nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
|
|
context.graph = _fresh_graph(GraphConfig(name="tdd-10398-tracked", nodes=nodes))
|
|
|
|
# Replace the real disposables with mocks so we can assert dispose() calls.
|
|
context.mock_disposables: list[MagicMock] = []
|
|
for _i in range(len(context.graph._subscriptions)):
|
|
mock_disp = MagicMock()
|
|
context.mock_disposables.append(mock_disp)
|
|
context.graph._subscriptions = list(context.mock_disposables)
|
|
|
|
|
|
@when("I call stop() on the graph")
|
|
def step_call_stop(context: object) -> None:
|
|
context.stop_error: Exception | None = None
|
|
try:
|
|
context.graph.stop()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.stop_error = exc
|
|
|
|
|
|
@then("every tracked disposable should have been disposed")
|
|
def step_assert_all_disposed(context: object) -> None:
|
|
assert context.stop_error is None, (
|
|
f"stop() raised unexpectedly: {context.stop_error}"
|
|
)
|
|
for i, mock_disp in enumerate(context.mock_disposables):
|
|
(
|
|
mock_disp.dispose.assert_called_once(),
|
|
(f"Disposable #{i} was not disposed by stop()"),
|
|
)
|
|
|
|
|
|
@then("the _subscriptions list should be empty after stop")
|
|
def step_assert_subscriptions_cleared(context: object) -> None:
|
|
assert context.graph._subscriptions == [], (
|
|
"_subscriptions should be cleared after stop()"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenario: stop() clears subscriptions even when dispose raises
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a LangGraph instance with a subscription that raises on dispose")
|
|
def step_graph_with_failing_disposable(context: object) -> None:
|
|
context.graph = _fresh_graph(GraphConfig(name="tdd-10398-fail-disp"))
|
|
|
|
failing_disp = MagicMock()
|
|
failing_disp.dispose.side_effect = RuntimeError("dispose failed")
|
|
context.graph._subscriptions = [failing_disp]
|
|
|
|
|
|
@when("I call stop() on the graph with a failing disposable")
|
|
def step_call_stop_with_failing_disposable(context: object) -> None:
|
|
context.stop_error = None
|
|
try:
|
|
context.graph.stop()
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.stop_error = exc
|
|
|
|
|
|
@then("stop() should complete without raising an exception")
|
|
def step_assert_stop_no_exception(context: object) -> None:
|
|
assert context.stop_error is None, (
|
|
f"stop() raised unexpectedly when a disposable failed: {context.stop_error}"
|
|
)
|
|
|
|
|
|
@then("the _subscriptions list should be empty after the failing stop")
|
|
def step_assert_subscriptions_cleared_after_fail(context: object) -> None:
|
|
assert context.graph._subscriptions == [], (
|
|
"_subscriptions should be cleared even when dispose raises"
|
|
)
|