test_step6_approvals.py
3.8 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
# Step 6 验证：approvals 新增端点（dry-run / batch-send / batch-cancel）
# 安全策略：不真正外发微信。batch-send 只测「非 pending 拒绝」路径。
import sys, json
sys.path.insert(0, "/data/data/com.termux/files/home/wechat-viewer")
import local_db
from app import app
# ── 准备：插入一条测试 pending 回复（用假 talker，绝不被真实发送）
rid = local_db.reply_add(
msg_id=0, talker="step6_test_wxid", sender="step6_test_wxid",
sender_name="Step6测试", kind="private", incoming="测试消息",
draft="Step6 dry-run 测试草稿", action="reply", reason="step6-test",
status="pending",
)
print(f"[setup] inserted test reply id={rid}")
client = app.test_client()
# ── T1: dry-run 模拟执行（不真正外发）
r = client.post(f"/api/reply/{rid}/dry-run", json={"text": "模拟内容abc"})
j = r.get_json()
assert r.status_code == 200 and j.get("ok"), f"T1 FAIL: {r.status_code} {j}"
s = j["simulation"]
assert s["to_talker"] == "step6_test_wxid" and s["text"] == "模拟内容abc"
assert s["channel"] == "wechat_hook_send" and "未发送" in s["note"]
print(f"[T1 PASS] dry-run 模拟: to={s['to_talker']} text={s['text']!r}")
# ── T2: dry-run 空内容拒绝
r = client.post(f"/api/reply/{rid}/dry-run", json={"text": " "})
j = r.get_json()
assert not j.get("ok"), f"T2 FAIL: {j}"
print(f"[T2 PASS] dry-run 空内容拒绝: {j['error']}")
# ── T3: batch-cancel 对称操作（不外发）
# 2026-10-12 语义修正（PAP Step 6 审计）：原断言 cancelled==2 把「不存在的 id」也算成取消成功
# ——那是假成功读数（silent no-op reported as success）。现在逐条返回 gate，cancelled 只计真实取消数。
r = client.post("/api/replies/batch-cancel", json={"ids": [rid, 999999]})
j = r.get_json()
assert j.get("ok") and j.get("cancelled") == 1 and j.get("requested") == 2, f"T3 FAIL: {j}"
assert [x["gate"] for x in j["results"]] == ["CANCELLED", "REPLY_NOT_FOUND"], f"T3 FAIL gates: {j}"
print(f"[T3 PASS] batch-cancel 对称: cancelled={j['cancelled']}/{j['requested']} "
f"gates={[x['gate'] for x in j['results']]}")
# ── T4: batch-send 对非 pending 条目拒绝（不真发）
r = client.post("/api/replies/batch-send", json={"ids": [rid]})
j = r.get_json()
assert j.get("ok") is True and j["results"][0]["ok"] is False, f"T4 FAIL: {j}"
assert "非待审批" in j["results"][0]["error"], f"T4 FAIL err: {j}"
print(f"[T4 PASS] batch-send 拒绝非 pending: {j['results'][0]['error']}")
# ── T5: batch-send 空 ids 拒绝
r = client.post("/api/replies/batch-send", json={"ids": []})
j = r.get_json()
assert not j.get("ok"), f"T5 FAIL: {j}"
print(f"[T5 PASS] batch-send 空 ids 拒绝: {j['error']}")
# ── T6: 真正发送路径（pending 状态）→ 只验证会走到 _send_wechat_text 之前的状态检查
rid2 = local_db.reply_add(
msg_id=0, talker="step6_test_wxid2", sender="step6_test_wxid2",
sender_name="Step6测试2", kind="private", incoming="测试消息2",
draft="step6 pending 测试", action="reply", reason="step6-test",
status="pending",
)
# 直接验证 pending 状态能被读取（batch-send 的守卫逻辑：status 检查通过后才会 send）
item = local_db.reply_get(rid2)
assert item["status"] == "pending", f"T6 FAIL: {item['status']}"
# 清理：把 rid2 标记 cancelled，不触发真实发送
local_db.reply_update(rid2, "cancelled")
print(f"[T6 PASS] pending 守卫路径确认: id={rid2} status={item['status']} (未触发真实发送)")
# ── 清理测试数据
try:
conn = local_db.get_conn()
conn.execute("DELETE FROM reply_queue WHERE msg_id=0 AND sender_name LIKE 'Step6测试%'")
conn.commit()
conn.close()
print("[cleanup] 测试数据已清理")
except Exception as e:
print(f"[cleanup] 清理失败(不阻塞): {e}")
print("\nALL_STEP6_TESTS_PASS")