123

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⚠️  部分测试失败，请检查服务器状态。")