test_smoke.py
3.37 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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
"""回归测试 — 每次修改 app.py 后运行: python3 test_smoke.py"""
import urllib.request, json, sys
BASE = "http://127.0.0.1:5001"
passed = 0
failed = 0
def test(label, method, path, data=None, check=None):
global passed, failed
try:
url = f"{BASE}{path}"
if data:
req = urllib.request.Request(url, data=json.dumps(data).encode(),
headers={"Content-Type": "application/json"}, method=method)
else:
req = urllib.request.Request(url, method=method)
with urllib.request.urlopen(req, timeout=10) as r:
body = r.read()
if check:
ok, msg = check(r.status, body)
else:
ok, msg = r.status == 200, f"HTTP {r.status}"
if ok:
passed += 1
print(f" PASS {label}: {msg}")
else:
failed += 1
print(f" FAIL {label}: {msg}")
except Exception as e:
failed += 1
print(f" FAIL {label}: {e}")
# T1
test("首页", "GET", "/")
# T2
def check_sessions(status, body):
d = json.loads(body)
previews = sum(1 for s in d["data"] if s.get("preview"))
return d["ok"] and previews > 0, f"total={len(d['data'])} previews={previews}"
test("会话列表", "GET", "/api/sessions", check=check_sessions)
# T3
def check_messages(status, body):
d = json.loads(body)
return d["ok"] and len(d["data"]) > 0, f"msgs={len(d['data'])}"
test("消息列表", "GET", "/api/messages?talker=filehelper&limit=3", check=check_messages)
# T4 — 分页格式
def check_articles(status, body):
d = json.loads(body)
return (d["ok"] and d.get("total", 0) > 0
and d.get("page") == 1
and d.get("per_page", 0) > 0
and len(d["data"]) <= d["per_page"]
and isinstance(d.get("data"), list)), f"total={d.get('total')} page={d.get('page')} per_page={d.get('per_page')} len={len(d.get('data',[]))}"
test("文章列表(分页)", "GET", "/api/articles?page=1&per_page=5", check=check_articles)
# T5
def check_dedup(status, body):
d = json.loads(body)
return d["ok"], f"results={len(d['data']['results'])} missing={len(d['data']['missing'])}"
test("去重查询", "POST", "/api/articles/check-summaries",
data={"urls": [{"url": "https://mp.weixin.qq.com/s?__biz=test&mid=1&idx=1&sn=abc", "title": "t"}], "type": "summary"},
check=check_dedup)
# T6
def check_reports(status, body):
d = json.loads(body)
return d["ok"], f"reports={len(d['data'])}"
test("日报列表", "GET", "/api/reports", check=check_reports)
# T7
test("图片代理", "GET", "/media/image?path=THUMBNAIL_DIRPATH://th_c669882238b8abc023f04c2e3e30ba57")
# T8
def check_chat(status, body):
html = body.decode()
bubbles = html.count("msg-bubble")
return bubbles > 0, f"msg-bubbles={bubbles}"
test("聊天页", "GET", "/chat/filehelper", check=check_chat)
# T9 — Agents page
def check_agents(status, body):
html = body.decode()
return status == 200 and "Agents" in html, f"HTTP {status}"
test("Agents页", "GET", "/agents", check=check_agents)
# T10 — Agent detail page
def check_agent_detail(status, body):
html = body.decode()
return status == 200 and "SOUL" in html, f"HTTP {status}"
test("Agent详情", "GET", "/agent/default", check=check_agent_detail)
print(f"\n{passed}/{passed+failed} 通过")
sys.exit(0 if failed == 0 else 1)