64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
告警同步功能测试脚本
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import time
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from utils.logger import get_logger
|
||
|
||
def test_alert_sync():
|
||
"""测试告警同步功能"""
|
||
|
||
print("开始测试告警同步功能...")
|
||
|
||
# 创建不同模块的logger
|
||
task_logger = get_logger("services.task_service")
|
||
system_logger = get_logger("system.database")
|
||
robot_logger = get_logger("services.robot_scheduler")
|
||
|
||
print("\n1. 测试任务告警...")
|
||
task_logger.warning("任务执行超时警告")
|
||
task_logger.error("任务执行失败", exc_info=False)
|
||
task_logger.critical("任务调度器严重错误", exc_info=False)
|
||
|
||
print("\n2. 测试系统告警...")
|
||
system_logger.warning("数据库连接不稳定")
|
||
system_logger.error("系统配置错误", exc_info=False)
|
||
|
||
print("\n3. 测试机器人告警...")
|
||
robot_logger.warning("机器人电池电量低")
|
||
robot_logger.error("机器人失去控制权", exc_info=False)
|
||
|
||
print("\n4. 测试异常告警...")
|
||
try:
|
||
1 / 0
|
||
except Exception as e:
|
||
task_logger.exception("发生了一个除零异常")
|
||
|
||
print("\n5. 测试INFO级别（不应该同步）...")
|
||
task_logger.info("这是一条普通信息，不应该同步到主系统")
|
||
|
||
print("\n测试完成，请检查主系统是否收到告警信息")
|
||
print("等待5秒让异步处理完成...")
|
||
time.sleep(5)
|
||
|
||
# 获取告警同步服务状态
|
||
from utils.alert_sync import get_alert_sync_service
|
||
service = get_alert_sync_service()
|
||
|
||
print(f"\n告警同步服务状态:")
|
||
print(f"- 启用状态: {service.enabled}")
|
||
print(f"- 同步URL: {service.sync_url}")
|
||
print(f"- 队列大小: {service.alert_queue.qsize()}")
|
||
print(f"- 最小同步级别: {service.min_level}")
|
||
|
||
if __name__ == "__main__":
|
||
test_alert_sync() |