Agent Zero
122 lines
5.3 KiB
Python
122 lines
5.3 KiB
Python
"""Tests for Phase I — Integration tools: I-AW, I-AK."""
|
|
from __future__ import annotations
|
|
import uuid
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestIntegrationTools:
|
|
"""Test the integration tools module (I-AW, I-AK)."""
|
|
|
|
def test_all_tools_importable(self):
|
|
"""All integration tools are importable."""
|
|
from app.ai.integration_tools import (
|
|
start_workflow_tool,
|
|
check_workflow_status_tool,
|
|
ask_knowledge_tool,
|
|
search_knowledge_tool,
|
|
register_integration_tools,
|
|
)
|
|
assert callable(start_workflow_tool)
|
|
assert callable(check_workflow_status_tool)
|
|
assert callable(ask_knowledge_tool)
|
|
assert callable(search_knowledge_tool)
|
|
assert callable(register_integration_tools)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_workflow_tool_returns_result(self):
|
|
"""start_workflow_tool returns instance info on success."""
|
|
from app.ai.integration_tools import start_workflow_tool
|
|
|
|
with patch("app.services.workflow_service.create_instance", new_callable=AsyncMock) as mock_create:
|
|
mock_create.return_value = {"id": "inst-123", "status": "pending"}
|
|
result = await start_workflow_tool(
|
|
db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
workflow_id="wf-123", context={"key": "value"},
|
|
)
|
|
assert result["instance_id"] == "inst-123"
|
|
assert result["status"] == "pending"
|
|
assert result["workflow_id"] == "wf-123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_workflow_tool_handles_not_found(self):
|
|
"""start_workflow_tool returns error when workflow not found."""
|
|
from app.ai.integration_tools import start_workflow_tool
|
|
|
|
with patch("app.services.workflow_service.create_instance", new_callable=AsyncMock) as mock_create:
|
|
mock_create.return_value = None
|
|
result = await start_workflow_tool(
|
|
db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
workflow_id="nonexistent",
|
|
)
|
|
assert result["error"] == "Workflow not found"
|
|
assert result["status"] == "not_found"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ask_knowledge_tool_delegates_to_ask_knowledge(self):
|
|
"""ask_knowledge_tool delegates to knowledge_lifecycle.ask_knowledge."""
|
|
from app.ai.integration_tools import ask_knowledge_tool
|
|
|
|
with patch("app.ai.knowledge_lifecycle.ask_knowledge", new_callable=AsyncMock) as mock_ask:
|
|
mock_ask.return_value = {"answer": "Test answer", "evidence": [], "query": "test"}
|
|
result = await ask_knowledge_tool(
|
|
db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
query="test query",
|
|
)
|
|
assert result["answer"] == "Test answer"
|
|
mock_ask.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_knowledge_tool_handles_no_search(self):
|
|
"""search_knowledge_tool returns error when search not available."""
|
|
from app.ai.integration_tools import search_knowledge_tool
|
|
|
|
with patch("app.plugins.builtins.unified_search.contracts.UnifiedSearchContract") as mock_contract:
|
|
mock_contract.get_function = MagicMock(return_value=None)
|
|
result = await search_knowledge_tool(
|
|
db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
|
|
query="test",
|
|
)
|
|
assert result["error"] == "Search not available"
|
|
assert result["results"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_workflow_status_tool_handles_not_found(self):
|
|
"""check_workflow_status_tool returns error when instance not found."""
|
|
from app.ai.integration_tools import check_workflow_status_tool
|
|
|
|
with patch("app.services.workflow_service.get_instance", new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = None
|
|
result = await check_workflow_status_tool(
|
|
db=MagicMock(), tenant_id=uuid.uuid4(),
|
|
instance_id=str(uuid.uuid4()),
|
|
)
|
|
assert result["error"] == "Instance not found"
|
|
assert result["status"] == "not_found"
|
|
|
|
|
|
# ─── I-APPR-LOOP: Agent Loop Human-in-the-Loop Approval ──────────────────────
|
|
|
|
|
|
class TestAgentLoopApproval:
|
|
"""Test the I-APPR-LOOP approval integration in run_react_loop."""
|
|
|
|
def test_run_react_loop_has_approval_params(self):
|
|
"""run_react_loop has require_approval and approval_tools parameters."""
|
|
import inspect
|
|
from app.ai.agent_loop import run_react_loop
|
|
|
|
sig = inspect.signature(run_react_loop)
|
|
assert "require_approval" in sig.parameters
|
|
assert "approval_tools" in sig.parameters
|
|
assert sig.parameters["require_approval"].default is False
|
|
assert sig.parameters["approval_tools"].default is None
|
|
|
|
def test_react_result_has_waiting_for_approval_status(self):
|
|
"""ReActResult supports waiting_for_approval status."""
|
|
from app.ai.agent_loop import ReActResult
|
|
|
|
result = ReActResult(final_content="", status="waiting_for_approval")
|
|
assert result.status == "waiting_for_approval"
|
|
assert result.final_content == ""
|