63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""公共 Agent 注册表契约：业务组员接入时必须通过这些机器检查。"""
|
|
|
|
from collections.abc import Iterable
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
|
|
from app.service.agent.base import BaseAgent
|
|
from app.service.agent.factory import AgentFactory
|
|
|
|
|
|
class ContractAgent(BaseAgent):
|
|
definition = AgentDefinition(
|
|
agent_type="contract_agent", version="1", allowed_roles=("customer",),
|
|
allowed_portals=("api",)
|
|
)
|
|
|
|
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
|
return CoreResult(text=request.message)
|
|
|
|
|
|
def assert_registered_agents(factory: AgentFactory, context: RequestContext) -> None:
|
|
"""扫描全部注册项，保证定义、构造器和公共基类不可漂移。"""
|
|
builders: Iterable[tuple[str, object]] = factory._builders.items()
|
|
for agent_type, builder in builders:
|
|
definition = factory.definition(agent_type)
|
|
assert definition.agent_type == agent_type
|
|
agent = builder(context) # type: ignore[operator]
|
|
assert isinstance(agent, BaseAgent)
|
|
assert agent.definition == definition
|
|
|
|
|
|
def test_registration_table_is_scannable_and_builder_contract_holds() -> None:
|
|
factory = AgentFactory()
|
|
factory.register(
|
|
ContractAgent.definition,
|
|
lambda _context: ContractAgent(ContractAgent.definition),
|
|
)
|
|
context = RequestContext(
|
|
user_id="1", trace_id="contract", roles=("customer",), permissions=("agent:run",)
|
|
)
|
|
assert_registered_agents(factory, context)
|
|
assert factory.create("contract_agent", context).__class__ is ContractAgent
|
|
|
|
|
|
def test_contract_rejects_builder_that_does_not_return_base_agent() -> None:
|
|
class NotAnAgent(BaseModel):
|
|
value: str = "bad"
|
|
|
|
factory = AgentFactory()
|
|
definition = AgentDefinition(
|
|
agent_type="invalid_contract", version="1", allowed_roles=("customer",),
|
|
allowed_portals=("api",)
|
|
)
|
|
factory.register(definition, lambda _context: NotAnAgent()) # type: ignore[arg-type]
|
|
with pytest.raises(Exception, match="BaseAgent"):
|
|
factory.create(
|
|
"invalid_contract", RequestContext(
|
|
user_id="1", trace_id="bad", roles=("customer",), permissions=("agent:run",)
|
|
)
|
|
)
|