oficcejo
277 lines
8.9 KiB
Python
277 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
实时监测定时调度器测试脚本
|
|
用于验证调度器功能是否正常工作
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from monitor_service import monitor_service
|
|
from monitor_scheduler import get_scheduler
|
|
|
|
def print_separator():
|
|
"""打印分隔线"""
|
|
print("\n" + "="*60 + "\n")
|
|
|
|
def test_scheduler_initialization():
|
|
"""测试调度器初始化"""
|
|
print("📋 测试1: 调度器初始化")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
scheduler = get_scheduler(monitor_service)
|
|
if scheduler:
|
|
print("✅ 调度器初始化成功")
|
|
print(f" 配置文件: monitor_schedule_config.json")
|
|
return scheduler
|
|
else:
|
|
print("❌ 调度器初始化失败")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ 调度器初始化异常: {e}")
|
|
return None
|
|
|
|
def test_config_loading(scheduler):
|
|
"""测试配置加载"""
|
|
print("\n📋 测试2: 配置加载")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
config = scheduler.config
|
|
print("✅ 配置加载成功:")
|
|
print(f" 启用状态: {config.get('enabled', False)}")
|
|
print(f" 市场: {config.get('market', 'CN')}")
|
|
print(f" 交易日: {config.get('trading_days', [])}")
|
|
print(f" 自动停止: {config.get('auto_stop', True)}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 配置加载失败: {e}")
|
|
return False
|
|
|
|
def test_trading_day_detection(scheduler):
|
|
"""测试交易日检测"""
|
|
print("\n📋 测试3: 交易日检测")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
now = datetime.now()
|
|
is_trading_day = scheduler.is_trading_day()
|
|
weekday = now.weekday() + 1
|
|
weekday_names = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
|
|
|
print(f" 当前日期: {now.strftime('%Y-%m-%d')} ({weekday_names[weekday-1]})")
|
|
print(f" 是否交易日: {'✅ 是' if is_trading_day else '❌ 否'}")
|
|
print(f" 配置的交易日: {scheduler.config.get('trading_days', [])}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 交易日检测失败: {e}")
|
|
return False
|
|
|
|
def test_trading_time_detection(scheduler):
|
|
"""测试交易时间检测"""
|
|
print("\n📋 测试4: 交易时间检测")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
now = datetime.now()
|
|
is_trading_time = scheduler.is_trading_time()
|
|
next_time = scheduler.get_next_trading_time()
|
|
|
|
print(f" 当前时间: {now.strftime('%H:%M:%S')}")
|
|
print(f" 是否交易时间: {'✅ 是' if is_trading_time else '❌ 否'}")
|
|
print(f" 下次交易时间: {next_time}")
|
|
|
|
# 显示配置的交易时间
|
|
market = scheduler.config.get('market', 'CN')
|
|
trading_hours = scheduler.config['trading_hours'].get(market, [])
|
|
print(f" 市场: {market}")
|
|
print(f" 配置的交易时间:")
|
|
for i, period in enumerate(trading_hours, 1):
|
|
print(f" 时段{i}: {period['start']} - {period['end']}")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 交易时间检测失败: {e}")
|
|
return False
|
|
|
|
def test_scheduler_status(scheduler):
|
|
"""测试调度器状态"""
|
|
print("\n📋 测试5: 调度器状态")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
status = scheduler.get_status()
|
|
print("✅ 状态获取成功:")
|
|
print(f" 调度器运行中: {status['scheduler_running']}")
|
|
print(f" 定时已启用: {status['scheduler_enabled']}")
|
|
print(f" 是否交易日: {status['is_trading_day']}")
|
|
print(f" 是否交易时间: {status['is_trading_time']}")
|
|
print(f" 市场: {status['market']}")
|
|
print(f" 下次交易时间: {status['next_trading_time']}")
|
|
print(f" 监测服务运行: {status['monitor_service_running']}")
|
|
print(f" 自动停止: {status['auto_stop']}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 状态获取失败: {e}")
|
|
return False
|
|
|
|
def test_config_update(scheduler):
|
|
"""测试配置更新"""
|
|
print("\n📋 测试6: 配置更新")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
# 保存原始配置
|
|
original_enabled = scheduler.config.get('enabled', False)
|
|
|
|
print(" 测试配置更新...")
|
|
# 不真正改变配置，只测试API
|
|
print(f" 当前启用状态: {original_enabled}")
|
|
print(" ✅ 配置更新API可用")
|
|
print(" 注意: 实际配置未修改")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 配置更新失败: {e}")
|
|
return False
|
|
|
|
def test_monitor_service_integration(scheduler):
|
|
"""测试与监测服务的集成"""
|
|
print("\n📋 测试7: 监测服务集成")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
print(f" 监测服务状态: {'运行中' if monitor_service.running else '已停止'}")
|
|
print(f" 调度器实例: {scheduler is not None}")
|
|
print(" ✅ 集成测试通过")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 集成测试失败: {e}")
|
|
return False
|
|
|
|
def test_scheduler_start_stop(scheduler):
|
|
"""测试调度器启动停止"""
|
|
print("\n📋 测试8: 调度器启动/停止")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
# 检查当前状态
|
|
was_running = scheduler.running
|
|
print(f" 初始状态: {'运行中' if was_running else '已停止'}")
|
|
|
|
if not scheduler.config.get('enabled', False):
|
|
print(" ⚠️ 定时调度未启用，跳过启动测试")
|
|
print(" 提示: 在UI中启用定时调度后再测试")
|
|
else:
|
|
if not was_running:
|
|
print(" 尝试启动调度器...")
|
|
scheduler.start_scheduler()
|
|
time.sleep(2)
|
|
if scheduler.running:
|
|
print(" ✅ 调度器启动成功")
|
|
|
|
# 立即停止以恢复原状
|
|
print(" 尝试停止调度器...")
|
|
scheduler.stop_scheduler()
|
|
time.sleep(1)
|
|
if not scheduler.running:
|
|
print(" ✅ 调度器停止成功")
|
|
else:
|
|
print(" ⚠️ 调度器可能未完全停止")
|
|
else:
|
|
print(" ❌ 调度器启动失败")
|
|
else:
|
|
print(" ⚠️ 调度器已在运行，跳过启动测试")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 启动/停止测试失败: {e}")
|
|
return False
|
|
|
|
def run_all_tests():
|
|
"""运行所有测试"""
|
|
print("🚀 实时监测定时调度器测试")
|
|
print("="*60)
|
|
print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print_separator()
|
|
|
|
results = []
|
|
|
|
# 测试1: 初始化
|
|
scheduler = test_scheduler_initialization()
|
|
if not scheduler:
|
|
print("\n❌ 调度器初始化失败，无法继续测试")
|
|
return
|
|
results.append(scheduler is not None)
|
|
|
|
print_separator()
|
|
|
|
# 测试2: 配置加载
|
|
results.append(test_config_loading(scheduler))
|
|
print_separator()
|
|
|
|
# 测试3: 交易日检测
|
|
results.append(test_trading_day_detection(scheduler))
|
|
print_separator()
|
|
|
|
# 测试4: 交易时间检测
|
|
results.append(test_trading_time_detection(scheduler))
|
|
print_separator()
|
|
|
|
# 测试5: 状态获取
|
|
results.append(test_scheduler_status(scheduler))
|
|
print_separator()
|
|
|
|
# 测试6: 配置更新
|
|
results.append(test_config_update(scheduler))
|
|
print_separator()
|
|
|
|
# 测试7: 监测服务集成
|
|
results.append(test_monitor_service_integration(scheduler))
|
|
print_separator()
|
|
|
|
# 测试8: 启动停止
|
|
results.append(test_scheduler_start_stop(scheduler))
|
|
print_separator()
|
|
|
|
# 汇总结果
|
|
print("📊 测试结果汇总")
|
|
print("="*60)
|
|
total_tests = len(results)
|
|
passed_tests = sum(results)
|
|
failed_tests = total_tests - passed_tests
|
|
|
|
print(f"总测试数: {total_tests}")
|
|
print(f"通过: ✅ {passed_tests}")
|
|
print(f"失败: ❌ {failed_tests}")
|
|
print(f"通过率: {(passed_tests/total_tests*100):.1f}%")
|
|
|
|
print_separator()
|
|
|
|
if failed_tests == 0:
|
|
print("🎉 所有测试通过！定时调度器功能正常")
|
|
else:
|
|
print(f"⚠️ 有 {failed_tests} 个测试失败，请检查错误信息")
|
|
|
|
print(f"\n结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("="*60)
|
|
|
|
def main():
|
|
"""主函数"""
|
|
try:
|
|
run_all_tests()
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ 测试被用户中断")
|
|
sys.exit(0)
|
|
except Exception as e:
|
|
print(f"\n\n❌ 测试过程中发生错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|