evotraders/scripts/test_sandbox.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
沙盒执行器测试脚本

测试多模式技能沙盒执行器的基本功能。
默认使用 none 模式（无沙盒）。
"""

import os
import sys

# 确保后端目录在路径中
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))


def test_sandbox_initialization():
    """测试沙盒初始化"""
    print("=" * 60)
    print("测试 1: 沙盒初始化")
    print("=" * 60)

    from backend.tools.sandboxed_executor import get_sandbox, SkillSandbox

    # 重置单例（确保测试干净）
    SkillSandbox._instance = None

    # 默认应该使用 none 模式
    sandbox = get_sandbox()

    assert sandbox.current_mode == "none", f"期望模式 'none'，实际 '{sandbox.current_mode}'"
    print(f"✓ 沙盒模式: {sandbox.current_mode}")
    print(f"✓ 后端类型: {type(sandbox._backend).__name__}")

    return sandbox


def test_no_sandbox_warning():
    """测试无沙盒模式的安全警告"""
    print("\n" + "=" * 60)
    print("测试 2: 无沙盒模式安全警告")
    print("=" * 60)

    import warnings

    from backend.tools.sandboxed_executor import NoSandboxBackend

    backend = NoSandboxBackend()

    # 捕获警告
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")

        # 执行会触发警告
        try:
            backend.execute(
                skill_name="builtin/valuation_review",
                function_name="build_dcf_report",
                function_args={"rows": [], "current_date": "2024-01-01"},
            )
        except Exception:
            pass  # 我们不关心执行结果，只关心警告

        # 检查是否产生了警告
        runtime_warnings = [x for x in w if issubclass(x.category, RuntimeWarning)]
        if runtime_warnings:
            print("✓ 安全警告已触发")
            print(f"  警告内容: {runtime_warnings[0].message}")
        else:
            print("⚠ 未触发安全警告（可能已缓存）")


def test_docker_config():
    """测试 Docker 模式配置解析"""
    print("\n" + "=" * 60)
    print("测试 3: Docker 模式配置解析")
    print("=" * 60)

    # 设置环境变量
    os.environ["SKILL_SANDBOX_MODE"] = "docker"
    os.environ["SKILL_SANDBOX_MEMORY_LIMIT"] = "1g"
    os.environ["SKILL_SANDBOX_CPU_LIMIT"] = "2.0"

    from backend.tools.sandboxed_executor import SkillSandbox

    # 重置单例
    SkillSandbox._instance = None

    try:
        sandbox = SkillSandbox()
        print(f"✓ 沙盒模式: {sandbox.current_mode}")
        print(f"✓ 后端类型: {type(sandbox._backend).__name__}")

        # 检查配置
        backend = sandbox._backend
        assert backend.config["memory_limit"] == "1g"
        assert backend.config["cpu_limit"] == 2.0
        print(f"✓ 内存限制: {backend.config['memory_limit']}")
        print(f"✓ CPU 限制: {backend.config['cpu_limit']}")

    except Exception as e:
        print(f"⚠ Docker 后端创建失败（预期，可能未安装 agentscope-runtime）: {e}")

    # 恢复环境变量
    os.environ["SKILL_SANDBOX_MODE"] = "none"
    SkillSandbox._instance = None


def test_analysis_tools_import():
    """测试分析工具导入"""
    print("\n" + "=" * 60)
    print("测试 4: 分析工具导入")
    print("=" * 60)

    try:
        from backend.tools.analysis_tools import (
            TOOL_REGISTRY,
            _sandbox,
            dcf_valuation_analysis,
        )

        print(f"✓ TOOL_REGISTRY 包含 {len(TOOL_REGISTRY)} 个工具")
        print(f"✓ _sandbox 实例模式: {_sandbox.current_mode}")
        print(f"✓ dcf_valuation_analysis 函数可用")

        # 检查估值分析工具是否都在
        valuation_tools = [
            "dcf_valuation_analysis",
            "owner_earnings_valuation_analysis",
            "ev_ebitda_valuation_analysis",
            "residual_income_valuation_analysis",
        ]

        for tool in valuation_tools:
            if tool in TOOL_REGISTRY:
                print(f"  ✓ {tool}")
            else:
                print(f"  ✗ {tool} 缺失")

    except Exception as e:
        print(f"✗ 导入失败: {e}")
        import traceback
        traceback.print_exc()


def test_skill_execution_mock():
    """测试技能执行（模拟）"""
    print("\n" + "=" * 60)
    print("测试 5: 技能执行（无沙盒模式）")
    print("=" * 60)

    from backend.tools.sandboxed_executor import get_sandbox

    sandbox = get_sandbox()

    # 使用模拟参数调用
    try:
        # 注意：这需要实际的技能模块存在
        result = sandbox.execute_skill(
            skill_name="builtin/valuation_review",
            function_name="build_dcf_report",
            function_args={
                "rows": [{"ticker": "AAPL", "current_fcf": 100000000}],
                "current_date": "2024-01-01",
            },
        )
        print(f"✓ 技能执行成功")
        print(f"  结果类型: {type(result)}")
        print(f"  结果预览: {str(result)[:100]}...")
    except Exception as e:
        print(f"⚠ 技能执行失败（可能缺少依赖或数据）: {e}")


def main():
    """主测试函数"""
    print("\n" + "=" * 60)
    print("技能沙盒执行器测试")
    print("=" * 60)
    print(f"当前 SKILL_SANDBOX_MODE: {os.getenv('SKILL_SANDBOX_MODE', '未设置（默认 none）')}")

    # 确保使用默认模式测试
    os.environ["SKILL_SANDBOX_MODE"] = "none"

    try:
        test_sandbox_initialization()
        test_no_sandbox_warning()
        test_docker_config()
        test_analysis_tools_import()
        test_skill_execution_mock()

        print("\n" + "=" * 60)
        print("测试完成")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()