You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1020 lines
37 KiB
1020 lines
37 KiB
"""
|
|
Comprehensive test suite for Board of Directors Swarm.
|
|
|
|
This module contains extensive tests for the Board of Directors swarm implementation,
|
|
covering all aspects including initialization, board operations, task execution,
|
|
error handling, and performance characteristics.
|
|
|
|
The test suite follows the Swarms testing philosophy:
|
|
- Comprehensive coverage of all functionality
|
|
- Proper mocking and isolation
|
|
- Performance and integration testing
|
|
- Error handling validation
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
import asyncio
|
|
from unittest.mock import Mock, patch, MagicMock, AsyncMock
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from swarms.structs.board_of_directors_swarm import (
|
|
BoardOfDirectorsSwarm,
|
|
BoardMember,
|
|
BoardMemberRole,
|
|
BoardDecisionType,
|
|
BoardOrder,
|
|
BoardDecision,
|
|
BoardSpec,
|
|
)
|
|
from swarms.structs.agent import Agent
|
|
from swarms.structs.conversation import Conversation
|
|
|
|
|
|
# Test fixtures
|
|
@pytest.fixture
|
|
def mock_agent():
|
|
"""Create a mock agent for testing."""
|
|
agent = Mock(spec=Agent)
|
|
agent.agent_name = "TestAgent"
|
|
agent.agent_description = "A test agent for unit testing"
|
|
agent.run = Mock(return_value="Test agent response")
|
|
agent.arun = AsyncMock(return_value="Async test agent response")
|
|
return agent
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_board_member(mock_agent):
|
|
"""Create a mock board member for testing."""
|
|
return BoardMember(
|
|
agent=mock_agent,
|
|
role=BoardMemberRole.CHAIRMAN,
|
|
voting_weight=1.5,
|
|
expertise_areas=["leadership", "strategy"]
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_agents():
|
|
"""Create sample agents for testing."""
|
|
agents = []
|
|
for i in range(3):
|
|
agent = Mock(spec=Agent)
|
|
agent.agent_name = f"Agent{i+1}"
|
|
agent.agent_description = f"Test agent {i+1}"
|
|
agent.run = Mock(return_value=f"Response from Agent{i+1}")
|
|
agents.append(agent)
|
|
return agents
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_board_members(sample_agents):
|
|
"""Create sample board members for testing."""
|
|
roles = [BoardMemberRole.CHAIRMAN, BoardMemberRole.VICE_CHAIRMAN, BoardMemberRole.SECRETARY]
|
|
board_members = []
|
|
|
|
for i, (agent, role) in enumerate(zip(sample_agents, roles)):
|
|
board_member = BoardMember(
|
|
agent=agent,
|
|
role=role,
|
|
voting_weight=1.0 + (i * 0.2),
|
|
expertise_areas=[f"expertise_{i+1}"]
|
|
)
|
|
board_members.append(board_member)
|
|
|
|
return board_members
|
|
|
|
|
|
@pytest.fixture
|
|
def basic_board_swarm(sample_agents):
|
|
"""Create a basic Board of Directors swarm for testing."""
|
|
return BoardOfDirectorsSwarm(
|
|
name="TestBoard",
|
|
agents=sample_agents,
|
|
verbose=False,
|
|
max_loops=1
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def configured_board_swarm(sample_agents, sample_board_members):
|
|
"""Create a configured Board of Directors swarm for testing."""
|
|
return BoardOfDirectorsSwarm(
|
|
name="ConfiguredBoard",
|
|
description="A configured board for testing",
|
|
board_members=sample_board_members,
|
|
agents=sample_agents,
|
|
max_loops=2,
|
|
verbose=True,
|
|
decision_threshold=0.7,
|
|
enable_voting=True,
|
|
enable_consensus=True,
|
|
max_workers=4
|
|
)
|
|
|
|
|
|
# Unit tests for enums and data models
|
|
class TestBoardMemberRole:
|
|
"""Test BoardMemberRole enum."""
|
|
|
|
def test_enum_values(self):
|
|
"""Test that all enum values are correctly defined."""
|
|
assert BoardMemberRole.CHAIRMAN == "chairman"
|
|
assert BoardMemberRole.VICE_CHAIRMAN == "vice_chairman"
|
|
assert BoardMemberRole.SECRETARY == "secretary"
|
|
assert BoardMemberRole.TREASURER == "treasurer"
|
|
assert BoardMemberRole.MEMBER == "member"
|
|
assert BoardMemberRole.EXECUTIVE_DIRECTOR == "executive_director"
|
|
|
|
|
|
class TestBoardDecisionType:
|
|
"""Test BoardDecisionType enum."""
|
|
|
|
def test_enum_values(self):
|
|
"""Test that all enum values are correctly defined."""
|
|
assert BoardDecisionType.UNANIMOUS == "unanimous"
|
|
assert BoardDecisionType.MAJORITY == "majority"
|
|
assert BoardDecisionType.CONSENSUS == "consensus"
|
|
assert BoardDecisionType.CHAIRMAN_DECISION == "chairman_decision"
|
|
|
|
|
|
class TestBoardMember:
|
|
"""Test BoardMember dataclass."""
|
|
|
|
def test_board_member_creation(self, mock_agent):
|
|
"""Test creating a board member."""
|
|
board_member = BoardMember(
|
|
agent=mock_agent,
|
|
role=BoardMemberRole.CHAIRMAN,
|
|
voting_weight=1.5,
|
|
expertise_areas=["leadership", "strategy"]
|
|
)
|
|
|
|
assert board_member.agent == mock_agent
|
|
assert board_member.role == BoardMemberRole.CHAIRMAN
|
|
assert board_member.voting_weight == 1.5
|
|
assert board_member.expertise_areas == ["leadership", "strategy"]
|
|
|
|
def test_board_member_defaults(self, mock_agent):
|
|
"""Test board member with default values."""
|
|
board_member = BoardMember(
|
|
agent=mock_agent,
|
|
role=BoardMemberRole.MEMBER
|
|
)
|
|
|
|
assert board_member.voting_weight == 1.0
|
|
assert board_member.expertise_areas == []
|
|
|
|
def test_board_member_post_init(self, mock_agent):
|
|
"""Test board member post-init with None expertise areas."""
|
|
board_member = BoardMember(
|
|
agent=mock_agent,
|
|
role=BoardMemberRole.MEMBER,
|
|
expertise_areas=None
|
|
)
|
|
|
|
assert board_member.expertise_areas == []
|
|
|
|
|
|
class TestBoardOrder:
|
|
"""Test BoardOrder model."""
|
|
|
|
def test_board_order_creation(self):
|
|
"""Test creating a board order."""
|
|
order = BoardOrder(
|
|
agent_name="TestAgent",
|
|
task="Test task",
|
|
priority=1,
|
|
deadline="2024-01-01",
|
|
assigned_by="Chairman"
|
|
)
|
|
|
|
assert order.agent_name == "TestAgent"
|
|
assert order.task == "Test task"
|
|
assert order.priority == 1
|
|
assert order.deadline == "2024-01-01"
|
|
assert order.assigned_by == "Chairman"
|
|
|
|
def test_board_order_defaults(self):
|
|
"""Test board order with default values."""
|
|
order = BoardOrder(
|
|
agent_name="TestAgent",
|
|
task="Test task"
|
|
)
|
|
|
|
assert order.priority == 3
|
|
assert order.deadline is None
|
|
assert order.assigned_by == "Board of Directors"
|
|
|
|
def test_board_order_validation(self):
|
|
"""Test board order validation."""
|
|
# Test priority validation
|
|
with pytest.raises(ValueError):
|
|
BoardOrder(
|
|
agent_name="TestAgent",
|
|
task="Test task",
|
|
priority=0 # Invalid priority
|
|
)
|
|
|
|
with pytest.raises(ValueError):
|
|
BoardOrder(
|
|
agent_name="TestAgent",
|
|
task="Test task",
|
|
priority=6 # Invalid priority
|
|
)
|
|
|
|
|
|
class TestBoardDecision:
|
|
"""Test BoardDecision model."""
|
|
|
|
def test_board_decision_creation(self):
|
|
"""Test creating a board decision."""
|
|
decision = BoardDecision(
|
|
decision_type=BoardDecisionType.MAJORITY,
|
|
decision="Approve the proposal",
|
|
votes_for=3,
|
|
votes_against=1,
|
|
abstentions=0,
|
|
reasoning="The proposal aligns with our strategic goals"
|
|
)
|
|
|
|
assert decision.decision_type == BoardDecisionType.MAJORITY
|
|
assert decision.decision == "Approve the proposal"
|
|
assert decision.votes_for == 3
|
|
assert decision.votes_against == 1
|
|
assert decision.abstentions == 0
|
|
assert decision.reasoning == "The proposal aligns with our strategic goals"
|
|
|
|
def test_board_decision_defaults(self):
|
|
"""Test board decision with default values."""
|
|
decision = BoardDecision(
|
|
decision_type=BoardDecisionType.CONSENSUS,
|
|
decision="Test decision"
|
|
)
|
|
|
|
assert decision.votes_for == 0
|
|
assert decision.votes_against == 0
|
|
assert decision.abstentions == 0
|
|
assert decision.reasoning == ""
|
|
|
|
|
|
class TestBoardSpec:
|
|
"""Test BoardSpec model."""
|
|
|
|
def test_board_spec_creation(self):
|
|
"""Test creating a board spec."""
|
|
orders = [
|
|
BoardOrder(agent_name="Agent1", task="Task 1"),
|
|
BoardOrder(agent_name="Agent2", task="Task 2")
|
|
]
|
|
decisions = [
|
|
BoardDecision(
|
|
decision_type=BoardDecisionType.MAJORITY,
|
|
decision="Decision 1"
|
|
)
|
|
]
|
|
|
|
spec = BoardSpec(
|
|
plan="Test plan",
|
|
orders=orders,
|
|
decisions=decisions,
|
|
meeting_summary="Test meeting summary"
|
|
)
|
|
|
|
assert spec.plan == "Test plan"
|
|
assert len(spec.orders) == 2
|
|
assert len(spec.decisions) == 1
|
|
assert spec.meeting_summary == "Test meeting summary"
|
|
|
|
def test_board_spec_defaults(self):
|
|
"""Test board spec with default values."""
|
|
spec = BoardSpec(
|
|
plan="Test plan",
|
|
orders=[]
|
|
)
|
|
|
|
assert spec.decisions == []
|
|
assert spec.meeting_summary == ""
|
|
|
|
|
|
# Unit tests for BoardOfDirectorsSwarm
|
|
class TestBoardOfDirectorsSwarmInitialization:
|
|
"""Test BoardOfDirectorsSwarm initialization."""
|
|
|
|
def test_basic_initialization(self, sample_agents):
|
|
"""Test basic swarm initialization."""
|
|
swarm = BoardOfDirectorsSwarm(
|
|
name="TestSwarm",
|
|
agents=sample_agents
|
|
)
|
|
|
|
assert swarm.name == "TestSwarm"
|
|
assert len(swarm.agents) == 3
|
|
assert swarm.max_loops == 1
|
|
assert swarm.verbose is False
|
|
assert swarm.decision_threshold == 0.6
|
|
|
|
def test_configured_initialization(self, sample_agents, sample_board_members):
|
|
"""Test configured swarm initialization."""
|
|
swarm = BoardOfDirectorsSwarm(
|
|
name="ConfiguredSwarm",
|
|
description="Test description",
|
|
board_members=sample_board_members,
|
|
agents=sample_agents,
|
|
max_loops=3,
|
|
verbose=True,
|
|
decision_threshold=0.8,
|
|
enable_voting=False,
|
|
enable_consensus=False,
|
|
max_workers=8
|
|
)
|
|
|
|
assert swarm.name == "ConfiguredSwarm"
|
|
assert swarm.description == "Test description"
|
|
assert len(swarm.board_members) == 3
|
|
assert len(swarm.agents) == 3
|
|
assert swarm.max_loops == 3
|
|
assert swarm.verbose is True
|
|
assert swarm.decision_threshold == 0.8
|
|
assert swarm.enable_voting is False
|
|
assert swarm.enable_consensus is False
|
|
assert swarm.max_workers == 8
|
|
|
|
def test_default_board_setup(self, sample_agents):
|
|
"""Test default board setup when no board members provided."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
|
|
assert len(swarm.board_members) == 3
|
|
assert swarm.board_members[0].role == BoardMemberRole.CHAIRMAN
|
|
assert swarm.board_members[1].role == BoardMemberRole.VICE_CHAIRMAN
|
|
assert swarm.board_members[2].role == BoardMemberRole.SECRETARY
|
|
|
|
def test_initialization_without_agents(self):
|
|
"""Test initialization without agents should raise error."""
|
|
with pytest.raises(ValueError, match="No agents found in the swarm"):
|
|
BoardOfDirectorsSwarm(agents=[])
|
|
|
|
def test_initialization_with_invalid_max_loops(self, sample_agents):
|
|
"""Test initialization with invalid max_loops."""
|
|
with pytest.raises(ValueError, match="Max loops must be greater than 0"):
|
|
BoardOfDirectorsSwarm(agents=sample_agents, max_loops=0)
|
|
|
|
def test_initialization_with_invalid_decision_threshold(self, sample_agents):
|
|
"""Test initialization with invalid decision threshold."""
|
|
with pytest.raises(ValueError, match="Decision threshold must be between 0.0 and 1.0"):
|
|
BoardOfDirectorsSwarm(agents=sample_agents, decision_threshold=1.5)
|
|
|
|
|
|
class TestBoardOfDirectorsSwarmMethods:
|
|
"""Test BoardOfDirectorsSwarm methods."""
|
|
|
|
def test_setup_default_board(self, sample_agents):
|
|
"""Test default board setup."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
|
|
assert len(swarm.board_members) == 3
|
|
assert all(hasattr(member.agent, 'agent_name') for member in swarm.board_members)
|
|
assert all(hasattr(member.agent, 'run') for member in swarm.board_members)
|
|
|
|
def test_get_chairman_prompt(self, sample_agents):
|
|
"""Test chairman prompt generation."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
prompt = swarm._get_chairman_prompt()
|
|
|
|
assert "Chairman" in prompt
|
|
assert "board meetings" in prompt
|
|
assert "consensus" in prompt
|
|
|
|
def test_get_vice_chairman_prompt(self, sample_agents):
|
|
"""Test vice chairman prompt generation."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
prompt = swarm._get_vice_chairman_prompt()
|
|
|
|
assert "Vice Chairman" in prompt
|
|
assert "supporting" in prompt
|
|
assert "operational" in prompt
|
|
|
|
def test_get_secretary_prompt(self, sample_agents):
|
|
"""Test secretary prompt generation."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
prompt = swarm._get_secretary_prompt()
|
|
|
|
assert "Secretary" in prompt
|
|
assert "documenting" in prompt
|
|
assert "records" in prompt
|
|
|
|
def test_format_board_members_info(self, configured_board_swarm):
|
|
"""Test board members info formatting."""
|
|
info = configured_board_swarm._format_board_members_info()
|
|
|
|
assert "Chairman" in info
|
|
assert "Vice-Chairman" in info
|
|
assert "Secretary" in info
|
|
assert "expertise" in info
|
|
|
|
def test_add_board_member(self, basic_board_swarm, mock_board_member):
|
|
"""Test adding a board member."""
|
|
initial_count = len(basic_board_swarm.board_members)
|
|
basic_board_swarm.add_board_member(mock_board_member)
|
|
|
|
assert len(basic_board_swarm.board_members) == initial_count + 1
|
|
assert mock_board_member in basic_board_swarm.board_members
|
|
|
|
def test_remove_board_member(self, configured_board_swarm):
|
|
"""Test removing a board member."""
|
|
member_to_remove = configured_board_swarm.board_members[0]
|
|
member_name = member_to_remove.agent.agent_name
|
|
|
|
initial_count = len(configured_board_swarm.board_members)
|
|
configured_board_swarm.remove_board_member(member_name)
|
|
|
|
assert len(configured_board_swarm.board_members) == initial_count - 1
|
|
assert member_to_remove not in configured_board_swarm.board_members
|
|
|
|
def test_get_board_member(self, configured_board_swarm):
|
|
"""Test getting a board member by name."""
|
|
member = configured_board_swarm.board_members[0]
|
|
member_name = member.agent.agent_name
|
|
|
|
found_member = configured_board_swarm.get_board_member(member_name)
|
|
assert found_member == member
|
|
|
|
# Test with non-existent member
|
|
not_found = configured_board_swarm.get_board_member("NonExistent")
|
|
assert not_found is None
|
|
|
|
def test_get_board_summary(self, configured_board_swarm):
|
|
"""Test getting board summary."""
|
|
summary = configured_board_swarm.get_board_summary()
|
|
|
|
assert "board_name" in summary
|
|
assert "total_members" in summary
|
|
assert "total_agents" in summary
|
|
assert "max_loops" in summary
|
|
assert "decision_threshold" in summary
|
|
assert "members" in summary
|
|
|
|
assert summary["board_name"] == "ConfiguredBoard"
|
|
assert summary["total_members"] == 3
|
|
assert summary["total_agents"] == 3
|
|
|
|
|
|
class TestBoardMeetingOperations:
|
|
"""Test board meeting operations."""
|
|
|
|
def test_create_board_meeting_prompt(self, configured_board_swarm):
|
|
"""Test board meeting prompt creation."""
|
|
task = "Test task for board meeting"
|
|
prompt = configured_board_swarm._create_board_meeting_prompt(task)
|
|
|
|
assert task in prompt
|
|
assert "BOARD OF DIRECTORS MEETING" in prompt
|
|
assert "INSTRUCTIONS" in prompt
|
|
assert "plan" in prompt
|
|
assert "orders" in prompt
|
|
|
|
def test_conduct_board_discussion(self, configured_board_swarm):
|
|
"""Test board discussion conduction."""
|
|
prompt = "Test board meeting prompt"
|
|
|
|
with patch.object(configured_board_swarm.board_members[0].agent, 'run') as mock_run:
|
|
mock_run.return_value = "Board discussion result"
|
|
result = configured_board_swarm._conduct_board_discussion(prompt)
|
|
|
|
assert result == "Board discussion result"
|
|
mock_run.assert_called_once_with(task=prompt, img=None)
|
|
|
|
def test_conduct_board_discussion_no_chairman(self, sample_agents):
|
|
"""Test board discussion when no chairman is found."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
# Remove all board members
|
|
swarm.board_members = []
|
|
|
|
with pytest.raises(ValueError, match="No chairman found in board members"):
|
|
swarm._conduct_board_discussion("Test prompt")
|
|
|
|
def test_parse_board_decisions_valid_json(self, configured_board_swarm):
|
|
"""Test parsing valid JSON board decisions."""
|
|
valid_json = """
|
|
{
|
|
"plan": "Test plan",
|
|
"orders": [
|
|
{
|
|
"agent_name": "Agent1",
|
|
"task": "Task 1",
|
|
"priority": 1,
|
|
"assigned_by": "Chairman"
|
|
}
|
|
],
|
|
"decisions": [
|
|
{
|
|
"decision_type": "majority",
|
|
"decision": "Test decision",
|
|
"votes_for": 2,
|
|
"votes_against": 1,
|
|
"abstentions": 0,
|
|
"reasoning": "Test reasoning"
|
|
}
|
|
],
|
|
"meeting_summary": "Test summary"
|
|
}
|
|
"""
|
|
|
|
result = configured_board_swarm._parse_board_decisions(valid_json)
|
|
|
|
assert isinstance(result, BoardSpec)
|
|
assert result.plan == "Test plan"
|
|
assert len(result.orders) == 1
|
|
assert len(result.decisions) == 1
|
|
assert result.meeting_summary == "Test summary"
|
|
|
|
def test_parse_board_decisions_invalid_json(self, configured_board_swarm):
|
|
"""Test parsing invalid JSON board decisions."""
|
|
invalid_json = "Invalid JSON content"
|
|
|
|
result = configured_board_swarm._parse_board_decisions(invalid_json)
|
|
|
|
assert isinstance(result, BoardSpec)
|
|
assert result.plan == invalid_json
|
|
assert len(result.orders) == 0
|
|
assert len(result.decisions) == 0
|
|
assert result.meeting_summary == "Parsing failed, using raw output"
|
|
|
|
def test_run_board_meeting(self, configured_board_swarm):
|
|
"""Test running a complete board meeting."""
|
|
task = "Test board meeting task"
|
|
|
|
with patch.object(configured_board_swarm, '_conduct_board_discussion') as mock_discuss:
|
|
with patch.object(configured_board_swarm, '_parse_board_decisions') as mock_parse:
|
|
mock_discuss.return_value = "Board discussion"
|
|
mock_parse.return_value = BoardSpec(
|
|
plan="Test plan",
|
|
orders=[],
|
|
decisions=[],
|
|
meeting_summary="Test summary"
|
|
)
|
|
|
|
result = configured_board_swarm.run_board_meeting(task)
|
|
|
|
assert isinstance(result, BoardSpec)
|
|
mock_discuss.assert_called_once()
|
|
mock_parse.assert_called_once_with("Board discussion")
|
|
|
|
|
|
class TestTaskExecution:
|
|
"""Test task execution methods."""
|
|
|
|
def test_call_single_agent(self, configured_board_swarm):
|
|
"""Test calling a single agent."""
|
|
agent_name = "Agent1"
|
|
task = "Test task"
|
|
|
|
with patch.object(configured_board_swarm.agents[0], 'run') as mock_run:
|
|
mock_run.return_value = "Agent response"
|
|
result = configured_board_swarm._call_single_agent(agent_name, task)
|
|
|
|
assert result == "Agent response"
|
|
mock_run.assert_called_once()
|
|
|
|
def test_call_single_agent_not_found(self, configured_board_swarm):
|
|
"""Test calling a non-existent agent."""
|
|
with pytest.raises(ValueError, match="Agent 'NonExistent' not found"):
|
|
configured_board_swarm._call_single_agent("NonExistent", "Test task")
|
|
|
|
def test_execute_single_order(self, configured_board_swarm):
|
|
"""Test executing a single order."""
|
|
order = BoardOrder(
|
|
agent_name="Agent1",
|
|
task="Test order task",
|
|
priority=1,
|
|
assigned_by="Chairman"
|
|
)
|
|
|
|
with patch.object(configured_board_swarm, '_call_single_agent') as mock_call:
|
|
mock_call.return_value = "Order execution result"
|
|
result = configured_board_swarm._execute_single_order(order)
|
|
|
|
assert result == "Order execution result"
|
|
mock_call.assert_called_once_with(
|
|
agent_name="Agent1",
|
|
task="Test order task"
|
|
)
|
|
|
|
def test_execute_orders(self, configured_board_swarm):
|
|
"""Test executing multiple orders."""
|
|
orders = [
|
|
BoardOrder(agent_name="Agent1", task="Task 1", priority=1),
|
|
BoardOrder(agent_name="Agent2", task="Task 2", priority=2),
|
|
]
|
|
|
|
with patch.object(configured_board_swarm, '_execute_single_order') as mock_execute:
|
|
mock_execute.side_effect = ["Result 1", "Result 2"]
|
|
results = configured_board_swarm._execute_orders(orders)
|
|
|
|
assert len(results) == 2
|
|
assert results[0]["agent_name"] == "Agent1"
|
|
assert results[0]["output"] == "Result 1"
|
|
assert results[1]["agent_name"] == "Agent2"
|
|
assert results[1]["output"] == "Result 2"
|
|
|
|
def test_generate_board_feedback(self, configured_board_swarm):
|
|
"""Test generating board feedback."""
|
|
outputs = [
|
|
{"agent_name": "Agent1", "output": "Output 1"},
|
|
{"agent_name": "Agent2", "output": "Output 2"}
|
|
]
|
|
|
|
with patch.object(configured_board_swarm.board_members[0].agent, 'run') as mock_run:
|
|
mock_run.return_value = "Board feedback"
|
|
result = configured_board_swarm._generate_board_feedback(outputs)
|
|
|
|
assert result == "Board feedback"
|
|
mock_run.assert_called_once()
|
|
|
|
def test_generate_board_feedback_no_chairman(self, sample_agents):
|
|
"""Test generating feedback when no chairman is found."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
swarm.board_members = [] # Remove all board members
|
|
|
|
with pytest.raises(ValueError, match="No chairman found for feedback"):
|
|
swarm._generate_board_feedback([])
|
|
|
|
|
|
class TestStepAndRunMethods:
|
|
"""Test step and run methods."""
|
|
|
|
def test_step_method(self, configured_board_swarm):
|
|
"""Test the step method."""
|
|
task = "Test step task"
|
|
|
|
with patch.object(configured_board_swarm, 'run_board_meeting') as mock_meeting:
|
|
with patch.object(configured_board_swarm, '_execute_orders') as mock_execute:
|
|
with patch.object(configured_board_swarm, '_generate_board_feedback') as mock_feedback:
|
|
mock_meeting.return_value = BoardSpec(
|
|
plan="Test plan",
|
|
orders=[BoardOrder(agent_name="Agent1", task="Task 1")],
|
|
decisions=[],
|
|
meeting_summary="Test summary"
|
|
)
|
|
mock_execute.return_value = [{"agent_name": "Agent1", "output": "Result"}]
|
|
mock_feedback.return_value = "Board feedback"
|
|
|
|
result = configured_board_swarm.step(task)
|
|
|
|
assert result == "Board feedback"
|
|
mock_meeting.assert_called_once_with(task=task, img=None)
|
|
mock_execute.assert_called_once()
|
|
mock_feedback.assert_called_once()
|
|
|
|
def test_step_method_no_feedback(self, configured_board_swarm):
|
|
"""Test the step method with feedback disabled."""
|
|
configured_board_swarm.board_feedback_on = False
|
|
task = "Test step task"
|
|
|
|
with patch.object(configured_board_swarm, 'run_board_meeting') as mock_meeting:
|
|
with patch.object(configured_board_swarm, '_execute_orders') as mock_execute:
|
|
mock_meeting.return_value = BoardSpec(
|
|
plan="Test plan",
|
|
orders=[BoardOrder(agent_name="Agent1", task="Task 1")],
|
|
decisions=[],
|
|
meeting_summary="Test summary"
|
|
)
|
|
mock_execute.return_value = [{"agent_name": "Agent1", "output": "Result"}]
|
|
|
|
result = configured_board_swarm.step(task)
|
|
|
|
assert result == [{"agent_name": "Agent1", "output": "Result"}]
|
|
|
|
def test_run_method(self, configured_board_swarm):
|
|
"""Test the run method."""
|
|
task = "Test run task"
|
|
|
|
with patch.object(configured_board_swarm, 'step') as mock_step:
|
|
with patch.object(configured_board_swarm, 'conversation') as mock_conversation:
|
|
mock_step.return_value = "Step result"
|
|
mock_conversation.add = Mock()
|
|
|
|
result = configured_board_swarm.run(task)
|
|
|
|
assert mock_step.call_count == 2 # max_loops = 2
|
|
assert mock_conversation.add.call_count == 2
|
|
|
|
def test_arun_method(self, configured_board_swarm):
|
|
"""Test the async run method."""
|
|
task = "Test async run task"
|
|
|
|
with patch.object(configured_board_swarm, 'run') as mock_run:
|
|
mock_run.return_value = "Async result"
|
|
|
|
async def test_async():
|
|
result = await configured_board_swarm.arun(task)
|
|
return result
|
|
|
|
result = asyncio.run(test_async())
|
|
assert result == "Async result"
|
|
mock_run.assert_called_once_with(task=task, img=None)
|
|
|
|
|
|
# Integration tests
|
|
class TestBoardOfDirectorsSwarmIntegration:
|
|
"""Integration tests for BoardOfDirectorsSwarm."""
|
|
|
|
def test_full_workflow_integration(self, sample_agents):
|
|
"""Test full workflow integration."""
|
|
swarm = BoardOfDirectorsSwarm(
|
|
agents=sample_agents,
|
|
verbose=False,
|
|
max_loops=1
|
|
)
|
|
|
|
task = "Create a simple report"
|
|
|
|
# Mock the board discussion to return structured output
|
|
mock_board_output = """
|
|
{
|
|
"plan": "Create a comprehensive report",
|
|
"orders": [
|
|
{
|
|
"agent_name": "Agent1",
|
|
"task": "Research the topic",
|
|
"priority": 1,
|
|
"assigned_by": "Chairman"
|
|
},
|
|
{
|
|
"agent_name": "Agent2",
|
|
"task": "Write the report",
|
|
"priority": 2,
|
|
"assigned_by": "Chairman"
|
|
}
|
|
],
|
|
"decisions": [
|
|
{
|
|
"decision_type": "consensus",
|
|
"decision": "Proceed with report creation",
|
|
"votes_for": 3,
|
|
"votes_against": 0,
|
|
"abstentions": 0,
|
|
"reasoning": "Report is needed for decision making"
|
|
}
|
|
],
|
|
"meeting_summary": "Board agreed to create a comprehensive report"
|
|
}
|
|
"""
|
|
|
|
with patch.object(swarm.board_members[0].agent, 'run') as mock_run:
|
|
mock_run.return_value = mock_board_output
|
|
result = swarm.run(task)
|
|
|
|
assert result is not None
|
|
assert isinstance(result, dict)
|
|
|
|
def test_board_member_management_integration(self, sample_agents):
|
|
"""Test board member management integration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents)
|
|
|
|
# Test adding a new board member
|
|
new_member = BoardMember(
|
|
agent=sample_agents[0],
|
|
role=BoardMemberRole.MEMBER,
|
|
voting_weight=1.0,
|
|
expertise_areas=["testing"]
|
|
)
|
|
|
|
initial_count = len(swarm.board_members)
|
|
swarm.add_board_member(new_member)
|
|
assert len(swarm.board_members) == initial_count + 1
|
|
|
|
# Test removing a board member
|
|
member_name = swarm.board_members[0].agent.agent_name
|
|
swarm.remove_board_member(member_name)
|
|
assert len(swarm.board_members) == initial_count
|
|
|
|
# Test getting board member
|
|
member = swarm.get_board_member(swarm.board_members[0].agent.agent_name)
|
|
assert member is not None
|
|
|
|
|
|
# Parameterized tests
|
|
@pytest.mark.parametrize("max_loops", [1, 2, 3])
|
|
def test_max_loops_parameterization(sample_agents, max_loops):
|
|
"""Test swarm with different max_loops values."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, max_loops=max_loops)
|
|
assert swarm.max_loops == max_loops
|
|
|
|
|
|
@pytest.mark.parametrize("decision_threshold", [0.5, 0.6, 0.7, 0.8, 0.9])
|
|
def test_decision_threshold_parameterization(sample_agents, decision_threshold):
|
|
"""Test swarm with different decision threshold values."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, decision_threshold=decision_threshold)
|
|
assert swarm.decision_threshold == decision_threshold
|
|
|
|
|
|
@pytest.mark.parametrize("board_model", ["gpt-4o-mini", "gpt-4", "claude-3-sonnet"])
|
|
def test_board_model_parameterization(sample_agents, board_model):
|
|
"""Test swarm with different board models."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_model_name=board_model)
|
|
assert swarm.board_model_name == board_model
|
|
|
|
|
|
# Error handling tests
|
|
class TestBoardOfDirectorsSwarmErrorHandling:
|
|
"""Test error handling in BoardOfDirectorsSwarm."""
|
|
|
|
def test_initialization_error_handling(self):
|
|
"""Test error handling during initialization."""
|
|
with pytest.raises(ValueError):
|
|
BoardOfDirectorsSwarm(agents=[])
|
|
|
|
def test_board_meeting_error_handling(self, configured_board_swarm):
|
|
"""Test error handling during board meeting."""
|
|
with patch.object(configured_board_swarm, '_conduct_board_discussion') as mock_discuss:
|
|
mock_discuss.side_effect = Exception("Board meeting failed")
|
|
|
|
with pytest.raises(Exception, match="Board meeting failed"):
|
|
configured_board_swarm.run_board_meeting("Test task")
|
|
|
|
def test_task_execution_error_handling(self, configured_board_swarm):
|
|
"""Test error handling during task execution."""
|
|
with patch.object(configured_board_swarm, '_call_single_agent') as mock_call:
|
|
mock_call.side_effect = Exception("Task execution failed")
|
|
|
|
with pytest.raises(Exception, match="Task execution failed"):
|
|
configured_board_swarm._call_single_agent("Agent1", "Test task")
|
|
|
|
def test_order_execution_error_handling(self, configured_board_swarm):
|
|
"""Test error handling during order execution."""
|
|
orders = [BoardOrder(agent_name="Agent1", task="Task 1")]
|
|
|
|
with patch.object(configured_board_swarm, '_execute_single_order') as mock_execute:
|
|
mock_execute.side_effect = Exception("Order execution failed")
|
|
|
|
# Should not raise exception, but log error
|
|
results = configured_board_swarm._execute_orders(orders)
|
|
assert len(results) == 1
|
|
assert "Error" in results[0]["output"]
|
|
|
|
|
|
# Performance tests
|
|
class TestBoardOfDirectorsSwarmPerformance:
|
|
"""Test performance characteristics of BoardOfDirectorsSwarm."""
|
|
|
|
def test_parallel_execution_performance(self, sample_agents):
|
|
"""Test parallel execution performance."""
|
|
import time
|
|
|
|
swarm = BoardOfDirectorsSwarm(
|
|
agents=sample_agents,
|
|
max_workers=3,
|
|
verbose=False
|
|
)
|
|
|
|
# Create multiple orders
|
|
orders = [
|
|
BoardOrder(agent_name=f"Agent{i+1}", task=f"Task {i+1}")
|
|
for i in range(3)
|
|
]
|
|
|
|
start_time = time.time()
|
|
|
|
with patch.object(swarm, '_execute_single_order') as mock_execute:
|
|
mock_execute.side_effect = lambda order: f"Result for {order.task}"
|
|
results = swarm._execute_orders(orders)
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
assert len(results) == 3
|
|
assert execution_time < 1.0 # Should complete quickly with parallel execution
|
|
|
|
def test_memory_usage(self, sample_agents):
|
|
"""Test memory usage characteristics."""
|
|
import psutil
|
|
import os
|
|
|
|
process = psutil.Process(os.getpid())
|
|
initial_memory = process.memory_info().rss
|
|
|
|
# Create multiple swarms
|
|
swarms = []
|
|
for i in range(5):
|
|
swarm = BoardOfDirectorsSwarm(
|
|
agents=sample_agents,
|
|
name=f"Swarm{i}",
|
|
verbose=False
|
|
)
|
|
swarms.append(swarm)
|
|
|
|
final_memory = process.memory_info().rss
|
|
memory_increase = final_memory - initial_memory
|
|
|
|
# Memory increase should be reasonable (less than 100MB)
|
|
assert memory_increase < 100 * 1024 * 1024
|
|
|
|
|
|
# Configuration tests
|
|
class TestBoardOfDirectorsSwarmConfiguration:
|
|
"""Test configuration options for BoardOfDirectorsSwarm."""
|
|
|
|
def test_verbose_configuration(self, sample_agents):
|
|
"""Test verbose configuration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, verbose=True)
|
|
assert swarm.verbose is True
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, verbose=False)
|
|
assert swarm.verbose is False
|
|
|
|
def test_collaboration_prompt_configuration(self, sample_agents):
|
|
"""Test collaboration prompt configuration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, add_collaboration_prompt=True)
|
|
assert swarm.add_collaboration_prompt is True
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, add_collaboration_prompt=False)
|
|
assert swarm.add_collaboration_prompt is False
|
|
|
|
def test_board_feedback_configuration(self, sample_agents):
|
|
"""Test board feedback configuration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_feedback_on=True)
|
|
assert swarm.board_feedback_on is True
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, board_feedback_on=False)
|
|
assert swarm.board_feedback_on is False
|
|
|
|
def test_voting_configuration(self, sample_agents):
|
|
"""Test voting configuration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_voting=True)
|
|
assert swarm.enable_voting is True
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_voting=False)
|
|
assert swarm.enable_voting is False
|
|
|
|
def test_consensus_configuration(self, sample_agents):
|
|
"""Test consensus configuration."""
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_consensus=True)
|
|
assert swarm.enable_consensus is True
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=sample_agents, enable_consensus=False)
|
|
assert swarm.enable_consensus is False
|
|
|
|
|
|
# Real integration tests (skipped if no API key)
|
|
@pytest.mark.skipif(
|
|
not os.getenv("OPENAI_API_KEY"),
|
|
reason="OpenAI API key not available"
|
|
)
|
|
class TestBoardOfDirectorsSwarmRealIntegration:
|
|
"""Real integration tests for BoardOfDirectorsSwarm."""
|
|
|
|
def test_real_board_meeting(self):
|
|
"""Test real board meeting with actual API calls."""
|
|
# Create real agents
|
|
agents = [
|
|
Agent(
|
|
agent_name="Researcher",
|
|
agent_description="Research analyst",
|
|
model_name="gpt-4o-mini",
|
|
max_loops=1
|
|
),
|
|
Agent(
|
|
agent_name="Writer",
|
|
agent_description="Content writer",
|
|
model_name="gpt-4o-mini",
|
|
max_loops=1
|
|
)
|
|
]
|
|
|
|
swarm = BoardOfDirectorsSwarm(
|
|
agents=agents,
|
|
verbose=False,
|
|
max_loops=1
|
|
)
|
|
|
|
task = "Create a brief market analysis report"
|
|
|
|
result = swarm.run(task)
|
|
|
|
assert result is not None
|
|
assert isinstance(result, dict)
|
|
assert "conversation_history" in result
|
|
|
|
def test_real_board_member_management(self):
|
|
"""Test real board member management."""
|
|
agents = [
|
|
Agent(
|
|
agent_name="TestAgent",
|
|
agent_description="Test agent",
|
|
model_name="gpt-4o-mini",
|
|
max_loops=1
|
|
)
|
|
]
|
|
|
|
swarm = BoardOfDirectorsSwarm(agents=agents, verbose=False)
|
|
|
|
# Test board summary
|
|
summary = swarm.get_board_summary()
|
|
assert summary["total_members"] == 3 # Default board
|
|
assert summary["total_agents"] == 1
|
|
|
|
|
|
# Test runner
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"]) |