Clean up tests

This commit is contained in:
Brandon Hancock
2025-03-14 10:09:21 -04:00
parent 4b6498de8b
commit c334feea7e
5 changed files with 53 additions and 81 deletions

View File

@@ -306,7 +306,6 @@ class BaseAgent(ABC, BaseModel):
copied_source.storage = shared_storage
existing_knowledge_sources.append(copied_source)
# Copy delegate_to if it exists
existing_delegate_to = None
if self.delegate_to:
existing_delegate_to = list(self.delegate_to)

View File

@@ -116,7 +116,7 @@ class Crew(BaseModel):
name: Optional[str] = Field(default=None)
cache: bool = Field(default=True)
tasks: List[Task] = Field(default_factory=list)
agents: Sequence[BaseAgent] = Field(default_factory=list)
agents: List[BaseAgent] = Field(default_factory=list)
process: Process = Field(default=Process.sequential)
verbose: bool = Field(default=False)
memory: bool = Field(

View File

@@ -1,4 +1,4 @@
from typing import Sequence
from typing import List
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.tools.base_tool import BaseTool
@@ -11,11 +11,11 @@ from .delegate_work_tool import DelegateWorkTool
class AgentTools:
"""Manager class for agent-related tools"""
def __init__(self, agents: Sequence[BaseAgent], i18n: I18N = I18N()):
def __init__(self, agents: List[BaseAgent], i18n: I18N = I18N()):
self.agents = agents
self.i18n = i18n
def tools(self) -> Sequence[BaseTool]:
def tools(self) -> List[BaseTool]:
"""Get all available agent tools"""
coworkers = ", ".join([f"{agent.role}" for agent in self.agents])

View File

@@ -1874,92 +1874,59 @@ def test_agent_delegation_to_specific_agents():
def test_agent_copy_with_delegate_to():
"""Test that the delegate_to property is properly copied when an agent is copied."""
# Create agents in order so we can reference them in delegate_to
agent2 = Agent(
role="Agent 2",
goal="Goal for Agent 2",
backstory="Backstory for Agent 2",
allow_delegation=True,
)
agent3 = Agent(
role="Agent 3",
goal="Goal for Agent 3",
backstory="Backstory for Agent 3",
allow_delegation=True,
)
# Create agent1 with delegate_to set during initialization
"""Test that the delegate_to attribute is properly copied when copying an agent."""
# Create a few agents for delegation
agent1 = Agent(
role="Agent 1",
goal="Goal for Agent 1",
backstory="Backstory for Agent 1",
allow_delegation=True,
delegate_to=[agent2, agent3],
)
# Copy agent1
agent1_copy = agent1.copy()
# Verify that delegate_to is properly copied
assert agent1_copy.delegate_to is not None
assert len(agent1_copy.delegate_to) == 2
# Verify that the copied delegate_to contains the same agents
delegate_roles = [agent.role for agent in agent1_copy.delegate_to]
assert "Agent 2" in delegate_roles
assert "Agent 3" in delegate_roles
# Verify that modifying the original agent's delegate_to doesn't affect the copy
agent1.delegate_to = [agent2]
assert len(agent1_copy.delegate_to) == 2
assert len(agent1.delegate_to) == 1
@pytest.mark.vcr(filter_headers=["authorization"])
def test_agent_delegation_to_all_agents():
"""Test that an agent with allow_delegation=True but without delegate_to specified can delegate to all agents."""
# Create three agents
agent1 = Agent(
role="Agent 1",
goal="Goal for Agent 1",
backstory="Backstory for Agent 1",
allow_delegation=True, # Allow delegation but don't specify delegate_to
role="Researcher",
goal="Research topics",
backstory="Experienced researcher",
)
agent2 = Agent(
role="Agent 2",
goal="Goal for Agent 2",
backstory="Backstory for Agent 2",
allow_delegation=True,
role="Writer",
goal="Write content",
backstory="Professional writer",
)
agent3 = Agent(
role="Agent 3",
goal="Goal for Agent 3",
backstory="Backstory for Agent 3",
role="Manager",
goal="Manage the team",
backstory="Expert manager",
allow_delegation=True,
delegate_to=[agent1, agent2], # This manager can delegate to agent1 and agent2
)
# Get delegation tools for agent1
all_agents = [agent1, agent2, agent3]
delegation_tools = agent1.get_delegation_tools(all_agents)
# Make a copy of the manager agent
copied_agent3 = agent3.copy()
# Verify that tools for all agents are returned
assert len(delegation_tools) == 2 # Delegate and Ask tools
# Verify the copied agent has the same delegation settings
assert copied_agent3.allow_delegation == agent3.allow_delegation
assert (
copied_agent3.delegate_to is not agent3.delegate_to
) # Should be different objects
assert copied_agent3.delegate_to is not None
assert agent3.delegate_to is not None
assert len(copied_agent3.delegate_to) == len(agent3.delegate_to)
assert all(a in copied_agent3.delegate_to for a in agent3.delegate_to)
# Check that the tools can delegate to all agents
delegate_tool = delegation_tools[0]
ask_tool = delegation_tools[1]
# Modify the original agent's delegate_to list
assert agent3.delegate_to is not None
agent3.delegate_to.pop()
# Verify the tools description includes all agents
assert "Agent 1" in delegate_tool.description
assert "Agent 2" in delegate_tool.description
assert "Agent 3" in delegate_tool.description
assert "Agent 1" in ask_tool.description
assert "Agent 2" in ask_tool.description
assert "Agent 3" in ask_tool.description
# Verify the copied agent's delegate_to list is not affected
assert copied_agent3.delegate_to is not None
assert agent3.delegate_to is not None
assert len(copied_agent3.delegate_to) == 2
assert len(agent3.delegate_to) == 1
# Verify that agent1.delegate_to is None
assert agent1.delegate_to is None
# Test copying an agent with delegate_to=None
agent4 = Agent(
role="Solo Worker",
goal="Work independently",
backstory="Independent worker",
allow_delegation=False,
delegate_to=None,
)
copied_agent4 = agent4.copy()
assert copied_agent4.delegate_to == agent4.delegate_to

View File

@@ -865,11 +865,17 @@ def test_crew_verbose_output(capsys):
crew._logger = Logger(verbose=False)
crew.kickoff()
captured = capsys.readouterr()
# Filter out event listener logs, escape codes, and now also 'tools:' lines
filtered_output = "\n".join(
line
for line in captured.out.split("\n")
if not line.startswith("[") and line.strip() and not line.startswith("\x1b")
if not line.startswith("[")
and line.strip()
and not line.startswith("\x1b")
and not "tools:" in line.lower() # Exclude 'tools:' lines
)
assert filtered_output == ""