476 lines
14 KiB
Python
476 lines
14 KiB
Python
import pytest
|
|
from unittest.mock import Mock, MagicMock
|
|
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
|
|
from src.naliiabot.bot.agent.agent import Agent, AgentConfig
|
|
|
|
|
|
"""
|
|
Tests robustos para las tools/herramientas de la clase Agent.
|
|
|
|
Estrategia:
|
|
- Tests de ejecución de herramientas individuales
|
|
- Tests de manejo de errores
|
|
- Tests del nodo de tools
|
|
- Tests de integración con el flujo del agente
|
|
"""
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE EJECUCION DE HERRAMIENTAS INDIVIDUALES
|
|
# ============================================================================
|
|
|
|
|
|
class TestExecuteTool:
|
|
"""Pruebas para la ejecución de herramientas individuales."""
|
|
|
|
def test_execute_tool_success(self, agent_with_tool):
|
|
"""
|
|
DADO: Una herramienta registrada en el agente
|
|
CUANDO: Se ejecuta exitosamente
|
|
ENTONCES: Retorna ToolMessage con status 'success'
|
|
"""
|
|
tool_call = {
|
|
"name": "calculator",
|
|
"id": "tool_123",
|
|
"args": {"x": 5, "y": 7}
|
|
}
|
|
|
|
result = agent_with_tool._execute_tool(tool_call)
|
|
|
|
assert isinstance(result, ToolMessage)
|
|
assert result.status == "success"
|
|
assert result.tool_call_id == "tool_123"
|
|
assert result.content == "42"
|
|
|
|
def test_execute_tool_not_found(self, agent_with_tool):
|
|
"""
|
|
DADO: Una herramienta que no existe
|
|
CUANDO: Se intenta ejecutar
|
|
ENTONCES: Retorna ToolMessage con error
|
|
"""
|
|
tool_call = {
|
|
"name": "nonexistent_tool",
|
|
"id": "tool_456",
|
|
"args": {}
|
|
}
|
|
|
|
result = agent_with_tool._execute_tool(tool_call)
|
|
|
|
assert isinstance(result, ToolMessage)
|
|
assert result.status == "error"
|
|
assert "not found" in result.content.lower()
|
|
|
|
def test_execute_tool_with_exception(self, mock_model, default_config):
|
|
"""
|
|
DADO: Una herramienta que lanza excepción
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Captura el error y retorna ToolMessage con estado error
|
|
"""
|
|
failing_tool = Mock()
|
|
failing_tool.name = "failing_tool"
|
|
failing_tool.invoke = Mock(side_effect=ValueError("Tool error"))
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[failing_tool]
|
|
)
|
|
|
|
tool_call = {
|
|
"name": "failing_tool",
|
|
"id": "tool_789",
|
|
"args": {"param": "value"}
|
|
}
|
|
|
|
result = agent._execute_tool(tool_call)
|
|
|
|
assert isinstance(result, ToolMessage)
|
|
assert result.status == "error"
|
|
assert "Tool error" in result.content
|
|
|
|
def test_execute_tool_with_no_args(self, agent_with_tool):
|
|
"""
|
|
DADO: Una herramienta sin argumentos en tool_call
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Usa dict vacío como argumentos
|
|
"""
|
|
tool_call = {
|
|
"name": "calculator",
|
|
"id": "tool_abc"
|
|
# Sin 'args'
|
|
}
|
|
|
|
result = agent_with_tool._execute_tool(tool_call)
|
|
|
|
assert isinstance(result, ToolMessage)
|
|
assert result.tool_call_id == "tool_abc"
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DEL NODO DE HERRAMIENTAS (TOOL NODE)
|
|
# ============================================================================
|
|
|
|
|
|
class TestToolNode:
|
|
"""Pruebas para el nodo que ejecuta herramientas."""
|
|
|
|
def test_tool_node_with_tool_calls(self, agent_with_tool):
|
|
"""
|
|
DADO: Un estado con mensaje AIMessage que contiene tool_calls
|
|
CUANDO: Se ejecuta el tool_node
|
|
ENTONCES: Retorna estado actualizado con ToolMessages
|
|
"""
|
|
# Mock del modelo para generar tool_call
|
|
ai_message = AIMessage(
|
|
content="Using calculator",
|
|
tool_calls=[
|
|
{
|
|
"name": "calculator",
|
|
"id": "call_1",
|
|
"args": {"x": 10}
|
|
}
|
|
]
|
|
)
|
|
|
|
state = {
|
|
"messages": [
|
|
HumanMessage(content="Calculate"),
|
|
ai_message
|
|
]
|
|
}
|
|
|
|
result = agent_with_tool._tool_node(state)
|
|
|
|
assert "messages" in result
|
|
assert len(result["messages"]) == 3 # original 2 + 1 ToolMessage
|
|
assert isinstance(result["messages"][-1], ToolMessage)
|
|
|
|
def test_tool_node_without_tool_calls(self, agent):
|
|
"""
|
|
DADO: Un estado con mensaje sin tool_calls
|
|
CUANDO: Se ejecuta el tool_node
|
|
ENTONCES: Retorna estado sin cambios
|
|
"""
|
|
state = {
|
|
"messages": [
|
|
HumanMessage(content="Hello"),
|
|
AIMessage(content="Hi there!")
|
|
]
|
|
}
|
|
|
|
result = agent._tool_node(state)
|
|
|
|
assert result["messages"] == state["messages"]
|
|
|
|
def test_tool_node_multiple_tool_calls(self, mock_model, default_config):
|
|
"""
|
|
DADO: Un AIMessage con múltiples tool_calls
|
|
CUANDO: Se ejecuta el tool_node
|
|
ENTONCES: Ejecuta todas las herramientas y retorna múltiples ToolMessages
|
|
"""
|
|
tool1 = Mock()
|
|
tool1.name = "tool_a"
|
|
tool1.invoke = Mock(return_value="Result A")
|
|
|
|
tool2 = Mock()
|
|
tool2.name = "tool_b"
|
|
tool2.invoke = Mock(return_value="Result B")
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[tool1, tool2]
|
|
)
|
|
|
|
ai_message = AIMessage(
|
|
content="Using tools",
|
|
tool_calls=[
|
|
{"name": "tool_a", "id": "call_1", "args": {}},
|
|
{"name": "tool_b", "id": "call_2", "args": {}}
|
|
]
|
|
)
|
|
|
|
state = {
|
|
"messages": [ai_message]
|
|
}
|
|
|
|
result = agent._tool_node(state)
|
|
|
|
# 1 AIMessage + 2 ToolMessages
|
|
assert len(result["messages"]) == 3
|
|
assert all(isinstance(msg, (AIMessage, ToolMessage)) for msg in result["messages"])
|
|
|
|
def test_tool_node_post_hook_called(self, agent_with_tool):
|
|
"""
|
|
DADO: Un agente con post_tool_hook registrado
|
|
CUANDO: Se ejecuta una herramienta
|
|
ENTONCES: Se invoca el hook después de cada ejecución
|
|
"""
|
|
hook = Mock()
|
|
agent_with_tool.set_post_tool_hook(hook)
|
|
|
|
ai_message = AIMessage(
|
|
content="Test",
|
|
tool_calls=[
|
|
{
|
|
"name": "calculator",
|
|
"id": "call_1",
|
|
"args": {}
|
|
}
|
|
]
|
|
)
|
|
|
|
state = {
|
|
"messages": [ai_message]
|
|
}
|
|
|
|
agent_with_tool._tool_node(state)
|
|
|
|
# Hook debe ser llamado una vez por herramienta
|
|
assert hook.call_count == 1
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE INTEGRACIÓN CON FLUJO COMPLETO
|
|
# ============================================================================
|
|
|
|
|
|
class TestToolsIntegration:
|
|
"""Pruebas de integración de tools en el flujo completo del agente."""
|
|
|
|
def test_agent_invokes_tool_in_flow(self, agent_with_tool):
|
|
"""
|
|
DADO: Un agente con herramientas configurado
|
|
CUANDO: Se invocan herramientas en la conversación
|
|
ENTONCES: El estado se actualiza con los resultados
|
|
"""
|
|
# Mock del modelo para retornar un AIMessage con tool_calls
|
|
ai_response = AIMessage(
|
|
content="I'll use the calculator",
|
|
tool_calls=[
|
|
{
|
|
"name": "calculator",
|
|
"id": "call_x",
|
|
"args": {"result": "42"}
|
|
}
|
|
]
|
|
)
|
|
agent_with_tool._model.bind_tools().invoke.return_value = ai_response
|
|
|
|
state = {
|
|
"messages": [HumanMessage(content="Calculate 2+2")]
|
|
}
|
|
|
|
result = agent_with_tool.invoke(state)
|
|
|
|
assert "messages" in result
|
|
# Debe contener original + AIMessage + ToolMessage
|
|
assert len(result["messages"]) >= 2
|
|
|
|
def test_tools_are_bound_to_model(self, agent_with_tool):
|
|
"""
|
|
DADO: Un agente con herramientas
|
|
CUANDO: Se invoca el LLM
|
|
ENTONCES: Las herramientas están vinculadas al modelo
|
|
"""
|
|
initial_state = {
|
|
"messages": [HumanMessage(content="Hi")]
|
|
}
|
|
|
|
try:
|
|
agent_with_tool.invoke(initial_state)
|
|
except:
|
|
# Es ok si falla por mocks incompletos
|
|
pass
|
|
|
|
# Verificar que bind_tools fue llamado
|
|
agent_with_tool._model.bind_tools.assert_called()
|
|
|
|
def test_tools_property_returns_copy(self, agent_with_tool):
|
|
"""
|
|
DADO: Un agente con herramientas
|
|
CUANDO: Se accede a la propiedad tools
|
|
ENTONCES: Retorna una copia de la lista de herramientas
|
|
"""
|
|
tools_copy = agent_with_tool.tools
|
|
|
|
assert len(tools_copy) == 1
|
|
assert tools_copy[0].name == "calculator"
|
|
|
|
# Modificar la copia no afecta al agente
|
|
tools_copy.clear()
|
|
assert len(agent_with_tool.tools) == 1
|
|
|
|
def test_multiple_tools_independently(self, mock_model, default_config):
|
|
"""
|
|
DADO: Un agente con múltiples herramientas diferentes
|
|
CUANDO: Se ejecutan herramientas
|
|
ENTONCES: Cada una se ejecuta de forma independiente
|
|
"""
|
|
weather_tool = Mock()
|
|
weather_tool.name = "get_weather"
|
|
weather_tool.invoke = Mock(return_value="Sunny 25°C")
|
|
|
|
time_tool = Mock()
|
|
time_tool.name = "get_time"
|
|
time_tool.invoke = Mock(return_value="14:30")
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[weather_tool, time_tool]
|
|
)
|
|
|
|
# Ejecutar weather_tool
|
|
result1 = agent._execute_tool({
|
|
"name": "get_weather",
|
|
"id": "w1",
|
|
"args": {}
|
|
})
|
|
assert "Sunny" in result1.content
|
|
|
|
# Ejecutar time_tool
|
|
result2 = agent._execute_tool({
|
|
"name": "get_time",
|
|
"id": "t1",
|
|
"args": {}
|
|
})
|
|
assert "14:30" in result2.content
|
|
|
|
|
|
# ============================================================================
|
|
# TESTS DE EDGE CASES
|
|
# ============================================================================
|
|
|
|
|
|
class TestToolsEdgeCases:
|
|
"""Pruebas de casos extremos y edge cases."""
|
|
|
|
def test_tool_returns_none(self, mock_model, default_config):
|
|
"""
|
|
DADO: Una herramienta que retorna None
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Se convierte a string adecuadamente
|
|
"""
|
|
tool = Mock()
|
|
tool.name = "none_tool"
|
|
tool.invoke = Mock(return_value=None)
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[tool]
|
|
)
|
|
|
|
result = agent._execute_tool({
|
|
"name": "none_tool",
|
|
"id": "n1",
|
|
"args": {}
|
|
})
|
|
|
|
assert result.content == "None"
|
|
assert result.status == "success"
|
|
|
|
def test_tool_returns_complex_object(self, mock_model, default_config):
|
|
"""
|
|
DADO: Una herramienta que retorna objeto complejo
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Se serializa a string correctamente
|
|
"""
|
|
tool = Mock()
|
|
tool.name = "complex_tool"
|
|
tool.invoke = Mock(return_value={"key": "value", "nested": {"data": 123}})
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[tool]
|
|
)
|
|
|
|
result = agent._execute_tool({
|
|
"name": "complex_tool",
|
|
"id": "c1",
|
|
"args": {}
|
|
})
|
|
|
|
assert isinstance(result.content, str)
|
|
assert "key" in result.content
|
|
assert "value" in result.content
|
|
|
|
def test_empty_tools_list(self, mock_model, default_config):
|
|
"""
|
|
DADO: Un agente sin herramientas
|
|
CUANDO: Se intenta ejecutar una herramienta
|
|
ENTONCES: Retorna error apropiado
|
|
"""
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[]
|
|
)
|
|
|
|
result = agent._execute_tool({
|
|
"name": "any_tool",
|
|
"id": "e1",
|
|
"args": {}
|
|
})
|
|
|
|
assert result.status == "error"
|
|
assert "not found" in result.content.lower()
|
|
|
|
def test_tool_with_special_characters_in_response(self, mock_model, default_config):
|
|
"""
|
|
DADO: Una herramienta que retorna strings especiales
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Se preservan los caracteres especiales
|
|
"""
|
|
tool = Mock()
|
|
tool.name = "special_tool"
|
|
tool.invoke = Mock(return_value="Response with special chars: ñ, é, 中文, 🚀")
|
|
|
|
agent = Agent(
|
|
model=mock_model,
|
|
config=default_config,
|
|
tools=[tool]
|
|
)
|
|
|
|
result = agent._execute_tool({
|
|
"name": "special_tool",
|
|
"id": "s1",
|
|
"args": {}
|
|
})
|
|
|
|
assert "ñ" in result.content
|
|
assert "中文" in result.content
|
|
|
|
def test_tool_call_missing_name(self, agent):
|
|
"""
|
|
DADO: Un tool_call sin campo 'name'
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Maneja el error gracefully
|
|
"""
|
|
result = agent._execute_tool({
|
|
"id": "m1",
|
|
"args": {}
|
|
# Sin 'name'
|
|
})
|
|
|
|
assert result.status == "error"
|
|
|
|
def test_tool_call_empty_args_dict(self, agent_with_tool):
|
|
"""
|
|
DADO: Un tool_call con args vacío
|
|
CUANDO: Se ejecuta
|
|
ENTONCES: Invoca la herramienta con dict vacío
|
|
"""
|
|
agent_with_tool._tools[0].invoke = Mock(return_value="Empty args result")
|
|
|
|
result = agent_with_tool._execute_tool({
|
|
"name": "calculator",
|
|
"id": "ea1",
|
|
"args": {}
|
|
})
|
|
|
|
agent_with_tool._tools[0].invoke.assert_called_with({})
|
|
assert result.status == "success"
|