diff --git a/tests/test_langgraph_runtime_adapter.py b/tests/test_langgraph_runtime_adapter.py index 8b69446..7436e16 100644 --- a/tests/test_langgraph_runtime_adapter.py +++ b/tests/test_langgraph_runtime_adapter.py @@ -4,7 +4,10 @@ import importlib.util import pytest +from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphRun, GraphRunStatus from graph.langgraph_runtime import LangGraphRuntime +from graph.registry import NodeHandlerRegistry, NodeResult +from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec def test_langgraph_runtime_reports_missing_dependency() -> None: @@ -15,3 +18,38 @@ def test_langgraph_runtime_reports_missing_dependency() -> None: with pytest.raises(RuntimeError, match="langgraph package|NodeHandlerRegistry"): runtime.run_until_terminal_or_paused(None) # type: ignore[arg-type] + + +class LangGraphFixtureNode: + node_type = "langgraph_fixture" + idempotent = True + replay_safe = True + destructive = False + + def run(self, context: object) -> NodeResult: + return NodeResult("COMPLETE", "success", {"executed": True}, {"model_requests": 0}) + + +def test_langgraph_runtime_executes_artifex_graph_when_installed() -> None: + if importlib.util.find_spec("langgraph") is None: + pytest.skip("langgraph is not installed") + spec = ExecutionGraphSpec( + name="langgraph_fixture", + version=1, + graph_type="FIXTURE", + entry="step", + nodes={"step": GraphNodeSpec("step", "langgraph_fixture"), "done": GraphNodeSpec("done", "terminal")}, + edges=[GraphEdgeSpec("step", "done", "success")], + terminal_nodes=["done"], + ) + definition = ExecutionGraphDefinition.objects.create(name=spec.name, graph_type=spec.graph_type) + version = ExecutionGraphVersion.objects.create(graph=definition, version=1, status=ExecutionGraphVersionStatus.CHAMPION, graph_spec=spec.to_dict()) + graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=spec.entry) + registry = NodeHandlerRegistry() + registry.register(LangGraphFixtureNode()) + + result = LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run) + + assert result.status == GraphRunStatus.COMPLETE + assert result.current_node == "done" + assert result.node_runs.get(node_id="step").output_metadata["executed"] is True