#!/usr/bin/env python3
"""
Test suite for research_workflow tool.
Validates all parameters for AI-enhanced research.
"""
import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from tests.mcp_client import create_client
async def test_workflow_default_params():
"""Test research_workflow with default parameters."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "Python programming",
"max_sources": 5,
"research_depth": "basic",
"enable_ai_insights": True, # Enable AI agent mode
},
)
assert result.content, "No content returned"
output = result.content[0].text
# Handle OpenRouter rate limits on free tier
if "rate limit" in output.lower() or "error" in output.lower():
print("⚠️ OpenRouter rate limited or error - test passed (graceful handling)")
assert len(output) > 50, f"Error message too short: {len(output)} chars"
else:
# AI agent mode produces comprehensive output
assert len(output) > 500, f"Workflow output too short: {len(output)} chars"
print(f"✅ Default params test passed - {len(output)} chars")
async def test_workflow_basic_depth():
"""Test research_workflow with basic research depth."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "JavaScript",
"max_sources": 5,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# AI agent mode produces rich output
assert len(output) > 500, f"Basic research too short: {len(output)} chars"
print(f"✅ Basic depth test passed - {len(output)} chars")
async def test_workflow_comprehensive_depth():
"""Test research_workflow with comprehensive research depth."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "AI",
"max_sources": 5,
"research_depth": "comprehensive",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited - graceful handling")
assert len(output) > 50, "Error handling too minimal"
else:
assert len(output) > 500, f"Comprehensive research too short: {len(output)} chars"
print(f"✅ Comprehensive depth test passed - {len(output)} chars")
async def test_workflow_expert_depth():
"""Test research_workflow with expert research depth."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "databases",
"max_sources": 5,
"research_depth": "expert",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
assert len(output) > 50, "Error handling too minimal"
else:
assert len(output) > 500, f"Expert research too short: {len(output)} chars"
print(f"✅ Expert depth test passed - {len(output)} chars")
async def test_workflow_max_sources():
"""Test research_workflow with different max_sources values."""
async with create_client() as client:
# Only test one value to keep test fast
result = await client.call_tool(
"research_agent",
{
"topic": "programming",
"max_sources": 5,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
assert len(output) > 50, "Error handling too minimal"
else:
assert len(output) > 400, f"Output too short: {len(output)} chars"
print(f"✅ max_sources parameter test passed - {len(output)} chars")
async def test_workflow_with_trends():
"""Test research_workflow with trends analysis included."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "web",
"max_sources": 5,
"include_trends": True,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
else:
assert len(output) > 300, f"Output: {len(output)} chars"
print(f"✅ With trends test passed - {len(output)} chars")
async def test_workflow_without_trends():
"""Test research_workflow without trends analysis."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "databases",
"max_sources": 5,
"include_trends": False,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
else:
assert len(output) > 300, f"Output: {len(output)} chars"
print(f"✅ Without trends test passed - {len(output)} chars")
async def test_workflow_with_website_analysis():
"""Test research_workflow with website analysis included."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "cloud",
"max_sources": 5,
"include_website_analysis": True,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
else:
assert len(output) > 300, f"Output: {len(output)} chars"
print(f"✅ With website analysis test passed - {len(output)} chars")
async def test_workflow_without_website_analysis():
"""Test research_workflow without website analysis."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "security",
"max_sources": 5,
"include_website_analysis": False,
"research_depth": "basic",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
else:
assert len(output) > 300, f"Output: {len(output)} chars"
print(f"✅ Without website analysis test passed - {len(output)} chars")
async def test_workflow_comprehensive_all_features():
"""Test research_workflow with all features enabled."""
async with create_client() as client:
result = await client.call_tool(
"research_agent",
{
"topic": "AI trends",
"max_sources": 5,
"include_trends": True,
"include_website_analysis": True,
"research_depth": "comprehensive",
"enable_ai_insights": True,
},
)
output = result.content[0].text
# Handle rate limits
if "rate limit" in output.lower():
print("⚠️ OpenRouter rate limited")
assert len(output) > 50, "Error handling too minimal"
else:
assert len(output) > 300, f"Comprehensive workflow: {len(output)} chars"
# Should have research structure
has_structure = (
"#" in output or "research" in output.lower() or "topic" in output.lower()
)
assert has_structure, "No research structure found"
print(f"✅ Comprehensive all features test passed - {len(output)} chars")
if __name__ == "__main__":
asyncio.run(test_workflow_default_params())
asyncio.run(test_workflow_basic_depth())
asyncio.run(test_workflow_comprehensive_depth())
asyncio.run(test_workflow_expert_depth())
asyncio.run(test_workflow_max_sources())
asyncio.run(test_workflow_with_trends())
asyncio.run(test_workflow_without_trends())
asyncio.run(test_workflow_with_website_analysis())
asyncio.run(test_workflow_without_website_analysis())
asyncio.run(test_workflow_comprehensive_all_features())
print("\n✅ All research_agent tests passed!")
