| import pytest |
|
|
| from aegis.core import Agent, AgentContext, AgentResult, Task |
| from aegis.core.conductor import Conductor |
| from aegis.core.registry import AgentNotFoundError, AgentRegistry |
|
|
|
|
| class StubAgent(Agent): |
| def __init__( |
| self, |
| name: str, |
| *, |
| result_name: str | None = None, |
| ) -> None: |
| self.name = name |
| self._result_name = result_name or name |
|
|
| async def run(self, context: AgentContext) -> AgentResult: |
| return AgentResult( |
| agent_name=self._result_name, |
| success=True, |
| output={"task": context.task.name}, |
| ) |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_conductor_executes_registered_agent() -> None: |
| registry = AgentRegistry([StubAgent("planner")]) |
| conductor = Conductor(registry) |
| context = AgentContext( |
| task=Task( |
| name="Create plan", |
| description="Build a deterministic execution plan.", |
| ) |
| ) |
|
|
| result = await conductor.execute("planner", context) |
|
|
| assert result.success is True |
| assert result.output == {"task": "Create plan"} |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_conductor_rejects_missing_agent() -> None: |
| conductor = Conductor(AgentRegistry()) |
| context = AgentContext( |
| task=Task( |
| name="Missing agent", |
| description="Verify missing-agent behavior.", |
| ) |
| ) |
|
|
| with pytest.raises(AgentNotFoundError): |
| await conductor.execute("missing", context) |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_conductor_rejects_mismatched_result_name() -> None: |
| registry = AgentRegistry([StubAgent("planner", result_name="research")]) |
| conductor = Conductor(registry) |
| context = AgentContext( |
| task=Task( |
| name="Validate result", |
| description="Verify agent identity enforcement.", |
| ) |
| ) |
|
|
| with pytest.raises( |
| ValueError, |
| match="Agent result name does not match", |
| ): |
| await conductor.execute("planner", context) |
|
|
