deepagents----/tests/test_coordinator.py

"""
ResearchCoordinator测试

测试主Agent的完整执行流程
"""

import sys
import os

# 添加src目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from src.agents.coordinator import create_research_coordinator, run_research
from src.config import Config


def test_coordinator_creation():
    """测试ResearchCoordinator创建"""
    print("=" * 60)
    print("测试1: ResearchCoordinator创建")
    print("=" * 60)

    try:
        # 测试默认参数
        agent = create_research_coordinator(
            question="什么是Python asyncio?",
            depth="quick"
        )

        print("✓ ResearchCoordinator创建成功")
        print(f"  Agent类型: {type(agent)}")
        return True

    except Exception as e:
        print(f"✗ ResearchCoordinator创建失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_config_validation():
    """测试配置验证"""
    print("\n" + "=" * 60)
    print("测试2: 配置验证")
    print("=" * 60)

    # 测试无效的深度模式
    try:
        agent = create_research_coordinator(
            question="测试问题",
            depth="invalid_depth"
        )
        print("✗ 应该抛出ValueError但没有")
        return False
    except ValueError as e:
        print(f"✓ 正确捕获无效深度模式: {e}")

    # 测试无效的min_tier
    try:
        agent = create_research_coordinator(
            question="测试问题",
            min_tier=5
        )
        print("✗ 应该抛出ValueError但没有")
        return False
    except ValueError as e:
        print(f"✓ 正确捕获无效min_tier: {e}")

    # 测试无效的格式
    try:
        agent = create_research_coordinator(
            question="测试问题",
            format="invalid_format"
        )
        print("✗ 应该抛出ValueError但没有")
        return False
    except ValueError as e:
        print(f"✓ 正确捕获无效格式: {e}")

    return True


def test_simple_research_dry_run():
    """测试简单研究流程（dry run，不执行真实搜索）"""
    print("\n" + "=" * 60)
    print("测试3: 简单研究流程（模拟）")
    print("=" * 60)

    print("\n注意: 这个测试需要API密钥才能执行真实的Agent调用")
    print("如果API密钥未配置，将跳过此测试\n")

    # 检查API密钥
    try:
        Config.validate()
    except ValueError as e:
        print(f"⚠️  跳过测试：{e}")
        return True  # 不算失败

    try:
        # 创建Agent但不执行
        agent = create_research_coordinator(
            question="Python装饰器的作用",
            depth="quick",
            format="technical"
        )

        print("✓ Agent创建成功，准备就绪")
        print("  如需运行完整测试，请确保API密钥已配置")
        print("  然后运行：python -m tests.test_integration")

        return True

    except Exception as e:
        print(f"✗ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_depth_configs():
    """测试三种深度模式的配置"""
    print("\n" + "=" * 60)
    print("测试4: 深度模式配置")
    print("=" * 60)

    depth_modes = ["quick", "standard", "deep"]

    for depth in depth_modes:
        try:
            agent = create_research_coordinator(
                question="测试问题",
                depth=depth
            )

            depth_config = Config.get_depth_config(depth)

            print(f"\n✓ {depth}模式配置正确:")
            print(f"  - 最大迭代: {depth_config['max_iterations']}")
            print(f"  - 置信度阈值: {depth_config['confidence_threshold']}")
            print(f"  - 目标来源数: {depth_config['target_sources']}")
            print(f"  - 并行搜索数: {depth_config['parallel_searches']}")

        except Exception as e:
            print(f"✗ {depth}模式配置失败: {e}")
            return False

    return True


def main():
    """运行所有测试"""
    print("\n")
    print("=" * 60)
    print("ResearchCoordinator测试套件")
    print("=" * 60)
    print("\n")

    results = []

    # 测试1: 创建
    results.append(("创建测试", test_coordinator_creation()))

    # 测试2: 配置验证
    results.append(("配置验证", test_config_validation()))

    # 测试3: 简单研究流程
    results.append(("简单研究流程", test_simple_research_dry_run()))

    # 测试4: 深度模式配置
    results.append(("深度模式配置", test_depth_configs()))

    # 总结
    print("\n" + "=" * 60)
    print("测试总结")
    print("=" * 60)

    for test_name, passed in results:
        status = "✓ 通过" if passed else "✗ 失败"
        print(f"{test_name}: {status}")

    all_passed = all(result[1] for result in results)

    print("\n" + "=" * 60)
    if all_passed:
        print("✓ 所有测试通过！ResearchCoordinator实现正确。")
    else:
        print("✗ 部分测试失败，请检查实现。")
    print("=" * 60 + "\n")

    return all_passed


if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)