72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
import pytest
|
|
|
|
from app.ai.tool_registry import ToolRegistry
|
|
from app.domain.schemas import ActionStatus
|
|
from app.domain.schemas import ToolDefinition
|
|
from app.executor.action_executor import ActionExecutor
|
|
from app.infrastructure.database import PendingAction
|
|
from app.memory.memory_manager import MemoryManager
|
|
|
|
|
|
def test_memory_retrieves_semantic_items():
|
|
memory = MemoryManager()
|
|
memory.upsert("preference", "Customer prefers white flowers")
|
|
|
|
results = memory.retrieve("white flower preference")
|
|
|
|
assert results[0]["type"] == "preference"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_rejects_unapproved_action():
|
|
registry = ToolRegistry(
|
|
[
|
|
ToolDefinition(
|
|
name="create_task",
|
|
description="Create task",
|
|
method="POST",
|
|
path="/tasks",
|
|
schema={"type": "object", "properties": {"title": {"type": "string"}}},
|
|
)
|
|
]
|
|
)
|
|
action = PendingAction(tool="create_task", status="pending_review", payload={"title": "x"})
|
|
|
|
with pytest.raises(ValueError, match="Only approved"):
|
|
await ActionExecutor(registry).execute(None, action) # type: ignore[arg-type]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_generates_call_without_http_request():
|
|
registry = ToolRegistry(
|
|
[
|
|
ToolDefinition(
|
|
name="update_reservation",
|
|
description="Update reservation",
|
|
method="PATCH",
|
|
path="/reservations/{reservation_id}",
|
|
path_params=["reservation_id"],
|
|
schema={"type": "object", "properties": {"reservation_id": {"type": "string"}, "status": {"type": "string"}}},
|
|
)
|
|
]
|
|
)
|
|
action = PendingAction(tool="update_reservation", status=ActionStatus.APPROVED, payload={"reservation_id": "abc", "status": "ACCEPTED"})
|
|
|
|
class FakeSession:
|
|
def add(self, _action):
|
|
return None
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def refresh(self, _action):
|
|
return None
|
|
|
|
result = await ActionExecutor(registry).execute(FakeSession(), action) # type: ignore[arg-type]
|
|
|
|
assert result.status == ActionStatus.APPROVED
|
|
assert result.execution_result["generated_only"] is True
|
|
assert result.execution_result["method"] == "PATCH"
|
|
assert result.execution_result["path"] == "/reservations/abc"
|
|
assert result.execution_result["json"] == {"status": "ACCEPTED"}
|