shujuyanshi/tets/test_auto_exit.py

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