489 lines
14 KiB
Python
489 lines
14 KiB
Python
import pytest
|
|
from unittest.mock import Mock
|
|
from src.naliiabot.bot.agent.agent import Agent
|
|
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
|
|
from langgraph.graph import END
|
|
|
|
|
|
"""
|
|
Tests unitarios robustos para el Agente.
|
|
|
|
Estrategia de testing:
|
|
- Tests de construcción (verificar grafo correcto)
|
|
- Tests de comportamiento (nodos individuales)
|
|
- Tests de integración (flujo completo con mocks)
|
|
- Tests de edge cases (errores, límites)
|
|
"""
|
|
|
|
# ============================================================================
|
|
# TESTS DE CONSTRUCCION
|
|
# ============================================================================
|
|
|
|
|
|
class TestAgentConstruction:
|
|
"""Verificar inicialización correcta del agente."""
|
|
|
|
def test_agent_initializes_with_defaults(self, mock_model):
|
|
"""
|
|
DADO: Modelo válido\n
|
|
CUANDO: Se crea agente sin config\n
|
|
ENTONCES: Usa configuración por defecto
|
|
"""
|
|
|
|
agent = Agent(model=mock_model)
|
|
|
|
assert agent.config.max_iterations == 10
|
|
assert "helpful assistant" in agent.config.system_prompt
|
|
assert agent.tools == []
|
|
|
|
def test_agent_accepts_custom_config(self, mock_model, default_config):
|
|
"""
|
|
DADO: Configuración personalizada\n
|
|
CUANDO: Se crea agente\n
|
|
ENTONCES: Aplica configuración correctamente
|
|
"""
|
|
|
|
agent = Agent(model=mock_model, config=default_config)
|
|
|
|
assert agent.config.max_iterations == 5
|
|
assert agent.config.system_prompt == "Test prompt"
|
|
|
|
def test_agent_registers_tools(self, mock_model, mock_tool):
|
|
"""
|
|
DADO: Lista de herramientas\n
|
|
CUANDO: Se crea agente\n
|
|
ENTONCES: Registra herramientas correctamente
|
|
"""
|
|
|
|
agent = Agent(model=mock_model, tools=[mock_tool])
|
|
|
|
assert len(agent.tools) == 1
|
|
assert agent.tools[0].name == "calculator"
|
|
|
|
def test_tools_list_is_immutable(self, mock_model, mock_tool):
|
|
"""
|
|
DADO: Agente con herramientas\n
|
|
CUANDO: Se modifica lista retornada\n
|
|
ENTONCES: No afecta estado interno
|
|
"""
|
|
|
|
agent = Agent(model=mock_model, tools=[mock_tool])
|
|
tools = agent.tools
|
|
tools.clear()
|
|
|
|
assert len(agent.tools) == 1
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE ESTRUCTURA DEL GRAFO
|
|
# ============================================================================
|
|
|
|
|
|
class TestGraphStructure:
|
|
"""Verificar que el grafo está construido correctamente."""
|
|
|
|
def test_graph_has_required_nodes(self, agent):
|
|
"""
|
|
DADO: Agente construido\n
|
|
CUANDO: Inspecciono grafo\n
|
|
ENTONCES: Tiene nodos llm_call y tool_node
|
|
"""
|
|
|
|
graph = agent.graph
|
|
|
|
# Verificar nodos existen (langgraph no expone directamente,
|
|
# pero podemos verificar que compile funciona)
|
|
compiled = graph.compile()
|
|
assert compiled is not None
|
|
|
|
def test_graph_compiles_successfully(self, agent):
|
|
"""
|
|
DADO: Agente construido\n
|
|
CUANDO: Compilo grafo\n
|
|
ENTONCES: No lanza excepciones
|
|
"""
|
|
|
|
graph = agent._build_agent()
|
|
compiled = graph.compile()
|
|
|
|
assert compiled is not None
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE COMPORTAMIENTO DE NODOS
|
|
# ============================================================================
|
|
|
|
class TestLLMNode:
|
|
"""Tests del nodo de llamada al LLM."""
|
|
|
|
def test_llm_node_increments_counter(self, agent):
|
|
"""
|
|
DADO: Estado con 2 llamadas previas\n
|
|
CUANDO: Ejecuto nodo LLM\n
|
|
ENTONCES: Contador aumenta a 3
|
|
"""
|
|
|
|
mock_response = AIMessage(content="Hola")
|
|
agent._model.invoke = Mock(return_value=mock_response)
|
|
|
|
state = {
|
|
"messages": [HumanMessage(content="Hi")],
|
|
"llm_calls": 2
|
|
}
|
|
|
|
result = agent._llm_call_node(state)
|
|
|
|
assert result["llm_calls"] == 3
|
|
assert len(result["messages"]) == 2 # Original + respuesta
|
|
|
|
def test_llm_node_prepends_system_message(self, agent):
|
|
"""
|
|
DADO: Mensajes de usuario\n
|
|
CUANDO: Ejecuto nodo LLM\n
|
|
ENTONCES: Prepende system message
|
|
"""
|
|
|
|
mock_response = AIMessage(content="Response")
|
|
agent._model.invoke = Mock(return_value=mock_response)
|
|
|
|
state = {"messages": [HumanMessage(content="Hi")]}
|
|
agent._llm_call_node(state)
|
|
|
|
# Verificar que se llamó al modelo con system message
|
|
call_args = agent._model.invoke.call_args[0][0]
|
|
assert call_args[0].type == "system"
|
|
assert "Test prompt" in call_args[0].content
|
|
|
|
def test_llm_node_respects_max_iterations(self, agent):
|
|
"""
|
|
DADO: Límite de 5 iteraciones alcanzado\n
|
|
CUANDO: Ejecuto nodo LLM\n
|
|
ENTONCES: Lanza RuntimeError
|
|
"""
|
|
|
|
mock_response = AIMessage(content="Hi")
|
|
agent._model.invoke = Mock(return_value=mock_response)
|
|
|
|
state = {
|
|
"messages": [HumanMessage(content="Hi")],
|
|
"llm_calls": 5 # Ya en límite
|
|
}
|
|
|
|
with pytest.raises(RuntimeError, match="Max iterations"):
|
|
agent._llm_call_node(state)
|
|
|
|
def test_llm_node_calls_pre_hook(self, agent):
|
|
"""
|
|
DADO: Hook registrado\n
|
|
CUANDO: Ejecuto nodo LLM\n
|
|
ENTONCES: Hook es llamado con estado
|
|
"""
|
|
|
|
mock_hook = Mock()
|
|
agent.set_pre_llm_hook(mock_hook)
|
|
|
|
mock_response = AIMessage(content="Hi")
|
|
agent._model.invoke = Mock(return_value=mock_response)
|
|
|
|
state = {"messages": [HumanMessage(content="Hi")]}
|
|
agent._llm_call_node(state)
|
|
|
|
mock_hook.assert_called_once_with(state)
|
|
|
|
|
|
class TestToolNode:
|
|
"""Tests del nodo de ejecución de herramientas."""
|
|
|
|
def test_tool_node_executes_single_tool(self, agent_with_tool):
|
|
"""
|
|
DADO: Mensaje con tool_call\n
|
|
CUANDO: Ejecuto nodo tools\n
|
|
ENTONCES: Ejecuta herramienta y retorna resultado
|
|
"""
|
|
|
|
ai_msg = AIMessage(
|
|
content="",
|
|
tool_calls=[{
|
|
"name": "calculator",
|
|
"args": {"x": 1, "y": 2},
|
|
"id": "call_123"
|
|
}]
|
|
)
|
|
state = {"messages": [HumanMessage(content="Calc"), ai_msg]}
|
|
|
|
result = agent_with_tool._tool_node(state)
|
|
|
|
assert len(result["messages"]) == 3 # + ToolMessage
|
|
tool_msg = result["messages"][-1]
|
|
assert tool_msg.type == "tool"
|
|
assert tool_msg.content == "42"
|
|
|
|
def test_tool_node_handles_tool_not_found(self, agent):
|
|
"""
|
|
DADO: Tool call a herramienta inexistente\n
|
|
CUANDO: Ejecuto nodo\n
|
|
ENTONCES: Retorna mensaje de error
|
|
"""
|
|
|
|
ai_msg = AIMessage(
|
|
content="",
|
|
tool_calls=[{
|
|
"name": "unknown_tool",
|
|
"args": {},
|
|
"id": "call_123"
|
|
}]
|
|
)
|
|
state = {"messages": [ai_msg]}
|
|
|
|
result = agent._tool_node(state)
|
|
|
|
tool_msg = result["messages"][-1]
|
|
assert "not found" in tool_msg.content
|
|
assert tool_msg.status == "success"
|
|
|
|
def test_tool_node_handles_execution_error(self, agent_with_tool):
|
|
"""
|
|
DADO: Herramienta que lanza excepción\n
|
|
CUANDO: Ejecuto nodo\n
|
|
ENTONCES: Retorna mensaje de error sin propagar
|
|
"""
|
|
|
|
agent_with_tool._tools_by_name["calculator"].invoke = Mock(
|
|
side_effect=ValueError("Invalid input")
|
|
)
|
|
|
|
ai_msg = AIMessage(
|
|
content="",
|
|
tool_calls=[{"name": "calculator", "args": {}, "id": "call_123"}]
|
|
)
|
|
state = {"messages": [ai_msg]}
|
|
|
|
result = agent_with_tool._tool_node(state)
|
|
|
|
tool_msg = result["messages"][-1]
|
|
assert "Error executing tool" in tool_msg.content
|
|
assert tool_msg.status == "success"
|
|
|
|
def test_tool_node_calls_post_hook(self, agent_with_tool):
|
|
"""
|
|
DADO: Hook post-tool registrado\n
|
|
CUANDO: Ejecuto nodo\n
|
|
ENTONCES: Hook es llamado por cada tool
|
|
"""
|
|
|
|
mock_hook = Mock()
|
|
agent_with_tool.set_post_tool_hook(mock_hook)
|
|
|
|
ai_msg = AIMessage(
|
|
content="",
|
|
tool_calls=[{
|
|
"name": "calculator",
|
|
"args": {},
|
|
"id": "call_123"
|
|
}]
|
|
)
|
|
state = {"messages": [ai_msg]}
|
|
|
|
agent_with_tool._tool_node(state)
|
|
|
|
assert mock_hook.call_count == 1
|
|
|
|
def test_tool_node_noop_without_tool_calls(self, agent):
|
|
"""
|
|
DADO: Mensaje sin tool_calls\n
|
|
CUANDO: Ejecuto nodo\n
|
|
ENTONCES: Retorna estado sin cambios
|
|
"""
|
|
|
|
ai_msg = AIMessage(content="Just chatting")
|
|
state = {"messages": [ai_msg]}
|
|
|
|
result = agent._tool_node(state)
|
|
|
|
assert result["messages"] == state["messages"]
|
|
|
|
|
|
class TestShouldContinue:
|
|
"""Tests de la lógica de decisión del grafo."""
|
|
|
|
def test_continue_to_tools_when_tool_calls_present(self, agent):
|
|
"""
|
|
DADO: Mensaje con tool_calls\n
|
|
CUANDO: Evalúo should_continue\n
|
|
ENTONCES: Retorna 'tool_node'
|
|
"""
|
|
ai_msg = AIMessage(
|
|
content="",
|
|
tool_calls=[{"name": "calc", "args": {}, "id": "1"}]
|
|
)
|
|
state = {"messages": [ai_msg]}
|
|
|
|
decision = agent._should_continue(state)
|
|
|
|
assert decision == "tool_node"
|
|
|
|
def test_end_when_no_tool_calls(self, agent):
|
|
"""
|
|
DADO: Mensaje sin tool_calls\n
|
|
CUANDO: Evalúo should_continue\n
|
|
ENTONCES: Retorna END
|
|
"""
|
|
|
|
ai_msg = AIMessage(content="Final answer")
|
|
state = {"messages": [ai_msg]}
|
|
|
|
decision = agent._should_continue(state)
|
|
|
|
assert decision == END
|
|
|
|
def test_end_on_empty_messages(self, agent):
|
|
"""
|
|
DADO: Estado sin mensajes\n
|
|
CUANDO: Evalúo should_continue\n
|
|
ENTONCES: Retorna END
|
|
"""
|
|
|
|
state = {"messages": []}
|
|
|
|
decision = agent._should_continue(state)
|
|
|
|
assert decision == END
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE INTEGRACIÓN CON FLUJO COMPLETO
|
|
# ============================================================================
|
|
|
|
class TestIntegrationFlows:
|
|
"""Tests de flujos completos con mocks controlados."""
|
|
|
|
def test_simple_conversation_flow_no_tools(self, agent):
|
|
"""
|
|
DADO: Agente sin herramientas
|
|
CUANDO: Usuario pregunta algo simple
|
|
ENTONCES: Flujo: START → llm_call → END
|
|
"""
|
|
final_response = AIMessage(content="¡Hola! ¿En qué puedo ayudarte?")
|
|
agent._model.invoke = Mock(return_value=final_response)
|
|
|
|
initial_state = {
|
|
"messages": [HumanMessage(content="Hola")]
|
|
}
|
|
|
|
result = agent.invoke(initial_state)
|
|
|
|
assert "messages" in result
|
|
assert "llm_calls" in result
|
|
assert result["llm_calls"] == 1
|
|
assert len(result["messages"]) == 2
|
|
assert result["messages"][-1].content == (
|
|
"¡Hola! ¿En qué puedo ayudarte?")
|
|
|
|
def test_tool_use_flow(self, agent_with_tool):
|
|
"""
|
|
DADO: Agente con calculadora
|
|
CUANDO: Usuario pide cálculo
|
|
ENTONCES: Flujo: START → llm_call → tool_node → llm_call → END
|
|
"""
|
|
# Primera llamada: LLM decide usar tool
|
|
first_response = AIMessage(
|
|
content="",
|
|
tool_calls=[{
|
|
"name": "calculator",
|
|
"args": {"expr": "2+2"},
|
|
"id": "calc_1"
|
|
}]
|
|
)
|
|
# Segunda llamada: LLM responde con resultado
|
|
second_response = AIMessage(content="El resultado es 42")
|
|
|
|
agent_with_tool._model.invoke = Mock(side_effect=[
|
|
first_response,
|
|
second_response
|
|
])
|
|
|
|
initial_state = {
|
|
"messages": [HumanMessage(content="Calcula 2+2")]
|
|
}
|
|
|
|
result = agent_with_tool.invoke(initial_state)
|
|
|
|
# Verificar que se hicieron 2 llamadas al LLM
|
|
assert result["llm_calls"] == 2
|
|
# Verificar flujo: Human → AI(tool) → Tool → AI(final)
|
|
assert len(result["messages"]) == 4
|
|
assert result["messages"][-1].content == "El resultado es 42"
|
|
|
|
def test_validation_rejects_missing_messages(self, agent):
|
|
"""
|
|
DADO: Estado sin 'messages'\n
|
|
CUANDO: Invoco agente\n
|
|
ENTONCES: Lanza ValueError
|
|
"""
|
|
|
|
invalid_state = {"llm_calls": 0}
|
|
|
|
with pytest.raises(ValueError, match="messages"):
|
|
agent.invoke(invalid_state)
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE EJECUCIÓN DE TOOLS (UNITARIO DETALLADO)
|
|
# ============================================================================
|
|
|
|
class TestToolExecution:
|
|
"""Tests específicos del método _execute_tool."""
|
|
|
|
def test_execute_tool_success(self, agent_with_tool):
|
|
"""
|
|
DADO: Tool call válido
|
|
CUANDO: Ejecuto _execute_tool
|
|
ENTONCES: Retorna ToolMessage exitoso
|
|
"""
|
|
tool_call = {
|
|
"name": "calculator",
|
|
"args": {"x": 10},
|
|
"id": "call_123"
|
|
}
|
|
|
|
result = agent_with_tool._execute_tool(tool_call)
|
|
|
|
assert isinstance(result, ToolMessage)
|
|
assert result.content == "42"
|
|
assert result.tool_call_id == "call_123"
|
|
assert result.status == "success"
|
|
|
|
def test_execute_tool_not_found(self, agent):
|
|
"""
|
|
DADO: Tool call a herramienta inexistente
|
|
CUANDO: Ejecuto _execute_tool
|
|
ENTONCES: Retorna ToolMessage con error
|
|
"""
|
|
tool_call = {
|
|
"name": "nonexistent",
|
|
"args": {},
|
|
"id": "call_404"
|
|
}
|
|
|
|
result = agent._execute_tool(tool_call)
|
|
|
|
assert result.status == "success"
|
|
assert "nonexistent" in result.content
|
|
assert result.tool_call_id == "call_404"
|
|
|
|
def test_execute_tool_exception_handling(self, agent_with_tool):
|
|
"""
|
|
DADO: Tool que lanza excepción
|
|
CUANDO: Ejecuto _execute_tool
|
|
ENTONCES: Captura error y retorna ToolMessage
|
|
"""
|
|
agent_with_tool._tools_by_name["calculator"].invoke = Mock(
|
|
side_effect=Exception("DB Connection failed")
|
|
)
|
|
|
|
tool_call = {"name": "calculator", "args": {}, "id": "call_err"}
|
|
result = agent_with_tool._execute_tool(tool_call)
|
|
|
|
assert result.status == "success"
|
|
assert "DB Connection failed" in result.content
|