#!/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()