diff --git a/tets/test_auto_exit.py b/tets/test_auto_exit.py new file mode 100644 index 0000000..1222432 --- /dev/null +++ b/tets/test_auto_exit.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +""" +测试智能退出功能的简单脚本 +""" + +import requests +import time +import json + +def test_single_stream_auto_exit(): + """测试单流智能退出功能""" + print("=" * 50) + print("测试单流智能退出功能") + print("=" * 50) + + try: + # 1. 测试健康检查 + print("1. 测试健康检查...") + response = requests.get("http://localhost:5009/api/health", timeout=5) + if response.status_code == 200: + print("✓ 服务器运行正常") + print(f" 响应: {response.json()}") + else: + print("✗ 服务器未运行，请先启动单流服务器") + return False + + # 2. 测试获取智能退出配置 + print("\n2. 测试获取智能退出配置...") + try: + response = requests.get("http://localhost:5009/api/config/auto_exit", timeout=5) + if response.status_code == 200: + config = response.json() + print("✓ 获取智能退出配置成功") + print(f" 当前配置: {config}") + else: + print("✗ 获取智能退出配置失败") + except Exception as e: + print(f"✗ 请求失败: {e}") + + # 3. 测试配置智能退出参数 + print("\n3. 测试配置智能退出参数...") + config_data = { + "timeout": 30, # 30秒无图片自动退出 + "check_interval": 3 # 每3秒检查一次 + } + + try: + response = requests.post( + "http://localhost:5009/api/config/auto_exit", + json=config_data, + timeout=5 + ) + if response.status_code == 200: + result = response.json() + print("✓ 配置智能退出参数成功") + print(f" 配置结果: {result}") + else: + print("✗ 配置智能退出参数失败") + print(f" 错误响应: {response.text}") + except Exception as e: + print(f"✗ 请求失败: {e}") + + # 4. 测试获取统计信息（包含智能退出状态） + print("\n4. 测试获取统计信息...") + try: + response = requests.get("http://localhost:5009/api/stats", timeout=5) + if response.status_code == 200: + stats = response.json() + print("✓ 获取统计信息成功") + print(f" 智能退出相关字段:") + print(f" - 无帧超时: {stats.get('no_frame_timeout', 'N/A')}秒") + print(f" - 检查间隔: {stats.get('frame_check_interval', 'N/A')}秒") + print(f" - 无帧持续时间: {stats.get('no_frame_duration', 'N/A')}秒") + print(f" - 剩余退出时间: {stats.get('auto_exit_remaining', 'N/A')}秒") + else: + print("✗ 获取统计信息失败") + except Exception as e: + print(f"✗ 请求失败: {e}") + + print("\n✓ 单流智能退出功能测试完成") + return True + + except requests.exceptions.ConnectionError: + print("✗ 无法连接到服务器，请确保单流服务器正在运行在 http://localhost:5009") + return False + except Exception as e: + print(f"✗ 测试过程中出现错误: {e}") + return False + +def test_multi_stream_auto_exit(): + """测试多流智能退出功能""" + print("\n" + "=" * 50) + print("测试多流智能退出功能") + print("=" * 50) + + try: + # 1. 测试健康检查 + print("1. 测试健康检查...") + response = requests.get("http://localhost:5010/api/health", timeout=5) + if response.status_code == 200: + print("✓ 多流服务器运行正常") + print(f" 响应: {response.json()}") + else: + print("✗ 多流服务器未运行，请先启动多流服务器") + return False + + # 2. 测试获取智能退出配置 + print("\n2. 测试获取智能退出配置...") + try: + response = requests.get("http://localhost:5010/api/config/auto_exit", timeout=5) + if response.status_code == 200: + config = response.json() + print("✓ 获取多流智能退出配置成功") + print(f" 当前配置: {config}") + else: + print("✗ 获取多流智能退出配置失败") + except Exception as e: + print(f"✗ 请求失败: {e}") + + # 3. 测试配置智能退出参数 + print("\n3. 测试配置智能退出参数...") + config_data = { + "enabled": True, + "timeout": 45, # 45秒无图片自动退出 + "check_interval": 5 # 每5秒检查一次 + } + + try: + response = requests.post( + "http://localhost:5010/api/config/auto_exit", + json=config_data, + timeout=5 + ) + if response.status_code == 200: + result = response.json() + print("✓ 配置多流智能退出参数成功") + print(f" 配置结果: {result}") + else: + print("✗ 配置多流智能退出参数失败") + print(f" 错误响应: {response.text}") + except Exception as e: + print(f"✗ 请求失败: {e}") + + # 4. 测试获取系统统计（包含智能退出状态） + print("\n4. 测试获取系统统计...") + try: + response = requests.get("http://localhost:5010/api/system/stats", timeout=5) + if response.status_code == 200: + stats = response.json() + print("✓ 获取系统统计成功") + print(f" 智能退出相关字段:") + print(f" - 智能退出启用: {stats.get('auto_exit_enabled', 'N/A')}") + print(f" - 无帧超时: {stats.get('auto_exit_timeout', 'N/A')}秒") + print(f" - 检查间隔: {stats.get('auto_exit_check_interval', 'N/A')}秒") + print(f" - 智能停止的流数量: {stats.get('auto_stopped_streams', 'N/A')}") + print(f" - 处于风险中的流数量: {stats.get('streams_at_risk', 'N/A')}") + else: + print("✗ 获取系统统计失败") + except Exception as e: + print(f"✗ 请求失败: {e}") + + # 5. 测试创建流（验证新流会自动配置智能退出） + print("\n5. 测试创建新流...") + stream_data = { + "name": "测试智能退出流", + "url": "rtmp://test.example.com/live/test" + } + + try: + response = requests.post( + "http://localhost:5010/api/streams", + json=stream_data, + timeout=5 + ) + if response.status_code == 201: + result = response.json() + print("✓ 创建新流成功") + print(f" 流信息: {result}") + + # 获取流的统计信息验证智能退出配置 + if 'stream_id' in result: + stream_id = result['stream_id'] + print(f" 验证流 {stream_id[:8]} 的智能退出配置...") + + stats_response = requests.get(f"http://localhost:5010/api/streams/{stream_id}/stats", timeout=5) + if stats_response.status_code == 200: + stream_stats = stats_response.json() + print(f" - 无帧超时: {stream_stats.get('no_frame_timeout', 'N/A')}秒") + print(f" - 检查间隔: {stream_stats.get('frame_check_interval', 'N/A')}秒") + + else: + print("✗ 创建新流失败") + print(f" 错误响应: {response.text}") + except Exception as e: + print(f"✗ 请求失败: {e}") + + print("\n✓ 多流智能退出功能测试完成") + return True + + except requests.exceptions.ConnectionError: + print("✗ 无法连接到多流服务器，请确保多流服务器正在运行在 http://localhost:5010") + return False + except Exception as e: + print(f"✗ 测试过程中出现错误: {e}") + return False + +if __name__ == "__main__": + print("开始测试后端智能退出功能") + print("请确保服务器正在运行:") + print(" - 单流服务器: http://localhost:5009") + print(" - 多流服务器: http://localhost:5010") + print() + + # 测试单流功能 + single_ok = test_single_stream_auto_exit() + + # 等待一下 + time.sleep(2) + + # 测试多流功能 + multi_ok = test_multi_stream_auto_exit() + + # 总结 + print("\n" + "=" * 50) + print("测试总结") + print("=" * 50) + print(f"单流智能退出功能: {'✓ 通过' if single_ok else '✗ 失败'}") + print(f"多流智能退出功能: {'✓ 通过' if multi_ok else '✗ 失败'}") + + if single_ok and multi_ok: + print("\n🎉 所有测试通过！后端智能退出功能已成功实现。") + else: + print("\n⚠️ 部分测试失败，请检查服务器状态。") \ No newline at end of file