af53111928
Address all architecture review findings: P0 fixes: - Add API key authentication for admin endpoints (analytics, replay, openapi) and WebSocket connections via ADMIN_API_KEY env var - Add PostgreSQL-backed PgSessionManager and PgInterruptManager for multi-worker production deployments (in-memory defaults preserved) P1 fixes: - Implement actual tool generation in OpenAPI approve_job endpoint using generate_tool_code() and generate_agent_yaml() - Add missing clarification, interrupt_expired, and tool_result message handlers in frontend ChatPage P2 fixes: - Replace monkey-patching on CompiledStateGraph with typed GraphContext - Replace 9-param dispatch_message with WebSocketContext dataclass - Extract duplicate _envelope() into shared app/api_utils.py - Replace mutable module-level counter with crypto.randomUUID() - Remove hardcoded mock data from ReviewPage, use api.ts wrappers - Remove `as any` type escape from ReplayPage All 516 tests passing, 0 TypeScript errors.
95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
"""Tests for app.graph module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from langgraph.checkpoint.memory import InMemorySaver
|
|
|
|
from app.graph import build_agent_nodes, build_graph
|
|
from app.graph_context import GraphContext
|
|
from app.intent import ClassificationResult, IntentTarget
|
|
|
|
if TYPE_CHECKING:
|
|
from app.registry import AgentRegistry
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestBuildAgentNodes:
|
|
def test_creates_correct_number_of_nodes(self, sample_registry: AgentRegistry) -> None:
|
|
mock_llm = MagicMock()
|
|
nodes = build_agent_nodes(sample_registry, mock_llm)
|
|
assert len(nodes) == 3
|
|
|
|
def test_nodes_are_runnable(self, sample_registry: AgentRegistry) -> None:
|
|
mock_llm = MagicMock()
|
|
nodes = build_agent_nodes(sample_registry, mock_llm)
|
|
for node in nodes:
|
|
assert hasattr(node, "invoke") or hasattr(node, "ainvoke")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestBuildGraph:
|
|
def test_graph_compiles_with_mock_checkpointer(self, sample_registry: AgentRegistry) -> None:
|
|
mock_llm = MagicMock()
|
|
mock_llm.bind_tools = MagicMock(return_value=mock_llm)
|
|
mock_llm.with_structured_output = MagicMock(return_value=mock_llm)
|
|
checkpointer = InMemorySaver()
|
|
|
|
graph_ctx = build_graph(sample_registry, mock_llm, checkpointer)
|
|
assert graph_ctx is not None
|
|
assert graph_ctx.graph is not None
|
|
|
|
def test_graph_has_classifier_attached(self, sample_registry: AgentRegistry) -> None:
|
|
mock_llm = MagicMock()
|
|
mock_llm.bind_tools = MagicMock(return_value=mock_llm)
|
|
mock_llm.with_structured_output = MagicMock(return_value=mock_llm)
|
|
checkpointer = InMemorySaver()
|
|
mock_classifier = MagicMock()
|
|
|
|
graph_ctx = build_graph(
|
|
sample_registry, mock_llm, checkpointer, intent_classifier=mock_classifier
|
|
)
|
|
assert graph_ctx.intent_classifier is mock_classifier
|
|
assert graph_ctx.registry is sample_registry
|
|
|
|
def test_graph_without_classifier(self, sample_registry: AgentRegistry) -> None:
|
|
mock_llm = MagicMock()
|
|
mock_llm.bind_tools = MagicMock(return_value=mock_llm)
|
|
mock_llm.with_structured_output = MagicMock(return_value=mock_llm)
|
|
checkpointer = InMemorySaver()
|
|
|
|
graph_ctx = build_graph(sample_registry, mock_llm, checkpointer)
|
|
assert graph_ctx.intent_classifier is None
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestClassifyIntent:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_without_classifier(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list_agents = MagicMock(return_value=())
|
|
graph_ctx = GraphContext(graph=MagicMock(), registry=mock_registry, intent_classifier=None)
|
|
result = await graph_ctx.classify_intent("hello")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_calls_classifier(self) -> None:
|
|
expected = ClassificationResult(
|
|
intents=(IntentTarget(agent_name="order_lookup", confidence=0.9, reasoning="test"),),
|
|
)
|
|
mock_classifier = AsyncMock()
|
|
mock_classifier.classify = AsyncMock(return_value=expected)
|
|
|
|
mock_registry = MagicMock()
|
|
mock_registry.list_agents = MagicMock(return_value=())
|
|
graph_ctx = GraphContext(
|
|
graph=MagicMock(), registry=mock_registry, intent_classifier=mock_classifier,
|
|
)
|
|
|
|
result = await graph_ctx.classify_intent("check order")
|
|
assert result is not None
|
|
assert result.intents[0].agent_name == "order_lookup"
|