qyyfw
78ff261b5a
- 前端 React+TypeScript (源码+构建产物) - 后端 Python FastAPI - 数据库迁移 001-015 - Docker部署配置 - 微信ClawBot API频道绑定 - 项目文档 README.md
437 lines
17 KiB
Python
437 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
频道模块 - API数据完整性回归验证（改进规范版）
|
|
覆盖：字段类型验证、空数据处理、错误码验证
|
|
"""
|
|
import requests
|
|
import json
|
|
from datetime import datetime
|
|
|
|
BASE = "http://localhost:18080"
|
|
API = f"{BASE}/app/api/channels"
|
|
|
|
passed = 0
|
|
failed = 0
|
|
results_log = []
|
|
|
|
def login(username="admin", password="admin123"):
|
|
r = requests.post(f"{BASE}/app/api/auth/login", json={"username": username, "password": password})
|
|
if r.status_code == 200:
|
|
return r.json().get("data", {}).get("access_token", "")
|
|
return ""
|
|
|
|
def hdr(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
def check_field(obj, field, expect_type, path=""):
|
|
full = f"{path}.{field}" if path else field
|
|
if field not in obj:
|
|
return False, f"缺少字段 '{full}'"
|
|
val = obj[field]
|
|
if expect_type == "int" and not (isinstance(val, int) and not isinstance(val, bool)):
|
|
return False, f"'{full}' 期望int, 实际{type(val).__name__}={val}"
|
|
if expect_type == "float_or_int" and not (isinstance(val, (int, float)) and not isinstance(val, bool)):
|
|
return False, f"'{full}' 期望number, 实际{type(val).__name__}={val}"
|
|
if expect_type == "str_or_none" and not (val is None or isinstance(val, str)):
|
|
return False, f"'{full}' 期望str|None, 实际{type(val).__name__}={val}"
|
|
if expect_type == "bool" and not isinstance(val, bool):
|
|
return False, f"'{full}' 期望bool, 实际{type(val).__name__}={val}"
|
|
if expect_type == "dict_or_none" and not (val is None or isinstance(val, dict)):
|
|
return False, f"'{full}' 期望dict|None, 实际{type(val).__name__}={val}"
|
|
if expect_type == "list" and not isinstance(val, list):
|
|
return False, f"'{full}' 期望list, 实际{type(val).__name__}={val}"
|
|
if expect_type == "str" and not isinstance(val, str):
|
|
return False, f"'{full}' 期望str, 实际{type(val).__name__}={val}"
|
|
return True, ""
|
|
|
|
def record(name, method, path, ok, detail):
|
|
global passed, failed
|
|
s = "PASS" if ok else "FAIL"
|
|
print(f"[{s}] {name}")
|
|
print(f" {method} {path} | {detail}")
|
|
results_log.append({"name": name, "method": method, "path": path, "passed": ok, "detail": detail})
|
|
if ok: passed += 1
|
|
else: failed += 1
|
|
|
|
# ===== 各测试用例 =====
|
|
|
|
def t1_platform_configs():
|
|
"""平台凭证配置 - 字段完整性"""
|
|
tok = login()
|
|
r = requests.get(f"{API}/platforms/config", headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t1", "GET", "/platforms/config", False, f"HTTP {r.status_code}")
|
|
|
|
data = r.json()
|
|
if data.get("code") != 0:
|
|
return record("t1", "GET", "/platforms/config", False, f"code={data.get('code')}")
|
|
|
|
platforms = data.get("data", {})
|
|
expected = ["feishu", "wechat", "dingtalk", "whatsapp", "telegram", "slack"]
|
|
ok = True
|
|
details = []
|
|
for p in expected:
|
|
if p not in platforms:
|
|
ok = False; details.append(f"缺少平台 {p}")
|
|
continue
|
|
cfg = platforms[p]
|
|
for f in ["name", "description", "fields", "doc_url"]:
|
|
f_ok, f_d = check_field(cfg, f, "str" if f != "fields" else "list", p)
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
for i, field in enumerate(cfg.get("fields", [])):
|
|
for ff in ["key", "label", "type", "placeholder"]:
|
|
f_ok, f_d = check_field(field, ff, "str", f"{p}.fields[{i}]")
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
detail = "6个平台配置完整" if ok else "; ".join(details)
|
|
record("t1", "GET", "/platforms/config", ok, detail)
|
|
|
|
def t2_list_empty():
|
|
"""空频道列表场景"""
|
|
tok = login()
|
|
# 先清空所有频道
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
if r0.status_code == 200:
|
|
for ch in r0.json().get("data", []):
|
|
requests.delete(f"{API}/{ch['id']}", headers=hdr(tok))
|
|
|
|
r = requests.get(API + "/", headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t2", "GET", "/", False, f"HTTP {r.status_code}")
|
|
|
|
items = r.json().get("data", [])
|
|
if not isinstance(items, list):
|
|
record("t2", "GET", "/", False, f"data不是list: {type(items)}")
|
|
elif len(items) != 0:
|
|
record("t2", "GET", "/", False, f"清理后仍有{len(items)}条记录")
|
|
else:
|
|
record("t2", "GET", "/", True, "空列表返回 [] 正确")
|
|
|
|
def t3_create_all_platforms():
|
|
"""创建6个平台频道"""
|
|
tok = login()
|
|
platforms = [
|
|
("feishu", {"app_id": "t1", "app_secret": "s1"}),
|
|
("wechat", {"app_id": "w1", "app_secret": "s2", "token": "t1"}),
|
|
("dingtalk", {"app_key": "d1", "app_secret": "s3"}),
|
|
("whatsapp", {"phone_number_id": "123", "access_token": "a1"}),
|
|
("telegram", {"bot_token": "b1"}),
|
|
("slack", {"bot_token": "b2", "signing_secret": "s4"}),
|
|
]
|
|
created = []
|
|
ok = True
|
|
details = []
|
|
for p_name, cred in platforms:
|
|
r = requests.post(API + "/", json={
|
|
"name": f"测试-{p_name}",
|
|
"platform": p_name,
|
|
"description": f"{p_name}频道",
|
|
"credential": cred
|
|
}, headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
ok = False; details.append(f"{p_name}: HTTP {r.status_code}")
|
|
continue
|
|
data = r.json()
|
|
if data.get("code") != 0:
|
|
ok = False; details.append(f"{p_name}: code={data.get('code')}")
|
|
continue
|
|
ch = data.get("data", {})
|
|
# 字段检查
|
|
for f in ["id", "name", "platform", "status", "is_active", "credential_masked"]:
|
|
f_ok, f_d = check_field(ch, f, "str" if f != "is_active" else "bool", p_name)
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
if ch.get("platform") != p_name:
|
|
ok = False; details.append(f"{p_name}: platform不匹配={ch.get('platform')}")
|
|
created.append(ch["id"])
|
|
|
|
detail = f"{len(created)}/6创建成功" + ("; "+"; ".join(details[-3:]) if details else "")
|
|
record("t3", "POST", "/", ok, detail)
|
|
return created
|
|
|
|
def t4_channel_detail():
|
|
"""频道详情字段完整性"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
if not chs:
|
|
return record("t4", "GET", "/{id}", False, "无可用频道")
|
|
|
|
ch_id = chs[0]["id"]
|
|
r = requests.get(f"{API}/{ch_id}", headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t4", "GET", f"/{ch_id}", False, f"HTTP {r.status_code}")
|
|
|
|
ch = r.json().get("data", {})
|
|
ok = True; details = []
|
|
checks = [
|
|
("id","str"),("name","str"),("platform","str"),
|
|
("status","str"),("is_active","bool"),
|
|
("description","str_or_none"),("webhook_url","str_or_none"),
|
|
("role","str_or_none"),("member_count","int"),("message_count","int"),
|
|
("credential_masked","dict_or_none")
|
|
]
|
|
for f, t in checks:
|
|
f_ok, f_d = check_field(ch, f, t)
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
detail = "所有字段类型正确" if ok else "; ".join(details)
|
|
record("t4", "GET", f"/{ch_id}", ok, detail)
|
|
|
|
def t5_update_settings():
|
|
"""更新频道AI配置"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
if not chs:
|
|
return record("t5", "PUT", "/{id}", False, "无可用频道")
|
|
|
|
ch_id = chs[0]["id"]
|
|
r = requests.put(f"{API}/{ch_id}", json={
|
|
"name": "已更新名称",
|
|
"description": "已更新描述",
|
|
"settings": {
|
|
"agent_id": "agent-001",
|
|
"knowledge_base_id": "kb-001",
|
|
"skill_ids": ["skill-1"],
|
|
"welcome_message": "欢迎"
|
|
}
|
|
}, headers=hdr(tok))
|
|
|
|
if r.status_code != 200:
|
|
return record("t5", "PUT", f"/{ch_id}", False, f"HTTP {r.status_code}")
|
|
data = r.json()
|
|
ch = data.get("data", {})
|
|
ok = True; details = []
|
|
if ch.get("name") != "已更新名称":
|
|
ok = False; details.append(f"name未更新={ch.get('name')}")
|
|
if ch.get("default_agent_id") != "agent-001":
|
|
details.append(f"agent_id同步={ch.get('default_agent_id')}(期望agent-001)")
|
|
if ch.get("knowledge_base_id") != "kb-001":
|
|
details.append(f"kb_id同步={ch.get('knowledge_base_id')}(期望kb-001)")
|
|
detail = "更新成功" if ok else "; ".join(details)
|
|
record("t5", "PUT", f"/{ch_id}", ok, detail)
|
|
|
|
def t6_connect_disconnect():
|
|
"""连接/断开频道"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
if not chs:
|
|
return record("t6", "POST", "/{id}/connect", False, "无可用频道")
|
|
|
|
ch_id = chs[0]["id"]
|
|
# 连接
|
|
r = requests.post(f"{API}/{ch_id}/connect", headers=hdr(tok))
|
|
if r.status_code not in (200, 400):
|
|
return record("t6", "POST", f"/{ch_id}/connect", False, f"HTTP {r.status_code}")
|
|
|
|
data = r.json()
|
|
ok = True; details = []
|
|
if r.json().get("code") == 0:
|
|
conn = data.get("data", {})
|
|
for f in ["connected", "platform", "status", "message"]:
|
|
if f not in conn:
|
|
ok = False; details.append(f"缺少字段{f}")
|
|
details.append("连接成功")
|
|
|
|
# 断开
|
|
r2 = requests.post(f"{API}/{ch_id}/disconnect", headers=hdr(tok))
|
|
if r2.status_code == 200 and r2.json().get("code") == 0:
|
|
details.append("断开成功")
|
|
else:
|
|
ok = False; details.append(f"断开失败: HTTP {r2.status_code}")
|
|
|
|
# 重新连接
|
|
r3 = requests.post(f"{API}/{ch_id}/connect", headers=hdr(tok))
|
|
if r3.status_code == 200 and r3.json().get("code") == 0:
|
|
details.append("重新连接成功")
|
|
else:
|
|
details.append(f"重新连接: HTTP {r3.status_code}")
|
|
else:
|
|
details.append(f"连接失败: code={data.get('code')}, msg={data.get('message')}")
|
|
|
|
detail = "; ".join(details)
|
|
record("t6", "POST", f"/{ch_id}/connect|disconnect", ok, detail)
|
|
|
|
def t7_webhook_message():
|
|
"""Webhook消息收发"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
connected = [c for c in chs if c.get("status") == "connected"]
|
|
if not connected:
|
|
return record("t7", "POST", "/{id}/webhook", False, "无已连接频道")
|
|
|
|
ch_id = connected[0]["id"]
|
|
r = requests.post(f"{API}/{ch_id}/webhook", json={
|
|
"from_user": "wx_user_001",
|
|
"content": "你好测试消息",
|
|
"message_type": "text"
|
|
}, headers=hdr(tok))
|
|
|
|
if r.status_code not in (200, 400):
|
|
return record("t7", "POST", f"/{ch_id}/webhook", False, f"HTTP {r.status_code}")
|
|
|
|
data = r.json()
|
|
if data.get("code") != 0:
|
|
return record("t7", "POST", f"/{ch_id}/webhook", False, f"code={data.get('code')}, msg={data.get('message')}")
|
|
|
|
wd = data.get("data", {})
|
|
ok = True; details = []
|
|
# received检查
|
|
if "received" in wd:
|
|
rc = wd["received"]
|
|
for f in ["id", "from_user", "content", "message_type", "created_at"]:
|
|
if f not in rc: ok = False; details.append(f"received缺{f}")
|
|
else:
|
|
ok = False; details.append("缺received")
|
|
# reply检查
|
|
if "reply" in wd:
|
|
rp = wd["reply"]
|
|
for f in ["id", "content", "role", "created_at"]:
|
|
if f not in rp: ok = False; details.append(f"reply缺{f}")
|
|
if rp.get("role") != "assistant":
|
|
ok = False; details.append(f"role={rp.get('role')}")
|
|
else:
|
|
ok = False; details.append("缺reply")
|
|
|
|
detail = "received+reply字段完整" if ok else "; ".join(details)
|
|
record("t7", "POST", f"/{ch_id}/webhook", ok, detail)
|
|
|
|
def t8_messages_list():
|
|
"""消息列表字段完整性"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
connected = [c for c in chs if c.get("status") == "connected"]
|
|
if not connected:
|
|
return record("t8", "GET", "/{id}/messages", False, "无已连接频道")
|
|
|
|
ch_id = connected[0]["id"]
|
|
r = requests.get(f"{API}/{ch_id}/messages", params={"limit": 10}, headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t8", "GET", f"/{ch_id}/messages", False, f"HTTP {r.status_code}")
|
|
|
|
msgs = r.json().get("data", [])
|
|
if not isinstance(msgs, list):
|
|
return record("t8", "GET", f"/{ch_id}/messages", False, f"data不是list: {type(msgs)}")
|
|
|
|
ok = True; details = [f"共{len(msgs)}条"]
|
|
for i, m in enumerate(msgs):
|
|
for f in ["id", "channel_id", "user_id", "content", "role"]:
|
|
f_ok, f_d = check_field(m, f, "str", f"msgs[{i}]")
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
detail = "; ".join(details)
|
|
record("t8", "GET", f"/{ch_id}/messages", ok, detail)
|
|
|
|
def t9_send_message():
|
|
"""发送消息字段完整性"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
connected = [c for c in chs if c.get("status") == "connected"]
|
|
if not connected:
|
|
return record("t9", "POST", "/{id}/messages", False, "无已连接频道")
|
|
|
|
ch_id = connected[0]["id"]
|
|
r = requests.post(f"{API}/{ch_id}/messages", json={"content": "测试消息"}, headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t9", "POST", f"/{ch_id}/messages", False, f"HTTP {r.status_code}, body={r.text[:100]}")
|
|
|
|
data = r.json()
|
|
if data.get("code") != 0:
|
|
return record("t9", "POST", f"/{ch_id}/messages", False, f"code={data.get('code')}")
|
|
|
|
wd = data.get("data", {})
|
|
ok = True; details = []
|
|
for section in ["message", "reply", "webhook_response"]:
|
|
if section not in wd:
|
|
ok = False; details.append(f"缺{section}")
|
|
elif section == "message":
|
|
for f in ["id", "channel_id", "user_id", "content"]:
|
|
if f not in wd[section]: ok = False; details.append(f"message缺{f}")
|
|
elif section == "reply":
|
|
for f in ["id", "content", "role"]:
|
|
if f not in wd[section]: ok = False; details.append(f"reply缺{f}")
|
|
elif section == "webhook_response":
|
|
for f in ["status", "platform", "message"]:
|
|
if f not in wd[section]: ok = False; details.append(f"webhook_response缺{f}")
|
|
detail = "message/reply/webhook_response完整" if ok else "; ".join(details)
|
|
record("t9", "POST", f"/{ch_id}/messages", ok, detail)
|
|
|
|
def t10_members():
|
|
"""频道成员管理"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
if not chs:
|
|
return record("t10", "GET", "/{id}/members", False, "无可用频道")
|
|
|
|
ch_id = chs[0]["id"]
|
|
r = requests.get(f"{API}/{ch_id}/members", headers=hdr(tok))
|
|
if r.status_code != 200:
|
|
return record("t10", "GET", f"/{ch_id}/members", False, f"HTTP {r.status_code}")
|
|
|
|
members = r.json().get("data", [])
|
|
ok = True; details = [f"共{len(members)}名成员"]
|
|
for i, m in enumerate(members):
|
|
for f in ["id", "user_id", "username", "role"]:
|
|
f_ok, f_d = check_field(m, f, "str", f"members[{i}]")
|
|
if not f_ok: ok = False; details.append(f_d)
|
|
detail = "; ".join(details)
|
|
record("t10", "GET", f"/{ch_id}/members", ok, detail)
|
|
|
|
def t11_delete_and_cleanup():
|
|
"""删除频道"""
|
|
tok = login()
|
|
r0 = requests.get(API + "/", headers=hdr(tok))
|
|
chs = r0.json().get("data", [])
|
|
if not chs:
|
|
return record("t11", "DELETE", "/{id}", True, "无频道需删除，跳过")
|
|
|
|
ch_id = chs[0]["id"]
|
|
r = requests.delete(f"{API}/{ch_id}", headers=hdr(tok))
|
|
ok = r.status_code == 200 and r.json().get("code") == 0
|
|
detail = "删除成功" if ok else f"HTTP {r.status_code}"
|
|
record("t11", "DELETE", f"/{ch_id}", ok, detail)
|
|
|
|
# ===== 主函数 =====
|
|
|
|
def main():
|
|
global passed, failed
|
|
print("=" * 70)
|
|
print(f"频道模块 API数据完整性回归 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("=" * 70)
|
|
|
|
t1_platform_configs()
|
|
t2_list_empty()
|
|
created = t3_create_all_platforms()
|
|
t4_channel_detail()
|
|
t5_update_settings()
|
|
t6_connect_disconnect()
|
|
t7_webhook_message()
|
|
t8_messages_list()
|
|
t9_send_message()
|
|
t10_members()
|
|
t11_delete_and_cleanup()
|
|
|
|
print()
|
|
print("=" * 70)
|
|
total = passed + failed
|
|
rate = passed / total * 100 if total > 0 else 0
|
|
print(f"汇总: {passed}/{total} PASS, {failed} FAIL, {rate:.1f}%")
|
|
print("=" * 70)
|
|
|
|
# JSON report
|
|
report = {
|
|
"test_time": datetime.now().isoformat(),
|
|
"module": "频道模块-API数据完整性",
|
|
"total": total, "passed": passed, "failed": failed,
|
|
"pass_rate": f"{rate:.1f}%",
|
|
"results": results_log
|
|
}
|
|
print("\n=== JSON_REPORT ===")
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
print("=== END_JSON_REPORT ===")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|