@ -1,10 +1,8 @@
import os
import json
import time
from datetime import datetime
from typing import List , Dict , Any , Callable
import requests
from dotenv import load_dotenv
# Basic Imports for Swarms
@ -20,16 +18,19 @@ from swarms.structs import (
MajorityVoting ,
SwarmRouter ,
RoundRobinSwarm ,
InteractiveGroupChat
InteractiveGroupChat ,
)
# Import swarms not in __init__.py directly
from swarms . structs . hiearchical_swarm import HierarchicalSwarm
from swarms . structs . tree_swarm import ForestSwarm , Tree , TreeAgent
from swarms . tools . base_tool import BaseTool
# Setup Logging
from loguru import logger
logger . add ( " test_runs/test_failures.log " , rotation = " 10 MB " , level = " ERROR " )
logger . add (
" test_runs/test_failures.log " , rotation = " 10 MB " , level = " ERROR "
)
# Load environment variables
load_dotenv ( )
@ -49,18 +50,24 @@ API_KEY = os.getenv("OPENAI_API_KEY")
# --- Helper Functions ---
def generate_timestamp ( ) - > str :
""" Generate a timestamp string for filenames """
return datetime . now ( ) . strftime ( " % Y % m %d _ % H % M % S " )
def write_markdown_report ( results : List [ Dict [ str , Any ] ] , filename : str ) :
def write_markdown_report (
results : List [ Dict [ str , Any ] ] , filename : str
) :
""" Write test results to a markdown file """
if not os . path . exists ( " test_runs " ) :
os . makedirs ( " test_runs " )
with open ( f " test_runs/ { filename } .md " , " w " ) as f :
f . write ( " # Swarms Comprehensive Test Report \n \n " )
f . write ( f " Test Run: { datetime . now ( ) . strftime ( ' % Y- % m- %d % H: % M: % S ' ) } \n \n " )
f . write (
f " Test Run: { datetime . now ( ) . strftime ( ' % Y- % m- %d % H: % M: % S ' ) } \n \n "
)
total = len ( results )
passed = sum ( 1 for r in results if r [ " status " ] == " passed " )
@ -80,16 +87,23 @@ def write_markdown_report(results: List[Dict[str, Any]], filename: str):
f . write ( " Response: \n ```json \n " )
response_str = result [ " response " ]
try :
response_json = json . loads ( response_str ) if isinstance ( response_str , str ) else response_str
response_json = (
json . loads ( response_str )
if isinstance ( response_str , str )
else response_str
)
f . write ( json . dumps ( response_json , indent = 2 ) )
except ( json . JSONDecodeError , TypeError ) :
f . write ( str ( response_str ) )
f . write ( " \n ``` \n \n " )
if result . get ( " error " ) :
f . write ( f " **Error:** \n ``` \n { result [ ' error ' ] } \n ``` \n \n " )
f . write (
f " **Error:** \n ``` \n { result [ ' error ' ] } \n ``` \n \n "
)
f . write ( " --- \n \n " )
# def create_github_issue(test_result: Dict[str, Any]) -> Dict[str, Any]:
# """Create a GitHub issue for a failed test"""
# if not all([GITHUB_TOKEN, GITHUB_REPO_OWNER, GITHUB_REPO_NAME]):
@ -141,43 +155,66 @@ def write_markdown_report(results: List[Dict[str, Any]], filename: str):
# logger.error(f"Failed to create GitHub issue: {e.response.text if e.response else str(e)}")
# return None
def create_test_agent ( name : str , system_prompt : str = None , model_name : str = " gpt-4o-mini " , tools : List [ Callable ] = None , * * kwargs ) - > Agent :
def create_test_agent (
name : str ,
system_prompt : str = None ,
model_name : str = " gpt-4o-mini " ,
tools : List [ Callable ] = None ,
* * kwargs ,
) - > Agent :
""" Create a properly configured test agent with error handling """
try :
return Agent (
agent_name = name ,
system_prompt = system_prompt or f " You are { name } , a helpful AI assistant. " ,
system_prompt = system_prompt
or f " You are { name } , a helpful AI assistant. " ,
model_name = model_name , # Use mini model for faster/cheaper testing
max_loops = 1 ,
max_tokens = 200 ,
tools = tools ,
* * kwargs
* * kwargs ,
)
except Exception as e :
logger . error ( f " Failed to create agent { name } : { e } " )
raise
# --- Basic Agent Tests ---
def test_basic_agent_functionality ( ) :
""" Test basic agent creation and execution """
agent = create_test_agent ( " BasicAgent " )
response = agent . run ( " Say hello and explain what you are. " )
assert isinstance ( response , str ) and len ( response ) > 0
return { " test_name " : " test_basic_agent_functionality " , " status " : " passed " , " response " : " Agent created and responded successfully " }
return {
" test_name " : " test_basic_agent_functionality " ,
" status " : " passed " ,
" response " : " Agent created and responded successfully " ,
}
def test_agent_with_custom_prompt ( ) :
""" Test agent with custom system prompt """
custom_prompt = " You are a mathematician who only responds with numbers and mathematical expressions. "
agent = create_test_agent ( " MathAgent " , system_prompt = custom_prompt )
agent = create_test_agent (
" MathAgent " , system_prompt = custom_prompt
)
response = agent . run ( " What is 2+2? " )
assert isinstance ( response , str ) and len ( response ) > 0
return { " test_name " : " test_agent_with_custom_prompt " , " status " : " passed " , " response " : response [ : 100 ] }
return {
" test_name " : " test_agent_with_custom_prompt " ,
" status " : " passed " ,
" response " : response [ : 100 ] ,
}
def test_tool_execution_with_agent ( ) :
""" Test agent ' s ability to use tools """
def simple_calculator ( a : int , b : int ) - > int :
""" Add two numbers together """
return a + b
@ -189,118 +226,218 @@ def test_tool_execution_with_agent():
agent = create_test_agent (
" ToolAgent " ,
system_prompt = " You are a helpful assistant that can use tools to help users. " ,
tools = [ simple_calculator , get_weather ]
tools = [ simple_calculator , get_weather ] ,
)
response = agent . run (
" What ' s 5 + 7 and what ' s the weather like in New York? "
)
response = agent . run ( " What ' s 5 + 7 and what ' s the weather like in New York? " )
assert isinstance ( response , str ) and len ( response ) > 0
return { " test_name " : " test_tool_execution_with_agent " , " status " : " passed " , " response " : " Tool execution completed " }
return {
" test_name " : " test_tool_execution_with_agent " ,
" status " : " passed " ,
" response " : " Tool execution completed " ,
}
# --- Multi-Modal Tests ---
def test_multimodal_execution ( ) :
""" Test agent ' s ability to process images """
agent = create_test_agent ( " VisionAgent " , model_name = " gpt-4o " , multi_modal = True )
agent = create_test_agent (
" VisionAgent " , model_name = " gpt-4o " , multi_modal = True
)
try :
# Check if test images exist, if not skip the test
if os . path . exists ( " tests/test_data/image1.jpg " ) :
response = agent . run ( " Describe this image. " , img = " tests/test_data/image1.jpg " )
response = agent . run (
" Describe this image. " ,
img = " tests/test_data/image1.jpg " ,
)
assert isinstance ( response , str ) and len ( response ) > 0
else :
logger . warning ( " Test image not found, skipping multimodal test " )
logger . warning (
" Test image not found, skipping multimodal test "
)
response = " Test skipped - no test image available "
return { " test_name " : " test_multimodal_execution " , " status " : " passed " , " response " : " Multimodal response received " }
return {
" test_name " : " test_multimodal_execution " ,
" status " : " passed " ,
" response " : " Multimodal response received " ,
}
except Exception as e :
logger . warning ( f " Multimodal test failed: { e } " )
return { " test_name " : " test_multimodal_execution " , " status " : " passed " , " response " : " Multimodal test skipped due to missing dependencies " }
return {
" test_name " : " test_multimodal_execution " ,
" status " : " passed " ,
" response " : " Multimodal test skipped due to missing dependencies " ,
}
# --- Workflow Tests ---
def test_sequential_workflow ( ) :
""" Test SequentialWorkflow with multiple agents """
agents = [
create_test_agent ( " ResearchAgent " , " You are a research specialist who gathers information. " ) ,
create_test_agent ( " AnalysisAgent " , " You are an analyst who analyzes information and provides insights. " ) ,
create_test_agent ( " WriterAgent " , " You are a writer who creates clear, concise summaries. " )
create_test_agent (
" ResearchAgent " ,
" You are a research specialist who gathers information. " ,
) ,
create_test_agent (
" AnalysisAgent " ,
" You are an analyst who analyzes information and provides insights. " ,
) ,
create_test_agent (
" WriterAgent " ,
" You are a writer who creates clear, concise summaries. " ,
) ,
]
workflow = SequentialWorkflow (
name = " research-analysis-workflow " ,
agents = agents ,
max_loops = 1
name = " research-analysis-workflow " , agents = agents , max_loops = 1
)
try :
response = workflow . run ( " Research and analyze the benefits of renewable energy, then write a brief summary. " )
logger . info ( f " SequentialWorkflow response type: { type ( response ) } " )
response = workflow . run (
" Research and analyze the benefits of renewable energy, then write a brief summary. "
)
logger . info (
f " SequentialWorkflow response type: { type ( response ) } "
)
# SequentialWorkflow returns conversation history
assert response is not None
return { " test_name " : " test_sequential_workflow " , " status " : " passed " , " response " : " Sequential workflow completed " }
return {
" test_name " : " test_sequential_workflow " ,
" status " : " passed " ,
" response " : " Sequential workflow completed " ,
}
except Exception as e :
logger . error ( f " SequentialWorkflow test failed with exception: { e } " )
return { " test_name " : " test_sequential_workflow " , " status " : " failed " , " error " : str ( e ) }
logger . error (
f " SequentialWorkflow test failed with exception: { e } "
)
return {
" test_name " : " test_sequential_workflow " ,
" status " : " failed " ,
" error " : str ( e ) ,
}
def test_concurrent_workflow ( ) :
""" Test ConcurrentWorkflow with multiple agents """
agents = [
create_test_agent ( " TechAnalyst " , " You are a technology analyst who focuses on tech trends. " ) ,
create_test_agent ( " MarketAnalyst " , " You are a market analyst who focuses on market conditions. " )
create_test_agent (
" TechAnalyst " ,
" You are a technology analyst who focuses on tech trends. " ,
) ,
create_test_agent (
" MarketAnalyst " ,
" You are a market analyst who focuses on market conditions. " ,
) ,
]
workflow = ConcurrentWorkflow (
name = " concurrent-analysis " ,
agents = agents ,
max_loops = 1
name = " concurrent-analysis " , agents = agents , max_loops = 1
)
try :
response = workflow . run ( " Analyze the current state of AI technology and its market impact. " )
logger . info ( f " ConcurrentWorkflow response type: { type ( response ) } " )
response = workflow . run (
" Analyze the current state of AI technology and its market impact. "
)
logger . info (
f " ConcurrentWorkflow response type: { type ( response ) } "
)
assert response is not None
return { " test_name " : " test_concurrent_workflow " , " status " : " passed " , " response " : " Concurrent workflow completed " }
return {
" test_name " : " test_concurrent_workflow " ,
" status " : " passed " ,
" response " : " Concurrent workflow completed " ,
}
except Exception as e :
logger . error ( f " ConcurrentWorkflow test failed with exception: { e } " )
return { " test_name " : " test_concurrent_workflow " , " status " : " failed " , " error " : str ( e ) }
logger . error (
f " ConcurrentWorkflow test failed with exception: { e } "
)
return {
" test_name " : " test_concurrent_workflow " ,
" status " : " failed " ,
" error " : str ( e ) ,
}
# --- Advanced Swarm Tests ---
def test_agent_rearrange ( ) :
""" Test AgentRearrange dynamic workflow """
agents = [
create_test_agent ( " Researcher " , " You are a researcher who gathers information. " ) ,
create_test_agent ( " Analyst " , " You are an analyst who analyzes information. " ) ,
create_test_agent ( " Writer " , " You are a writer who creates final reports. " )
create_test_agent (
" Researcher " ,
" You are a researcher who gathers information. " ,
) ,
create_test_agent (
" Analyst " , " You are an analyst who analyzes information. "
) ,
create_test_agent (
" Writer " , " You are a writer who creates final reports. "
) ,
]
flow = " Researcher -> Analyst -> Writer "
swarm = AgentRearrange ( agents = agents , flow = flow , max_loops = 1 )
response = swarm . run ( " Research renewable energy, analyze the benefits, and write a summary. " )
response = swarm . run (
" Research renewable energy, analyze the benefits, and write a summary. "
)
assert response is not None
return { " test_name " : " test_agent_rearrange " , " status " : " passed " , " response " : " AgentRearrange completed " }
return {
" test_name " : " test_agent_rearrange " ,
" status " : " passed " ,
" response " : " AgentRearrange completed " ,
}
def test_mixture_of_agents ( ) :
""" Test MixtureOfAgents collaboration """
agents = [
create_test_agent ( " TechExpert " , " You are a technology expert. " ) ,
create_test_agent ( " BusinessAnalyst " , " You are a business analyst. " ) ,
create_test_agent ( " Strategist " , " You are a strategic planner. " )
create_test_agent (
" TechExpert " , " You are a technology expert. "
) ,
create_test_agent (
" BusinessAnalyst " , " You are a business analyst. "
) ,
create_test_agent (
" Strategist " , " You are a strategic planner. "
) ,
]
swarm = MixtureOfAgents ( agents = agents , max_loops = 1 )
response = swarm . run ( " Analyze the impact of AI on modern businesses. " )
response = swarm . run (
" Analyze the impact of AI on modern businesses. "
)
assert response is not None
return { " test_name " : " test_mixture_of_agents " , " status " : " passed " , " response " : " MixtureOfAgents completed " }
return {
" test_name " : " test_mixture_of_agents " ,
" status " : " passed " ,
" response " : " MixtureOfAgents completed " ,
}
def test_spreadsheet_swarm ( ) :
""" Test SpreadSheetSwarm for data processing """
agents = [
create_test_agent ( " DataProcessor1 " , " You process and analyze numerical data. " ) ,
create_test_agent ( " DataProcessor2 " , " You perform calculations and provide insights. " )
create_test_agent (
" DataProcessor1 " ,
" You process and analyze numerical data. " ,
) ,
create_test_agent (
" DataProcessor2 " ,
" You perform calculations and provide insights. " ,
) ,
]
swarm = SpreadSheetSwarm (
@ -308,24 +445,39 @@ def test_spreadsheet_swarm():
description = " A swarm for processing data " ,
agents = agents ,
max_loops = 1 ,
autosave_on = False
autosave_on = False ,
)
response = swarm . run ( " Calculate the sum of 25 + 75 and provide analysis. " )
response = swarm . run (
" Calculate the sum of 25 + 75 and provide analysis. "
)
assert response is not None
return { " test_name " : " test_spreadsheet_swarm " , " status " : " passed " , " response " : " SpreadSheetSwarm completed " }
return {
" test_name " : " test_spreadsheet_swarm " ,
" status " : " passed " ,
" response " : " SpreadSheetSwarm completed " ,
}
def test_hierarchical_swarm ( ) :
""" Test HierarchicalSwarm structure """
try :
from swarms . utils . function_caller_model import OpenAIFunctionCaller
from swarms . utils . function_caller_model import (
OpenAIFunctionCaller ,
)
from swarms . structs . hiearchical_swarm import SwarmSpec
# Create worker agents
workers = [
create_test_agent ( " Worker1 " , " You are Worker1 who handles research tasks and data gathering. " ) ,
create_test_agent ( " Worker2 " , " You are Worker2 who handles analysis tasks and reporting. " )
create_test_agent (
" Worker1 " ,
" You are Worker1 who handles research tasks and data gathering. " ,
) ,
create_test_agent (
" Worker2 " ,
" You are Worker2 who handles analysis tasks and reporting. " ,
) ,
]
# Create director agent with explicit knowledge of available agents
@ -344,63 +496,107 @@ def test_hierarchical_swarm():
" 4. Provide actionable task descriptions "
) ,
temperature = 0.1 ,
max_tokens = 1000
max_tokens = 1000 ,
)
swarm = HierarchicalSwarm (
description = " A test hierarchical swarm for task delegation " ,
director = director ,
agents = workers ,
max_loops = 1
max_loops = 1 ,
)
response = swarm . run ( " Research current team meeting best practices and analyze them to create recommendations. " )
response = swarm . run (
" Research current team meeting best practices and analyze them to create recommendations. "
)
assert response is not None
return { " test_name " : " test_hierarchical_swarm " , " status " : " passed " , " response " : " HierarchicalSwarm completed " }
return {
" test_name " : " test_hierarchical_swarm " ,
" status " : " passed " ,
" response " : " HierarchicalSwarm completed " ,
}
except ImportError as e :
logger . warning ( f " HierarchicalSwarm test skipped due to missing dependencies: { e } " )
return { " test_name " : " test_hierarchical_swarm " , " status " : " passed " , " response " : " Test skipped due to missing dependencies " }
logger . warning (
f " HierarchicalSwarm test skipped due to missing dependencies: { e } "
)
return {
" test_name " : " test_hierarchical_swarm " ,
" status " : " passed " ,
" response " : " Test skipped due to missing dependencies " ,
}
def test_majority_voting ( ) :
""" Test MajorityVoting consensus mechanism """
agents = [
create_test_agent ( " Judge1 " , " You are a judge who evaluates options carefully. " ) ,
create_test_agent ( " Judge2 " , " You are a judge who provides thorough analysis. " ) ,
create_test_agent ( " Judge3 " , " You are a judge who considers all perspectives. " )
create_test_agent (
" Judge1 " ,
" You are a judge who evaluates options carefully. " ,
) ,
create_test_agent (
" Judge2 " ,
" You are a judge who provides thorough analysis. " ,
) ,
create_test_agent (
" Judge3 " ,
" You are a judge who considers all perspectives. " ,
) ,
]
swarm = MajorityVoting ( agents = agents )
response = swarm . run ( " Should companies invest more in renewable energy? Provide YES or NO with reasoning. " )
response = swarm . run (
" Should companies invest more in renewable energy? Provide YES or NO with reasoning. "
)
assert response is not None
return { " test_name " : " test_majority_voting " , " status " : " passed " , " response " : " MajorityVoting completed " }
return {
" test_name " : " test_majority_voting " ,
" status " : " passed " ,
" response " : " MajorityVoting completed " ,
}
def test_round_robin_swarm ( ) :
""" Test RoundRobinSwarm task distribution """
agents = [
create_test_agent ( " Agent1 " , " You handle counting tasks. " ) ,
create_test_agent ( " Agent2 " , " You handle color-related tasks. " ) ,
create_test_agent ( " Agent3 " , " You handle animal-related tasks. " )
create_test_agent (
" Agent2 " , " You handle color-related tasks. "
) ,
create_test_agent (
" Agent3 " , " You handle animal-related tasks. "
) ,
]
swarm = RoundRobinSwarm ( agents = agents )
tasks = [
" Count from 1 to 5 " ,
" Name 3 primary colors " ,
" List 3 common pets "
" List 3 common pets " ,
]
response = swarm . run ( tasks )
assert response is not None
return { " test_name " : " test_round_robin_swarm " , " status " : " passed " , " response " : " RoundRobinSwarm completed " }
return {
" test_name " : " test_round_robin_swarm " ,
" status " : " passed " ,
" response " : " RoundRobinSwarm completed " ,
}
def test_swarm_router ( ) :
""" Test SwarmRouter dynamic routing """
agents = [
create_test_agent ( " DataAnalyst " , " You specialize in data analysis and statistics. " ) ,
create_test_agent ( " ReportWriter " , " You specialize in writing clear, professional reports. " )
create_test_agent (
" DataAnalyst " ,
" You specialize in data analysis and statistics. " ,
) ,
create_test_agent (
" ReportWriter " ,
" You specialize in writing clear, professional reports. " ,
) ,
]
router = SwarmRouter (
@ -408,65 +604,111 @@ def test_swarm_router():
description = " Routes analysis and reporting tasks to appropriate agents " ,
agents = agents ,
swarm_type = " SequentialWorkflow " ,
max_loops = 1
max_loops = 1 ,
)
response = router . run ( " Analyze customer satisfaction data and write a summary report. " )
response = router . run (
" Analyze customer satisfaction data and write a summary report. "
)
assert response is not None
return { " test_name " : " test_swarm_router " , " status " : " passed " , " response " : " SwarmRouter completed " }
return {
" test_name " : " test_swarm_router " ,
" status " : " passed " ,
" response " : " SwarmRouter completed " ,
}
def test_groupchat ( ) :
""" Test GroupChat functionality """
agents = [
create_test_agent ( " Moderator " , " You are a discussion moderator who guides conversations. " ) ,
create_test_agent ( " Expert1 " , " You are a subject matter expert who provides insights. " ) ,
create_test_agent ( " Expert2 " , " You are another expert who offers different perspectives. " )
create_test_agent (
" Moderator " ,
" You are a discussion moderator who guides conversations. " ,
) ,
create_test_agent (
" Expert1 " ,
" You are a subject matter expert who provides insights. " ,
) ,
create_test_agent (
" Expert2 " ,
" You are another expert who offers different perspectives. " ,
) ,
]
groupchat = GroupChat (
agents = agents ,
messages = [ ] ,
max_round = 2
)
groupchat = GroupChat ( agents = agents , messages = [ ] , max_round = 2 )
# GroupChat requires a different interface than other swarms
response = groupchat . run ( " Discuss the benefits and challenges of remote work. " )
response = groupchat . run (
" Discuss the benefits and challenges of remote work. "
)
assert response is not None
return { " test_name " : " test_groupchat " , " status " : " passed " , " response " : " GroupChat completed " }
return {
" test_name " : " test_groupchat " ,
" status " : " passed " ,
" response " : " GroupChat completed " ,
}
def test_multi_agent_router ( ) :
""" Test MultiAgentRouter functionality """
agents = [
create_test_agent ( " TechAgent " , " You handle technology-related queries. " ) ,
create_test_agent ( " BusinessAgent " , " You handle business-related queries. " ) ,
create_test_agent ( " GeneralAgent " , " You handle general queries. " )
create_test_agent (
" TechAgent " , " You handle technology-related queries. "
) ,
create_test_agent (
" BusinessAgent " , " You handle business-related queries. "
) ,
create_test_agent (
" GeneralAgent " , " You handle general queries. "
) ,
]
router = MultiAgentRouter ( agents = agents )
response = router . run ( " What are the latest trends in business technology? " )
response = router . run (
" What are the latest trends in business technology? "
)
assert response is not None
return { " test_name " : " test_multi_agent_router " , " status " : " passed " , " response " : " MultiAgentRouter completed " }
return {
" test_name " : " test_multi_agent_router " ,
" status " : " passed " ,
" response " : " MultiAgentRouter completed " ,
}
def test_interactive_groupchat ( ) :
""" Test InteractiveGroupChat functionality """
agents = [
create_test_agent ( " Facilitator " , " You facilitate group discussions. " ) ,
create_test_agent ( " Participant1 " , " You are an active discussion participant. " ) ,
create_test_agent ( " Participant2 " , " You provide thoughtful contributions to discussions. " )
create_test_agent (
" Facilitator " , " You facilitate group discussions. "
) ,
create_test_agent (
" Participant1 " ,
" You are an active discussion participant. " ,
) ,
create_test_agent (
" Participant2 " ,
" You provide thoughtful contributions to discussions. " ,
) ,
]
interactive_chat = InteractiveGroupChat (
agents = agents ,
max_loops = 2
agents = agents , max_loops = 2
)
response = interactive_chat . run ( " Let ' s discuss the future of artificial intelligence. " )
response = interactive_chat . run (
" Let ' s discuss the future of artificial intelligence. "
)
assert response is not None
return { " test_name " : " test_interactive_groupchat " , " status " : " passed " , " response " : " InteractiveGroupChat completed " }
return {
" test_name " : " test_interactive_groupchat " ,
" status " : " passed " ,
" response " : " InteractiveGroupChat completed " ,
}
def test_forest_swarm ( ) :
""" Test ForestSwarm tree-based structure """
@ -475,23 +717,23 @@ def test_forest_swarm():
tree1_agents = [
TreeAgent (
system_prompt = " You analyze market trends " ,
agent_name = " Market-Analyst "
agent_name = " Market-Analyst " ,
) ,
TreeAgent (
system_prompt = " You provide financial insights " ,
agent_name = " Financial-Advisor "
)
agent_name = " Financial-Advisor " ,
) ,
]
tree2_agents = [
TreeAgent (
system_prompt = " You assess investment risks " ,
agent_name = " Risk-Assessor "
agent_name = " Risk-Assessor " ,
) ,
TreeAgent (
system_prompt = " You create investment strategies " ,
agent_name = " Strategy-Planner "
)
agent_name = " Strategy-Planner " ,
) ,
]
# Create trees
@ -501,29 +743,50 @@ def test_forest_swarm():
# Create ForestSwarm
forest = ForestSwarm ( trees = [ tree1 , tree2 ] )
response = forest . run ( " Analyze the current market and develop an investment strategy. " )
response = forest . run (
" Analyze the current market and develop an investment strategy. "
)
assert response is not None
return { " test_name " : " test_forest_swarm " , " status " : " passed " , " response " : " ForestSwarm completed " }
return {
" test_name " : " test_forest_swarm " ,
" status " : " passed " ,
" response " : " ForestSwarm completed " ,
}
except Exception as e :
logger . error ( f " ForestSwarm test failed: { e } " )
return { " test_name " : " test_forest_swarm " , " status " : " failed " , " error " : str ( e ) }
return {
" test_name " : " test_forest_swarm " ,
" status " : " failed " ,
" error " : str ( e ) ,
}
# --- Performance & Features Tests ---
def test_streaming_mode ( ) :
""" Test streaming response generation """
agent = create_test_agent ( " StreamingAgent " , streaming_on = True )
response = agent . run ( " Tell me a very short story about technology. " )
response = agent . run (
" Tell me a very short story about technology. "
)
assert response is not None
return { " test_name " : " test_streaming_mode " , " status " : " passed " , " response " : " Streaming mode tested " }
return {
" test_name " : " test_streaming_mode " ,
" status " : " passed " ,
" response " : " Streaming mode tested " ,
}
def test_agent_memory_persistence ( ) :
""" Test agent memory functionality """
agent = create_test_agent ( " MemoryAgent " ,
agent = create_test_agent (
" MemoryAgent " ,
system_prompt = " You remember information from previous conversations. " ,
return_history = True )
return_history = True ,
)
# First interaction
response1 = agent . run ( " My name is Alice. Please remember this. " )
@ -531,7 +794,12 @@ def test_agent_memory_persistence():
response2 = agent . run ( " What is my name? " )
assert response1 is not None and response2 is not None
return { " test_name " : " test_agent_memory_persistence " , " status " : " passed " , " response " : " Memory persistence tested " }
return {
" test_name " : " test_agent_memory_persistence " ,
" status " : " passed " ,
" response " : " Memory persistence tested " ,
}
def test_error_handling ( ) :
""" Test agent error handling with various inputs """
@ -546,46 +814,78 @@ def test_error_handling():
response = agent . run ( " Hi " )
assert response is not None
return { " test_name " : " test_error_handling " , " status " : " passed " , " response " : " Error handling tests passed " }
return {
" test_name " : " test_error_handling " ,
" status " : " passed " ,
" response " : " Error handling tests passed " ,
}
except Exception as e :
return { " test_name " : " test_error_handling " , " status " : " failed " , " error " : str ( e ) }
return {
" test_name " : " test_error_handling " ,
" status " : " failed " ,
" error " : str ( e ) ,
}
# --- Integration Tests ---
def test_complex_workflow_integration ( ) :
""" Test complex multi-agent workflow integration """
try :
# Create specialized agents
researcher = create_test_agent ( " Researcher " , " You research topics thoroughly and gather information. " )
analyst = create_test_agent ( " Analyst " , " You analyze research data and provide insights. " )
writer = create_test_agent ( " Writer " , " You write clear, comprehensive summaries. " )
researcher = create_test_agent (
" Researcher " ,
" You research topics thoroughly and gather information. " ,
)
analyst = create_test_agent (
" Analyst " ,
" You analyze research data and provide insights. " ,
)
writer = create_test_agent (
" Writer " , " You write clear, comprehensive summaries. "
)
# Test SequentialWorkflow
sequential = SequentialWorkflow (
name = " research-workflow " ,
agents = [ researcher , analyst , writer ] ,
max_loops = 1
max_loops = 1 ,
)
seq_response = sequential . run ( " Research AI trends, analyze them, and write a summary. " )
seq_response = sequential . run (
" Research AI trends, analyze them, and write a summary. "
)
# Test ConcurrentWorkflow
concurrent = ConcurrentWorkflow (
name = " parallel-analysis " ,
agents = [ researcher , analyst ] ,
max_loops = 1
max_loops = 1 ,
)
conc_response = concurrent . run ( " What are the benefits and challenges of AI? " )
conc_response = concurrent . run (
" What are the benefits and challenges of AI? "
)
assert seq_response is not None and conc_response is not None
return { " test_name " : " test_complex_workflow_integration " , " status " : " passed " , " response " : " Complex workflow integration completed " }
return {
" test_name " : " test_complex_workflow_integration " ,
" status " : " passed " ,
" response " : " Complex workflow integration completed " ,
}
except Exception as e :
logger . error ( f " Complex workflow integration test failed: { e } " )
return { " test_name " : " test_complex_workflow_integration " , " status " : " failed " , " error " : str ( e ) }
return {
" test_name " : " test_complex_workflow_integration " ,
" status " : " failed " ,
" error " : str ( e ) ,
}
# --- Test Orchestrator ---
def run_all_tests ( ) :
""" Run all tests and generate a comprehensive report """
logger . info ( " Starting Enhanced Swarms Comprehensive Test Suite " )
@ -595,14 +895,11 @@ def run_all_tests():
test_basic_agent_functionality ,
test_agent_with_custom_prompt ,
test_tool_execution_with_agent ,
# Multi-Modal Tests
test_multimodal_execution ,
# Workflow Tests
test_sequential_workflow ,
test_concurrent_workflow ,
# Advanced Swarm Tests
test_agent_rearrange ,
test_mixture_of_agents ,
@ -615,12 +912,10 @@ def run_all_tests():
test_multi_agent_router ,
# test_interactive_groupchat,
# test_forest_swarm,
# Performance & Features
test_streaming_mode ,
test_agent_memory_persistence ,
test_error_handling ,
# Integration Tests
test_complex_workflow_integration ,
]
@ -639,30 +934,39 @@ def run_all_tests():
" test_name " : test_name ,
" status " : " failed " ,
" error " : str ( e ) ,
" response " : " Test execution failed "
" response " : " Test execution failed " ,
}
results . append ( error_details )
# create_github_issue(error_details) # Uncomment to enable GitHub issue creation
timestamp = generate_timestamp ( )
write_markdown_report ( results , f " comprehensive_test_report_ { timestamp } " )
write_markdown_report (
results , f " comprehensive_test_report_ { timestamp } "
)
# Summary
total_tests = len ( results )
passed_tests = sum ( 1 for r in results if r [ ' status ' ] == ' passed ' )
passed_tests = sum ( 1 for r in results if r [ " status " ] == " passed " )
failed_tests = total_tests - passed_tests
logger . info ( f " Test Summary: { passed_tests } / { total_tests } passed ( { ( passed_tests / total_tests ) * 100 : .1f } %) " )
logger . info (
f " Test Summary: { passed_tests } / { total_tests } passed ( { ( passed_tests / total_tests ) * 100 : .1f } %) "
)
if failed_tests > 0 :
logger . error ( f " { failed_tests } tests failed. Check the report and logs. " )
logger . error (
f " { failed_tests } tests failed. Check the report and logs. "
)
exit ( 1 )
else :
logger . success ( " All tests passed successfully! " )
if __name__ == " __main__ " :
if not API_KEY :
logger . error ( " OPENAI_API_KEY environment variable not set. Aborting tests. " )
logger . error (
" OPENAI_API_KEY environment variable not set. Aborting tests. "
)
exit ( 1 )
else :
run_all_tests ( )
Kye Gomez
2 days ago