test/test_agent_composition.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
import pytest
from pydantic import BaseModel
from ag2 import Agent, TaskConfig, tool
from ag2.middleware import (
ApprovalRequired,
LoggingMiddleware,
Middleware,
TokenLimiter,
)
from ag2.observers import observer
def alpha(x: int) -> int:
"""Alpha."""
return x
def beta(x: int) -> int:
"""Beta."""
return x
def make_hook(): # type: ignore[no-untyped-def]
async def hook(call_next, event, context): # type: ignore[no-untyped-def]
return await call_next(event, context)
return hook
class Out(BaseModel):
value: int
class TestAgentComposition:
def test_reports_what_the_agent_is_made_of(self) -> None:
agent = Agent(
"bot",
prompt="hello",
tools=[alpha],
dependencies={"db": object()},
variables={"k": 1},
)
assert agent.system_prompt == ("hello",)
assert [t.name for t in agent.tools] == ["alpha"]
assert list(agent.dependencies) == ["db"]
assert list(agent.variables) == ["k"]
def test_unset_slots_read_as_defaults(self) -> None:
agent = Agent("bot", prompt="p")
assert agent.dynamic_prompt == ()
assert agent.observers == ()
assert agent.assembly == ()
assert agent.tasks is None
assert agent.response_schema is None
@pytest.mark.filterwarnings("ignore:TaskConfig, Agent:DeprecationWarning")
def test_set_slots_are_reported(self) -> None:
@observer
def watcher(event, context) -> None: # type: ignore[no-untyped-def]
return None
agent = Agent("bot", prompt="p", observers=[watcher], tasks=TaskConfig(), response_schema=Out)
assert len(agent.observers) == 1
assert isinstance(agent.tasks, TaskConfig)
assert agent.response_schema is not None
def test_dynamic_prompt_hooks_are_nameable(self) -> None:
agent = Agent("bot", prompt="p")
@agent.prompt
async def greeting() -> str:
return "hi"
assert [hook.__name__ for hook in agent.dynamic_prompt] == ["greeting"]
class TestViewsAreReadOnly:
def test_sequences_are_immutable_snapshots(self) -> None:
agent = Agent("bot", prompt="p", tools=[alpha])
assert isinstance(agent.system_prompt, tuple)
assert isinstance(agent.dynamic_prompt, tuple)
assert isinstance(agent.middleware, tuple)
assert isinstance(agent.observers, tuple)
assert isinstance(agent.assembly, tuple)
def test_mappings_cannot_be_mutated_through_the_view(self) -> None:
agent = Agent("bot", prompt="p", dependencies={"db": object()}, variables={"k": 1})
with pytest.raises(TypeError):
agent.dependencies["db"] = object() # type: ignore[index]
with pytest.raises(TypeError):
agent.variables["k"] = 2 # type: ignore[index]
def test_dependency_values_are_the_injected_objects(self) -> None:
db = object()
agent = Agent("bot", prompt="p", dependencies={"db": db})
assert agent.dependencies["db"] is db
class TestMiddlewareEntries:
def test_agent_middleware_pairs_object_with_description(self) -> None:
limiter = TokenLimiter(max_tokens=10)
agent = Agent("bot", prompt="p", middleware=[limiter])
entry = agent.middleware[0]
assert entry.middleware is limiter
assert entry.description.kind == "TokenLimiter"
assert entry.description.config == {"max_tokens": 10, "chars_per_token": 4}
def test_tool_middleware_pairs_object_with_description(self) -> None:
guard = ApprovalRequired(timeout=5)
built = tool(alpha, middleware=[guard])
entry = built.middleware[0]
assert entry.middleware is guard
assert entry.description.kind == "ApprovalRequired"
def test_middleware_without_describe_reports_incomplete(self) -> None:
agent = Agent("bot", prompt="p", middleware=[make_hook()])
description = agent.middleware[0].description
assert description.complete is False
assert description.config == {}
def test_entries_are_built_on_access(self) -> None:
limiter = TokenLimiter(max_tokens=10)
agent = Agent("bot", prompt="p", middleware=[limiter])
# The entry is a fresh view each time; the middleware inside is not.
assert agent.middleware[0] is not agent.middleware[0]
assert agent.middleware[0].middleware is agent.middleware[0].middleware
class TestSharingIsObservable:
def test_one_instance_across_two_tools_reads_as_the_same_object(self) -> None:
shared = make_hook()
agent = Agent(
"bot",
prompt="p",
tools=[tool(alpha, middleware=[shared]), tool(beta, middleware=[shared])],
)
first, second = agent.tools
assert first.middleware[0].middleware is second.middleware[0].middleware
def test_separate_instances_with_equal_config_are_not_the_same_object(self) -> None:
agent = Agent(
"bot",
prompt="p",
tools=[tool(alpha, middleware=[make_hook()]), tool(beta, middleware=[make_hook()])],
)
first, second = agent.tools
# Descriptions cannot tell these apart, which is why identity is exposed.
assert first.middleware[0].description == second.middleware[0].description
assert first.middleware[0].middleware is not second.middleware[0].middleware
class TestMiddlewareFactoryFields:
def test_reports_the_wrapped_class_and_its_options(self) -> None:
factory = Middleware(LoggingMiddleware, level=10)
assert factory.cls is LoggingMiddleware
assert dict(factory.options) == {"level": 10}
def test_options_expose_values_that_the_description_withholds(self) -> None:
factory = Middleware(LoggingMiddleware, api_key="sk-SECRET-123")
# A description may be logged or committed as a fixture, so it reports
# names only. Reading the factory you already hold is a deliberate act.
assert factory.options["api_key"] == "sk-SECRET-123"
assert "sk-SECRET-123" not in repr(factory.describe())
def test_options_cannot_be_mutated_through_the_view(self) -> None:
factory = Middleware(LoggingMiddleware, level=10)
with pytest.raises(TypeError):
factory.options["level"] = 20 # type: ignore[index]
class TestComparingTwoAgents:
def test_independently_built_identical_agents_compare_equal(self) -> None:
def build() -> Agent:
return Agent(
"bot",
prompt="hello",
tools=[alpha],
middleware=[TokenLimiter(max_tokens=10)],
dependencies={"db": "shared"},
)
def fingerprint(agent: Agent) -> object:
return (
agent.name,
agent.system_prompt,
tuple(t.name for t in agent.tools),
tuple(m.description for m in agent.middleware),
tuple(sorted(map(str, agent.dependencies))),
)
assert fingerprint(build()) == fingerprint(build())
def test_a_changed_middleware_setting_is_caught(self) -> None:
loose = Agent("bot", prompt="p", middleware=[TokenLimiter(max_tokens=10)])
tight = Agent("bot", prompt="p", middleware=[TokenLimiter(max_tokens=99)])
assert loose.middleware[0].description != tight.middleware[0].description