test_light_browser.py

#!/usr/bin/env python3
"""
Light Browser 功能测试脚本
测试 CDP 接口和 noVNC 服务是否正常工作
"""

import json
import time
import requests
import websocket
from loguru import logger

# 配置
CDP_PORT = 9222
NOVNC_PORT = 6080
TEST_URL = "https://httpbin.org/get"


def test_cdp_connection():
    """测试 CDP 连接"""
    try:
        logger.info("测试 CDP 连接...")
        
        # 获取版本信息
        response = requests.get(f"http://localhost:{CDP_PORT}/json/version", timeout=10)
        if response.status_code == 200:
            version_info = response.json()
            logger.info(f"CDP 版本信息: {version_info.get('Browser', 'Unknown')}")
            return True
        else:
            logger.error(f"CDP 版本接口返回错误: {response.status_code}")
            return False
            
    except Exception as e:
        logger.error(f"CDP 连接测试失败: {e}")
        return False


def test_cdp_websocket():
    """测试 CDP WebSocket 功能"""
    try:
        logger.info("测试 CDP WebSocket...")
        
        # 获取可用的标签页
        response = requests.get(f"http://localhost:{CDP_PORT}/json", timeout=10)
        if response.status_code != 200:
            logger.error("无法获取标签页列表")
            return False
        
        tabs = response.json()
        if not tabs:
            logger.error("没有可用的标签页")
            return False
        
        # 连接到第一个标签页的 WebSocket
        ws_url = tabs[0]['webSocketDebuggerUrl']
        logger.info(f"连接到 WebSocket: {ws_url}")
        
        ws = websocket.create_connection(ws_url, timeout=10)
        
        # 启用页面域
        ws.send(json.dumps({
            "id": 1,
            "method": "Page.enable"
        }))
        
        result = json.loads(ws.recv())
        if result.get("id") == 1:
            logger.info("Page.enable 成功")
        
        # 导航到测试页面
        ws.send(json.dumps({
            "id": 2,
            "method": "Page.navigate",
            "params": {"url": TEST_URL}
        }))
        
        result = json.loads(ws.recv())
        if result.get("id") == 2:
            logger.info(f"页面导航成功: {TEST_URL}")
        
        # 等待页面加载事件
        timeout = time.time() + 10
        while time.time() < timeout:
            try:
                message = json.loads(ws.recv())
                if message.get("method") == "Page.loadEventFired":
                    logger.info("页面加载完成")
                    break
            except:
                continue
        
        # 获取页面标题
        ws.send(json.dumps({
            "id": 3,
            "method": "Runtime.evaluate",
            "params": {"expression": "document.title"}
        }))
        
        result = json.loads(ws.recv())
        if result.get("id") == 3 and "result" in result:
            title = result["result"]["result"]["value"]
            logger.info(f"页面标题: {title}")
        
        ws.close()
        logger.info("CDP WebSocket 测试成功")
        return True
        
    except Exception as e:
        logger.error(f"CDP WebSocket 测试失败: {e}")
        return False


def test_novnc_service():
    """测试 noVNC 服务"""
    try:
        logger.info("测试 noVNC 服务...")
        
        # 检查 noVNC 主页
        response = requests.get(f"http://localhost:{NOVNC_PORT}", timeout=10)
        if response.status_code == 200:
            logger.info("noVNC 主页访问正常")
        else:
            logger.warning(f"noVNC 主页返回状态码: {response.status_code}")
        
        # 检查 vnc.html 页面
        response = requests.get(f"http://localhost:{NOVNC_PORT}/vnc.html", timeout=10)
        if response.status_code == 200:
            logger.info("noVNC 界面页面访问正常")
            return True
        else:
            logger.error(f"noVNC 界面页面返回错误: {response.status_code}")
            return False
            
    except Exception as e:
        logger.error(f"noVNC 服务测试失败: {e}")
        return False


def test_service_health():
    """测试服务健康状态"""
    try:
        logger.info("测试服务健康状态...")
        
        # 检查端口监听
        import socket
        
        # 检查 CDP 端口
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        result = sock.connect_ex(('localhost', CDP_PORT))
        sock.close()
        
        if result == 0:
            logger.info(f"CDP 端口 {CDP_PORT} 监听正常")
        else:
            logger.error(f"CDP 端口 {CDP_PORT} 未监听")
            return False
        
        # 检查 noVNC 端口
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        result = sock.connect_ex(('localhost', NOVNC_PORT))
        sock.close()
        
        if result == 0:
            logger.info(f"noVNC 端口 {NOVNC_PORT} 监听正常")
        else:
            logger.error(f"noVNC 端口 {NOVNC_PORT} 未监听")
            return False
        
        return True
        
    except Exception as e:
        logger.error(f"服务健康状态测试失败: {e}")
        return False


def main():
    """主测试函数"""
    logger.info("=" * 50)
    logger.info("开始 Light Browser 功能测试")
    logger.info("=" * 50)
    
    test_results = []
    
    # 测试服务健康状态
    test_results.append(("服务健康状态", test_service_health()))
    
    # 测试 CDP 连接
    test_results.append(("CDP 连接", test_cdp_connection()))
    
    # 测试 CDP WebSocket
    test_results.append(("CDP WebSocket", test_cdp_websocket()))
    
    # 测试 noVNC 服务
    test_results.append(("noVNC 服务", test_novnc_service()))
    
    # 输出测试结果
    logger.info("=" * 50)
    logger.info("测试结果汇总")
    logger.info("=" * 50)
    
    passed = 0
    total = len(test_results)
    
    for test_name, result in test_results:
        status = "✅ 通过" if result else "❌ 失败"
        logger.info(f"{test_name}: {status}")
        if result:
            passed += 1
    
    logger.info("=" * 50)
    logger.info(f"测试完成: {passed}/{total} 项通过")
    
    if passed == total:
        logger.info("🎉 所有测试通过！Light Browser 服务运行正常")
        return True
    else:
        logger.error("❌ 部分测试失败，请检查服务状态")
        return False


if __name__ == "__main__":
    # 配置日志
    logger.add(
        "test_light_browser_{time:YYYY-MM-DD}.log",
        rotation="1 day",
        retention="3 days",
        level="INFO",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}"
    )
    
    success = main()
    exit(0 if success else 1)