test_s4_stream.py
2.94 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
#!/usr/bin/env python3
"""RED test for S4: 流式输出 + 打字机效果
Tests:
TC1: SSE stream endpoint exists and works
TC2: EventSource pattern in aichat.html
TC3: Typing indicator exists
TC4: Stream produces incremental output
"""
import sys, os, json, subprocess, urllib.parse
BASE = "http://127.0.0.1:5001"
def curl_get(path):
r = subprocess.run(["curl", "-s", f"{BASE}{path}"], capture_output=True, text=True, timeout=15)
return json.loads(r.stdout) if r.stdout else {}
def curl_post(path, data):
body = json.dumps(data)
r = subprocess.run(["curl", "-s", "-X", "POST", f"{BASE}{path}",
"-H", "Content-Type: application/json", "-d", body], capture_output=True, text=True, timeout=15)
return json.loads(r.stdout) if r.stdout else {}
def main():
errors = []
# TC1: SSE stream endpoint exists
# Create a conversation and add a message, then check stream
conv = curl_post("/api/conversations", {"title": "test-s4"})
if not conv.get("ok"):
errors.append(f"TC1: POST /api/conversations failed")
for e in errors: print(f" ✗ {e}")
sys.exit(1)
conv_id = conv["data"]["id"]
# Post a message to trigger AI (this should use the streaming path internally)
msg = curl_post(f"/api/conversations/{conv_id}/messages", {"content": "hi"})
if msg.get("ok"):
print(f" TC1: POST message got reply (ok={msg.get('ok')})")
else:
# Stream might fail if hermes is not available, but endpoint should still exist
print(f" TC1: POST message returned: ok={msg.get('ok')}, error={msg.get('error','')}")
# TC2: Check aichat.html has EventSource
r = subprocess.run(["su", "-c", "cat /data/data/com.termux/files/home/wechat-viewer/templates/aichat.html"],
capture_output=True, text=True, timeout=10)
html = r.stdout
if 'EventSource' not in html:
errors.append("TC2: aichat.html missing EventSource (SSE)")
if '/stream' not in html:
errors.append("TC2: aichat.html missing /stream URL pattern")
# TC3: Typing indicator exists
if 'typing' not in html.lower():
errors.append("TC3: aichat.html missing typing indicator")
if '思考中' not in html:
errors.append("TC3: aichat.html missing '思考中' text")
# TC4: Stream endpoint returns SSE format
stream_url = f"/api/conversations/{conv_id}/stream?message=test"
r2 = subprocess.run(["curl", "-s", "-N", "--max-time", "3", f"{BASE}{stream_url}"],
capture_output=True, text=True, timeout=10)
# Even if hermes backend fails, the endpoint should not be 404
print(f" TC4: Stream response ({len(r2.stdout)} bytes): {r2.stdout[:200]}")
if errors:
print(f"\nRED: {len(errors)} failures")
for e in errors:
print(f" ✗ {e}")
sys.exit(1)
else:
print("\nGREEN: All checks passed")
sys.exit(0)
if __name__ == '__main__':
main()