155 lines
4.8 KiB
Python
155 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AI Brain E2E 端到端验证脚本
|
||
|
||
验证 AI Brain 完整链路：
|
||
API → SDK → WebSocket → Agent AI Brain → 执行 → 结果回传
|
||
|
||
前置条件：
|
||
1. SDK 运行：localhost:8899
|
||
2. Agent v3.1 连接且 AI Brain 已启用
|
||
3. 卡若AI API 可达（localhost:3102）
|
||
|
||
环境变量：
|
||
- SDK_BASE_URL / SDK_DEVICE_ID
|
||
- AI_API_URL (默认 http://localhost:3102)
|
||
"""
|
||
|
||
import httpx
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import json
|
||
|
||
BASE_URL = os.environ.get("SDK_BASE_URL", "http://localhost:8899")
|
||
DEVICE_ID = os.environ.get("SDK_DEVICE_ID", "emulator-5554")
|
||
AI_API_URL = os.environ.get("AI_API_URL", "http://localhost:3102")
|
||
|
||
PASS = 0
|
||
FAIL = 0
|
||
|
||
|
||
def report(name: str, ok: bool, detail: str = ""):
|
||
global PASS, FAIL
|
||
if ok:
|
||
PASS += 1
|
||
print(f" ✅ {name}" + (f" — {detail}" if detail else ""))
|
||
else:
|
||
FAIL += 1
|
||
print(f" ❌ {name}" + (f" — {detail}" if detail else ""))
|
||
|
||
|
||
async def check_ai_api_reachable():
|
||
"""卡若AI API 连通性"""
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
try:
|
||
resp = await client.post(
|
||
f"{AI_API_URL}/api/gateway/chat",
|
||
headers={"Content-Type": "application/json"},
|
||
json={"messages": [{"role": "user", "content": "ping"}], "max_tokens": 10},
|
||
)
|
||
ok = resp.status_code == 200
|
||
report("卡若AI API 连通", ok, f"HTTP {resp.status_code}")
|
||
return ok
|
||
except Exception as e:
|
||
report("卡若AI API 连通", False, str(e))
|
||
return False
|
||
|
||
|
||
async def check_sdk_health():
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
resp = await client.get(f"{BASE_URL}/health")
|
||
h = resp.json()
|
||
online = h.get("devices_online", 0)
|
||
report("SDK 健康", resp.status_code == 200, f"devices_online={online}")
|
||
return online > 0
|
||
|
||
|
||
async def check_ai_brain_status():
|
||
"""查询设备 AI Brain 状态"""
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
resp = await client.get(f"{BASE_URL}/api/v3/devices/{DEVICE_ID}/ai/status")
|
||
if resp.status_code == 200:
|
||
data = resp.json().get("data", {})
|
||
report("AI Brain 状态查询", True, f"enabled={data.get('enabled')}, online={data.get('online')}")
|
||
return True
|
||
report("AI Brain 状态查询", False, f"HTTP {resp.status_code}")
|
||
return False
|
||
|
||
|
||
async def check_push_ai_task():
|
||
"""推送 AI 任务到设备"""
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
resp = await client.post(
|
||
f"{BASE_URL}/api/v3/devices/{DEVICE_ID}/ai/task",
|
||
json={"instruction": "[E2E测试] 检查微信是否在运行", "priority": 5},
|
||
)
|
||
r = resp.json()
|
||
ok = resp.status_code == 200
|
||
report("推送 AI 任务", ok, json.dumps(r, ensure_ascii=False)[:150])
|
||
return ok
|
||
|
||
|
||
async def check_push_standing_order():
|
||
"""推送常驻指令"""
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
resp = await client.post(
|
||
f"{BASE_URL}/api/v3/devices/{DEVICE_ID}/ai/standing-order",
|
||
json={"order": "[E2E测试] 每10分钟检查微信运行状态"},
|
||
)
|
||
r = resp.json()
|
||
ok = resp.status_code == 200
|
||
report("推送常驻指令", ok, json.dumps(r, ensure_ascii=False)[:150])
|
||
return ok
|
||
|
||
|
||
async def check_ai_execute():
|
||
"""AI Agent 同步执行自然语言任务"""
|
||
async with httpx.AsyncClient(timeout=90) as client:
|
||
resp = await client.post(
|
||
f"{BASE_URL}/api/v3/devices/{DEVICE_ID}/ai/execute",
|
||
json={"task": "检查当前手机状态", "timeout": 60},
|
||
)
|
||
r = resp.json()
|
||
ok = resp.status_code == 200
|
||
detail = r.get("data", {}).get("result", r.get("detail", ""))
|
||
if isinstance(detail, dict):
|
||
detail = json.dumps(detail, ensure_ascii=False)[:150]
|
||
report("AI 同步执行", ok, str(detail)[:150])
|
||
return ok
|
||
|
||
|
||
async def main():
|
||
print("=" * 55)
|
||
print(" AI Brain E2E 端到端验证")
|
||
print("=" * 55)
|
||
print(f" SDK: {BASE_URL}")
|
||
print(f" 设备: {DEVICE_ID}")
|
||
print(f" AI API: {AI_API_URL}")
|
||
print()
|
||
|
||
print("--- 基础连通 ---")
|
||
ai_ok = await check_ai_api_reachable()
|
||
has_device = await check_sdk_health()
|
||
|
||
if not has_device:
|
||
print("\n⚠️ 无设备在线，以下测试以 API 正确返回为准")
|
||
|
||
print("\n--- AI Brain API ---")
|
||
await check_ai_brain_status()
|
||
await check_push_ai_task()
|
||
await check_push_standing_order()
|
||
|
||
if has_device and ai_ok:
|
||
print("\n--- AI 同步执行 ---")
|
||
await check_ai_execute()
|
||
|
||
print(f"\n{'=' * 55}")
|
||
print(f" 结果: ✅ {PASS} 通过 ❌ {FAIL} 失败")
|
||
print(f"{'=' * 55}")
|
||
sys.exit(0 if FAIL == 0 else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|