diff --git a/tests/structs/test_board_of_directors_swarm.py b/tests/structs/test_board_of_directors_swarm.py new file mode 100644 index 00000000..b87e563c --- /dev/null +++ b/tests/structs/test_board_of_directors_swarm.py @@ -0,0 +1,1020 @@ +""" +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"]) \ No newline at end of file