@ -15,8 +15,7 @@ The test suite follows the Swarms testing philosophy:
import os
import pytest
import asyncio
from unittest . mock import Mock , patch , MagicMock , AsyncMock
from typing import List , Dict , Any , Optional
from unittest . mock import Mock , patch , AsyncMock
from swarms . structs . board_of_directors_swarm import (
BoardOfDirectorsSwarm ,
@ -28,7 +27,6 @@ from swarms.structs.board_of_directors_swarm import (
BoardSpec ,
)
from swarms . structs . agent import Agent
from swarms . structs . conversation import Conversation
# Test fixtures
@ -50,7 +48,7 @@ def mock_board_member(mock_agent):
agent = mock_agent ,
role = BoardMemberRole . CHAIRMAN ,
voting_weight = 1.5 ,
expertise_areas = [ " leadership " , " strategy " ]
expertise_areas = [ " leadership " , " strategy " ] ,
)
@ -70,7 +68,11 @@ def sample_agents():
@pytest.fixture
def sample_board_members ( sample_agents ) :
""" Create sample board members for testing. """
roles = [ BoardMemberRole . CHAIRMAN , BoardMemberRole . VICE_CHAIRMAN , BoardMemberRole . SECRETARY ]
roles = [
BoardMemberRole . CHAIRMAN ,
BoardMemberRole . VICE_CHAIRMAN ,
BoardMemberRole . SECRETARY ,
]
board_members = [ ]
for i , ( agent , role ) in enumerate ( zip ( sample_agents , roles ) ) :
@ -78,7 +80,7 @@ def sample_board_members(sample_agents):
agent = agent ,
role = role ,
voting_weight = 1.0 + ( i * 0.2 ) ,
expertise_areas = [ f " expertise_ { i + 1 } " ]
expertise_areas = [ f " expertise_ { i + 1 } " ] ,
)
board_members . append ( board_member )
@ -92,7 +94,7 @@ def basic_board_swarm(sample_agents):
name = " TestBoard " ,
agents = sample_agents ,
verbose = False ,
max_loops = 1
max_loops = 1 ,
)
@ -109,7 +111,7 @@ def configured_board_swarm(sample_agents, sample_board_members):
decision_threshold = 0.7 ,
enable_voting = True ,
enable_consensus = True ,
max_workers = 4
max_workers = 4 ,
)
@ -124,7 +126,9 @@ class TestBoardMemberRole:
assert BoardMemberRole . SECRETARY == " secretary "
assert BoardMemberRole . TREASURER == " treasurer "
assert BoardMemberRole . MEMBER == " member "
assert BoardMemberRole . EXECUTIVE_DIRECTOR == " executive_director "
assert (
BoardMemberRole . EXECUTIVE_DIRECTOR == " executive_director "
)
class TestBoardDecisionType :
@ -135,7 +139,9 @@ class TestBoardDecisionType:
assert BoardDecisionType . UNANIMOUS == " unanimous "
assert BoardDecisionType . MAJORITY == " majority "
assert BoardDecisionType . CONSENSUS == " consensus "
assert BoardDecisionType . CHAIRMAN_DECISION == " chairman_decision "
assert (
BoardDecisionType . CHAIRMAN_DECISION == " chairman_decision "
)
class TestBoardMember :
@ -147,19 +153,21 @@ class TestBoardMember:
agent = mock_agent ,
role = BoardMemberRole . CHAIRMAN ,
voting_weight = 1.5 ,
expertise_areas = [ " leadership " , " strategy " ]
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 " ]
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
agent = mock_agent , role = BoardMemberRole . MEMBER
)
assert board_member . voting_weight == 1.0
@ -170,7 +178,7 @@ class TestBoardMember:
board_member = BoardMember (
agent = mock_agent ,
role = BoardMemberRole . MEMBER ,
expertise_areas = None
expertise_areas = None ,
)
assert board_member . expertise_areas == [ ]
@ -186,7 +194,7 @@ class TestBoardOrder:
task = " Test task " ,
priority = 1 ,
deadline = " 2024-01-01 " ,
assigned_by = " Chairman "
assigned_by = " Chairman " ,
)
assert order . agent_name == " TestAgent "
@ -197,10 +205,7 @@ class TestBoardOrder:
def test_board_order_defaults ( self ) :
""" Test board order with default values. """
order = BoardOrder (
agent_name = " TestAgent " ,
task = " Test task "
)
order = BoardOrder ( agent_name = " TestAgent " , task = " Test task " )
assert order . priority == 3
assert order . deadline is None
@ -213,14 +218,14 @@ class TestBoardOrder:
BoardOrder (
agent_name = " TestAgent " ,
task = " Test task " ,
priority = 0 # Invalid priority
priority = 0 , # Invalid priority
)
with pytest . raises ( ValueError ) :
BoardOrder (
agent_name = " TestAgent " ,
task = " Test task " ,
priority = 6 # Invalid priority
priority = 6 , # Invalid priority
)
@ -235,7 +240,7 @@ class TestBoardDecision:
votes_for = 3 ,
votes_against = 1 ,
abstentions = 0 ,
reasoning = " The proposal aligns with our strategic goals "
reasoning = " The proposal aligns with our strategic goals " ,
)
assert decision . decision_type == BoardDecisionType . MAJORITY
@ -243,13 +248,16 @@ class TestBoardDecision:
assert decision . votes_for == 3
assert decision . votes_against == 1
assert decision . abstentions == 0
assert decision . reasoning == " The proposal aligns with our strategic goals "
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 "
decision = " Test decision " ,
)
assert decision . votes_for == 0
@ -265,12 +273,12 @@ class TestBoardSpec:
""" Test creating a board spec. """
orders = [
BoardOrder ( agent_name = " Agent1 " , task = " Task 1 " ) ,
BoardOrder ( agent_name = " Agent2 " , task = " Task 2 " )
BoardOrder ( agent_name = " Agent2 " , task = " Task 2 " ) ,
]
decisions = [
BoardDecision (
decision_type = BoardDecisionType . MAJORITY ,
decision = " Decision 1 "
decision = " Decision 1 " ,
)
]
@ -278,7 +286,7 @@ class TestBoardSpec:
plan = " Test plan " ,
orders = orders ,
decisions = decisions ,
meeting_summary = " Test meeting summary "
meeting_summary = " Test meeting summary " ,
)
assert spec . plan == " Test plan "
@ -288,10 +296,7 @@ class TestBoardSpec:
def test_board_spec_defaults ( self ) :
""" Test board spec with default values. """
spec = BoardSpec (
plan = " Test plan " ,
orders = [ ]
)
spec = BoardSpec ( plan = " Test plan " , orders = [ ] )
assert spec . decisions == [ ]
assert spec . meeting_summary == " "
@ -304,8 +309,7 @@ class TestBoardOfDirectorsSwarmInitialization:
def test_basic_initialization ( self , sample_agents ) :
""" Test basic swarm initialization. """
swarm = BoardOfDirectorsSwarm (
name = " TestSwarm " ,
agents = sample_agents
name = " TestSwarm " , agents = sample_agents
)
assert swarm . name == " TestSwarm "
@ -314,7 +318,9 @@ class TestBoardOfDirectorsSwarmInitialization:
assert swarm . verbose is False
assert swarm . decision_threshold == 0.6
def test_configured_initialization ( self , sample_agents , sample_board_members ) :
def test_configured_initialization (
self , sample_agents , sample_board_members
) :
""" Test configured swarm initialization. """
swarm = BoardOfDirectorsSwarm (
name = " ConfiguredSwarm " ,
@ -326,7 +332,7 @@ class TestBoardOfDirectorsSwarmInitialization:
decision_threshold = 0.8 ,
enable_voting = False ,
enable_consensus = False ,
max_workers = 8
max_workers = 8 ,
)
assert swarm . name == " ConfiguredSwarm "
@ -346,23 +352,41 @@ class TestBoardOfDirectorsSwarmInitialization:
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
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 " ) :
with pytest . raises (
ValueError , match = " No agents found in the swarm "
) :
BoardOfDirectorsSwarm ( agents = [ ] )
def test_initialization_with_invalid_max_loops ( self , sample_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 " ) :
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 ) :
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 )
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 :
@ -373,8 +397,14 @@ class TestBoardOfDirectorsSwarmMethods:
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 )
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. """
@ -412,12 +442,16 @@ class TestBoardOfDirectorsSwarmMethods:
assert " Secretary " in info
assert " expertise " in info
def test_add_board_member ( self , basic_board_swarm , mock_board_member ) :
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 (
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 ) :
@ -428,19 +462,29 @@ class TestBoardOfDirectorsSwarmMethods:
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
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 )
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 " )
not_found = configured_board_swarm . get_board_member (
" NonExistent "
)
assert not_found is None
def test_get_board_summary ( self , configured_board_swarm ) :
@ -462,10 +506,14 @@ class TestBoardOfDirectorsSwarmMethods:
class TestBoardMeetingOperations :
""" Test board meeting operations. """
def test_create_board_meeting_prompt ( self , configured_board_swarm ) :
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 )
prompt = configured_board_swarm . _create_board_meeting_prompt (
task
)
assert task in prompt
assert " BOARD OF DIRECTORS MEETING " in prompt
@ -477,23 +525,33 @@ class TestBoardMeetingOperations:
""" Test board discussion conduction. """
prompt = " Test board meeting prompt "
with patch . object ( configured_board_swarm . board_members [ 0 ] . agent , ' run ' ) as mock_run :
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 )
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 ) :
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 " ) :
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 ) :
def test_parse_board_decisions_valid_json (
self , configured_board_swarm
) :
""" Test parsing valid JSON board decisions. """
valid_json = """
{
@ -520,7 +578,9 @@ class TestBoardMeetingOperations:
}
"""
result = configured_board_swarm . _parse_board_decisions ( valid_json )
result = configured_board_swarm . _parse_board_decisions (
valid_json
)
assert isinstance ( result , BoardSpec )
assert result . plan == " Test plan "
@ -528,33 +588,46 @@ class TestBoardMeetingOperations:
assert len ( result . decisions ) == 1
assert result . meeting_summary == " Test summary "
def test_parse_board_decisions_invalid_json ( self , configured_board_swarm ) :
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 )
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 "
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 :
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 "
meeting_summary = " Test summary " ,
)
result = configured_board_swarm . run_board_meeting ( task )
result = configured_board_swarm . run_board_meeting (
task
)
assert isinstance ( result , BoardSpec )
mock_discuss . assert_called_once ( )
@ -569,17 +642,27 @@ class TestTaskExecution:
agent_name = " Agent1 "
task = " Test task "
with patch . object ( configured_board_swarm . agents [ 0 ] , ' run ' ) as mock_run :
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 )
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 ) :
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 " )
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. """
@ -587,27 +670,36 @@ class TestTaskExecution:
agent_name = " Agent1 " ,
task = " Test order task " ,
priority = 1 ,
assigned_by = " Chairman "
assigned_by = " Chairman " ,
)
with patch . object ( configured_board_swarm , ' _call_single_agent ' ) as mock_call :
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 )
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 "
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 ) ,
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 :
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 )
@ -621,12 +713,16 @@ class TestTaskExecution:
""" Test generating board feedback. """
outputs = [
{ " agent_name " : " Agent1 " , " output " : " Output 1 " } ,
{ " agent_name " : " Agent2 " , " output " : " Output 2 " }
{ " agent_name " : " Agent2 " , " output " : " Output 2 " } ,
]
with patch . object ( configured_board_swarm . board_members [ 0 ] . agent , ' run ' ) as mock_run :
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 )
result = configured_board_swarm . _generate_board_feedback (
outputs
)
assert result == " Board feedback "
mock_run . assert_called_once ( )
@ -636,7 +732,9 @@ class TestTaskExecution:
swarm = BoardOfDirectorsSwarm ( agents = sample_agents )
swarm . board_members = [ ] # Remove all board members
with pytest . raises ( ValueError , match = " No chairman found for feedback " ) :
with pytest . raises (
ValueError , match = " No chairman found for feedback "
) :
swarm . _generate_board_feedback ( [ ] )
@ -647,22 +745,36 @@ class TestStepAndRunMethods:
""" 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 :
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 " ) ] ,
orders = [
BoardOrder (
agent_name = " Agent1 " , task = " Task 1 "
)
] ,
decisions = [ ] ,
meeting_summary = " Test summary "
meeting_summary = " Test summary " ,
)
mock_execute . return_value = [ { " agent_name " : " Agent1 " , " output " : " Result " } ]
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_meeting . assert_called_once_with (
task = task , img = None
)
mock_execute . assert_called_once ( )
mock_feedback . assert_called_once ( )
@ -671,30 +783,44 @@ class TestStepAndRunMethods:
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 :
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 " ) ] ,
orders = [
BoardOrder ( agent_name = " Agent1 " , task = " Task 1 " )
] ,
decisions = [ ] ,
meeting_summary = " Test summary "
meeting_summary = " Test summary " ,
)
mock_execute . return_value = [ { " agent_name " : " Agent1 " , " output " : " Result " } ]
mock_execute . return_value = [
{ " agent_name " : " Agent1 " , " output " : " Result " }
]
result = configured_board_swarm . step ( task )
assert result == [ { " agent_name " : " Agent1 " , " output " : " Result " } ]
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 :
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 )
. run ( task )
assert mock_step . call_count == 2 # max_loops = 2
assert mock_conversation . add . call_count == 2
@ -703,7 +829,7 @@ class TestStepAndRunMethods:
""" Test the async run method. """
task = " Test async run task "
with patch . object ( configured_board_swarm , ' run ' ) as mock_run :
with patch . object ( configured_board_swarm , " run " ) as mock_run :
mock_run . return_value = " Async result "
async def test_async ( ) :
@ -722,9 +848,7 @@ class TestBoardOfDirectorsSwarmIntegration:
def test_full_workflow_integration ( self , sample_agents ) :
""" Test full workflow integration. """
swarm = BoardOfDirectorsSwarm (
agents = sample_agents ,
verbose = False ,
max_loops = 1
agents = sample_agents , verbose = False , max_loops = 1
)
task = " Create a simple report "
@ -761,7 +885,9 @@ class TestBoardOfDirectorsSwarmIntegration:
}
"""
with patch . object ( swarm . board_members [ 0 ] . agent , ' run ' ) as mock_run :
with patch . object (
swarm . board_members [ 0 ] . agent , " run "
) as mock_run :
mock_run . return_value = mock_board_output
result = swarm . run ( task )
@ -777,7 +903,7 @@ class TestBoardOfDirectorsSwarmIntegration:
agent = sample_agents [ 0 ] ,
role = BoardMemberRole . MEMBER ,
voting_weight = 1.0 ,
expertise_areas = [ " testing " ]
expertise_areas = [ " testing " ] ,
)
initial_count = len ( swarm . board_members )
@ -790,7 +916,9 @@ class TestBoardOfDirectorsSwarmIntegration:
assert len ( swarm . board_members ) == initial_count
# Test getting board member
member = swarm . get_board_member ( swarm . board_members [ 0 ] . agent . agent_name )
member = swarm . get_board_member (
swarm . board_members [ 0 ] . agent . agent_name
)
assert member is not None
@ -798,21 +926,33 @@ class TestBoardOfDirectorsSwarmIntegration:
@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 )
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 ) :
@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 )
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 " ] )
@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 )
swarm = BoardOfDirectorsSwarm (
agents = sample_agents , board_model_name = board_model
)
assert swarm . board_model_name == board_model
@ -825,28 +965,50 @@ class TestBoardOfDirectorsSwarmErrorHandling:
with pytest . raises ( ValueError ) :
BoardOfDirectorsSwarm ( agents = [ ] )
def test_board_meeting_error_handling ( self , configured_board_swarm ) :
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 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 " ) :
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 ) :
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 :
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 " )
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 ) :
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 " )
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 )
@ -863,9 +1025,7 @@ class TestBoardOfDirectorsSwarmPerformance:
import time
swarm = BoardOfDirectorsSwarm (
agents = sample_agents ,
max_workers = 3 ,
verbose = False
agents = sample_agents , max_workers = 3 , verbose = False
)
# Create multiple orders
@ -876,15 +1036,21 @@ class TestBoardOfDirectorsSwarmPerformance:
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 } "
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
assert (
execution_time < 1.0
) # Should complete quickly with parallel execution
def test_memory_usage ( self , sample_agents ) :
""" Test memory usage characteristics. """
@ -898,9 +1064,7 @@ class TestBoardOfDirectorsSwarmPerformance:
swarms = [ ]
for i in range ( 5 ) :
swarm = BoardOfDirectorsSwarm (
agents = sample_agents ,
name = f " Swarm { i } " ,
verbose = False
agents = sample_agents , name = f " Swarm { i } " , verbose = False
)
swarms . append ( swarm )
@ -917,49 +1081,69 @@ class TestBoardOfDirectorsSwarmConfiguration:
def test_verbose_configuration ( self , sample_agents ) :
""" Test verbose configuration. """
swarm = BoardOfDirectorsSwarm ( agents = sample_agents , verbose = True )
swarm = BoardOfDirectorsSwarm (
agents = sample_agents , verbose = True
)
assert swarm . verbose is True
swarm = BoardOfDirectorsSwarm ( agents = sample_agents , verbose = False )
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 )
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 )
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 )
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 )
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 )
swarm = BoardOfDirectorsSwarm (
agents = sample_agents , enable_voting = True
)
assert swarm . enable_voting is True
swarm = BoardOfDirectorsSwarm ( agents = sample_agents , enable_voting = False )
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 )
swarm = BoardOfDirectorsSwarm (
agents = sample_agents , enable_consensus = True
)
assert swarm . enable_consensus is True
swarm = BoardOfDirectorsSwarm ( agents = sample_agents , enable_consensus = False )
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 "
reason = " OpenAI API key not available " ,
)
class TestBoardOfDirectorsSwarmRealIntegration :
""" Real integration tests for BoardOfDirectorsSwarm. """
@ -972,20 +1156,18 @@ class TestBoardOfDirectorsSwarmRealIntegration:
agent_name = " Researcher " ,
agent_description = " Research analyst " ,
model_name = " gpt-4o-mini " ,
max_loops = 1
max_loops = 1 ,
) ,
Agent (
agent_name = " Writer " ,
agent_description = " Content writer " ,
model_name = " gpt-4o-mini " ,
max_loops = 1
)
max_loops = 1 ,
) ,
]
swarm = BoardOfDirectorsSwarm (
agents = agents ,
verbose = False ,
max_loops = 1
agents = agents , verbose = False , max_loops = 1
)
task = " Create a brief market analysis report "
@ -1003,7 +1185,7 @@ class TestBoardOfDirectorsSwarmRealIntegration:
agent_name = " TestAgent " ,
agent_description = " Test agent " ,
model_name = " gpt-4o-mini " ,
max_loops = 1
max_loops = 1 ,
)
]
Kye Gomez
4 weeks ago