55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
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:
|
|
if importlib.util.find_spec("langgraph") is not None:
|
|
pytest.skip("langgraph is installed; adapter execution is covered by integration parity tests")
|
|
|
|
runtime = LangGraphRuntime()
|
|
|
|
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
|