Add LangGraph adapter execution test

This commit is contained in:
Daniel Maddern 2026-08-15 17:19:41 +07:00
parent 6d149f424b
commit de0f92e71b

View file

@ -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