28 lines
1.2 KiB
Python
28 lines
1.2 KiB
Python
|
import pytest
|
||
|
|
|
||
|
|
from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
|
||
|
|
from app.service.agent.base import BaseAgent
|
||
|
|
from app.service.agent.executor import AgentExecutor
|
||
|
|
from app.service.agent.factory import AgentFactory
|
||
|
|
|
||
|
|
|
||
|
|
class DemoAgent(BaseAgent):
|
||
|
|
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
||
|
|
return CoreResult(text=request.message)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_executor_uses_factory_and_emits_terminal_event(governance) -> None:
|
||
|
|
definition = AgentDefinition(agent_type="demo", version="1",
|
||
|
|
allowed_roles=("customer",), allowed_portals=("api",))
|
||
|
|
factory = AgentFactory(governance)
|
||
|
|
factory.register(definition, lambda _: DemoAgent(definition))
|
||
|
|
request = AgentRequest(
|
||
|
|
agent_type="demo", message="hello", session_id="s1", idempotency_key="key-123456789012"
|
||
|
|
)
|
||
|
|
events = [event async for event in AgentExecutor(factory).execute(
|
||
|
|
"demo", request, RequestContext(user_id="1", trace_id="t1",
|
||
|
|
roles=("customer",), permissions=("agent:run",)), "run-1"
|
||
|
|
)]
|
||
|
|
assert [event.event_type for event in events] == ["start", "done"]
|