stock-ai-agent/scripts/test_error_notification.py

#!/usr/bin/env python3
"""
测试系统异常通知功能

测试场景：
1. 测试普通异常通知
2. 测试带堆栈信息的异常通知
3. 测试冷却时间（避免重复通知）
"""
import sys
from pathlib import Path

# 添加项目路径
backend_dir = Path(__file__).parent.parent / "backend"
sys.path.insert(0, str(backend_dir))


def test_normal_exception():
    """测试普通异常通知"""
    print("\n" + "=" * 60)
    print("🧪 测试1: 普通异常通知")
    print("=" * 60)

    from app.utils.error_handler import get_exception_handler

    handler = get_exception_handler()

    # 模拟一个异常
    try:
        result = 1 / 0  # ZeroDivisionError
    except Exception as e:
        import traceback
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 已触发异常处理（请检查飞书是否收到通知）")


def test_custom_exception():
    """测试自定义异常通知"""
    print("\n" + "=" * 60)
    print("🧪 测试2: 自定义异常通知")
    print("=" * 60)

    from app.utils.error_handler import get_exception_handler

    handler = get_exception_handler()

    # 模拟一个自定义异常
    try:
        raise ValueError("这是一个测试异常，用于验证飞书通知功能")
    except Exception as e:
        import traceback
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 已触发自定义异常处理（请检查飞书是否收到通知）")


def test_nested_exception():
    """测试嵌套调用异常"""
    print("\n" + "=" * 60)
    print("🧪 测试3: 嵌套调用异常（带完整堆栈）")
    print("=" * 60)

    from app.utils.error_handler import get_exception_handler

    handler = get_exception_handler()

    def level_3():
        """第三层调用"""
        raise RuntimeError("深层错误：数据库连接失败")

    def level_2():
        """第二层调用"""
        level_3()

    def level_1():
        """第一层调用"""
        level_2()

    # 模拟嵌套调用异常
    try:
        level_1()
    except Exception as e:
        import traceback
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 已触发嵌套异常处理（请检查飞书是否收到完整堆栈信息）")


def test_cooldown():
    """测试冷却时间"""
    print("\n" + "=" * 60)
    print("🧪 测试4: 冷却时间（连续触发异常）")
    print("=" * 60)

    from app.utils.error_handler import get_exception_handler
    import time

    handler = get_exception_handler()
    handler.set_cooldown(10)  # 设置10秒冷却时间

    print(f"⏰ 冷却时间设置为 10 秒")

    # 第一次触发
    try:
        raise RuntimeError("第一次异常")
    except Exception:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 第一次异常已处理")

    # 立即第二次触发（应该在冷却期内）
    print("\n⏳ 立即触发第二次异常（应该在冷却期内）...")
    try:
        raise RuntimeError("第二次异常")
    except Exception:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 第二次异常已处理（应该被冷却，不发送飞书通知）")

    print("\n⏳ 等待 11 秒后再次触发...")
    time.sleep(11)

    # 第三次触发（冷却期已过）
    try:
        raise RuntimeError("第三次异常")
    except Exception:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 第三次异常已处理（冷却期已过，应该发送飞书通知）")


def test_with_feishu_integration():
    """测试与飞书集成的完整流程"""
    print("\n" + "=" * 60)
    print("🧪 测试5: 完整集成测试（带飞书服务）")
    print("=" * 60)

    from app.utils.error_handler import setup_global_exception_handler, get_exception_handler
    from app.services.feishu_service import get_feishu_service

    # 设置全局异常处理器
    setup_global_exception_handler()
    print("✅ 全局异常处理器已安装")

    # 初始化飞书通知
    feishu_service = get_feishu_service()
    handler = get_exception_handler()
    handler.set_feishu_service(feishu_service)
    handler.set_enabled(True)
    print("✅ 飞书通知已启用")

    # 触发一个测试异常
    print("\n🔔 触发测试异常...")
    try:
        raise Exception("【测试】这是一个集成测试异常，用于验证飞书通知功能是否正常工作")
    except Exception:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        handler.handle_exception(exc_type, exc_value, exc_traceback)
        print("✅ 测试异常已处理，请检查飞书是否收到通知")


def main():
    """主测试函数"""
    print("\n" + "🚀" * 30)
    print("系统异常通知功能测试")
    print("🚀" * 30)

    print("\n⚠️  注意：以下测试会触发真实的异常，并尝试发送飞书通知")
    print("⚠️  请确保飞书 webhook 已正确配置\n")

    try:
        # 测试1: 普通异常
        test_normal_exception()

        # 等待用户确认
        input("\n按 Enter 键继续下一个测试...")

        # 测试2: 自定义异常
        test_custom_exception()

        # 等待用户确认
        input("\n按 Enter 键继续下一个测试...")

        # 测试3: 嵌套异常
        test_nested_exception()

        # 等待用户确认
        input("\n按 Enter 键继续下一个测试...")

        # 测试4: 冷却时间
        test_cooldown()

        # 等待用户确认
        input("\n按 Enter 键继续最后一个测试...")

        # 测试5: 完整集成
        test_with_feishu_integration()

        print("\n" + "=" * 60)
        print("✅ 所有测试完成！")
        print("=" * 60)
        print("\n📋 请检查飞书聊天窗口，确认是否收到异常通知")
        print("💡 提示：如果未收到通知，请检查：")
        print("   1. 飞书 webhook URL 是否正确")
        print("   2. 网络连接是否正常")
        print("   3. .env 文件中的 FEISHU_ENABLED 是否为 true\n")

    except KeyboardInterrupt:
        print("\n\n⚠️  测试被用户中断")
    except Exception as e:
        print(f"\n\n❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()