105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试登录验证功能
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import requests
|
|
import websockets
|
|
from datetime import datetime
|
|
|
|
|
|
class LoginTester:
|
|
def __init__(self, base_url="http://localhost:18085"):
|
|
self.base_url = base_url
|
|
|
|
async def test_browser_status(self):
|
|
"""测试浏览器状态"""
|
|
print("\n=== 测试浏览器状态 ===")
|
|
try:
|
|
response = requests.get(f"{self.base_url}/api/debug/browser")
|
|
data = response.json()
|
|
print(f"浏览器实例存在: {data['browser_exists']}")
|
|
print(f"Context数量: {data['contexts_count']}")
|
|
print(f"数据目录: {data['data_dir']}")
|
|
print(f"Auth文件存在: {data['auth_file_exists']}")
|
|
print(f"Storage文件存在: {data['storage_file_exists']}")
|
|
return data
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
return None
|
|
|
|
async def test_login_status(self):
|
|
"""测试登录状态检查"""
|
|
print("\n=== 测试登录状态检查 ===")
|
|
try:
|
|
response = requests.get(f"{self.base_url}/api/check-login-status")
|
|
data = response.json()
|
|
print(f"响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
|
|
return data
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
return None
|
|
|
|
async def test_login_verify(self):
|
|
"""测试登录验证"""
|
|
print("\n=== 测试登录验证 ===")
|
|
try:
|
|
response = requests.post(f"{self.base_url}/api/login/verify")
|
|
data = response.json()
|
|
print(f"响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
|
|
return data
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
return None
|
|
|
|
async def test_sse_connection(self):
|
|
"""测试SSE连接"""
|
|
print("\n=== 测试SSE连接 ===")
|
|
try:
|
|
ws_url = f"{self.base_url.replace('http', 'ws')}/api/events"
|
|
async with websockets.connect(ws_url) as websocket:
|
|
# 等待几条消息
|
|
for i in range(3):
|
|
message = await asyncio.wait_for(websocket.recv(), timeout=5.0)
|
|
print(f"收到消息 {i+1}: {message}")
|
|
if i >= 2: # 只接收3条消息就关闭
|
|
break
|
|
except Exception as e:
|
|
print(f"✗ 测试失败: {e}")
|
|
|
|
async def run_all_tests(self):
|
|
"""运行所有测试"""
|
|
print(f"🚀 开始测试 - {datetime.now()}")
|
|
print("=" * 50)
|
|
|
|
# 等待服务启动
|
|
await asyncio.sleep(2)
|
|
|
|
# 测试1: 浏览器状态
|
|
browser_status = await self.test_browser_status()
|
|
|
|
# 测试2: 登录状态检查
|
|
login_status = await self.test_login_status()
|
|
|
|
# 测试3: 登录验证
|
|
login_verify = await self.test_login_verify()
|
|
|
|
# 测试4: SSE连接（如果服务正在运行）
|
|
if browser_status and browser_status.get('browser_exists'):
|
|
await self.test_sse_connection()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("🎯 测试完成")
|
|
|
|
# 输出总结
|
|
print("\n📊 测试总结:")
|
|
print(f"浏览器状态: {'✅ 正常' if browser_status and browser_status.get('browser_exists') else '❌ 异常'}")
|
|
print(f"登录状态检查: {'✅ 成功' if login_status and login_status.get('success') else '❌ 失败'}")
|
|
print(f"登录验证: {'✅ 成功' if login_verify and login_verify.get('success') else '❌ 失败'}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tester = LoginTester()
|
|
asyncio.run(tester.run_all_tests()) |