test_app_summary_shortcircuit.py
4.27 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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
#!/usr/bin/env python3
"""Dev Agent RED/GREEN 测试：POST /api/articles/summarize 摘要状态机 (app.py)。
运行: python3 test_app_summary_shortcircuit.py
实现前应先失败 = RED；实现后 ALL PASS = GREEN。
用 Flask test_client（REMOTE_ADDR=127.0.0.1 ⇒ /api/ 免登录）+ monkeypatch _spawn_hermes_job
避免真的派发 hermes 任务。config.LOCAL_DB 指向临时库。
"""
import os
import sys
import sqlite3
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import config
TMP = os.path.join(tempfile.mkdtemp(prefix="appsc_"), "test.db")
config.LOCAL_DB = TMP
import local_db # noqa: E402
local_db.init_db()
FAILS = []
def check(name, cond, detail=""):
print(f"[{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
if not cond:
FAILS.append(name)
def seed_article(url, title, status, attempts):
conn = sqlite3.connect(TMP)
conn.execute(
"INSERT INTO articles (url,title,dedup_key,summary_status,summary_attempts) VALUES (?,?,?,?,?)",
(url, title, url, status, attempts))
conn.commit()
conn.close()
def job_count():
conn = sqlite3.connect(TMP)
n = conn.execute("SELECT COUNT(*) FROM hermes_jobs").fetchone()[0]
conn.close()
return n
def attempts_of(url):
conn = sqlite3.connect(TMP)
row = conn.execute("SELECT summary_attempts, summary_status FROM articles WHERE dedup_key=?",
(url,)).fetchone()
conn.close()
return row
# ── import app AFTER patching config ──
import app as appmod # noqa: E402
spawn_calls = []
appmod._spawn_hermes_job = lambda *a, **k: spawn_calls.append((a, k)) # monkeypatch
client = appmod.app.test_client()
# TC1: failed 文章 → 短路，不派发，status='failed'，attempts 不增
U1 = "https://example.com/ac1-failed-unique"
seed_article(U1, "AC1失败样本", "failed", 3)
jb = job_count()
r = client.post("/api/articles/summarize", json={"url": U1, "title": "AC1失败样本"})
j = r.get_json()
check("TC1a HTTP 200 + ok", r.status_code == 200 and bool(j.get("ok")), f"{r.status_code} {j}")
check("TC1b data.status == failed", (j.get("data") or {}).get("status") == "failed", str(j))
check("TC1c no spawn", job_count() == jb and len(spawn_calls) == 0, f"jobs {jb}->{job_count()} spawn={len(spawn_calls)}")
check("TC1d attempts unchanged ==3", attempts_of(U1)[0] == 3, str(attempts_of(U1)))
# TC2: attempts>=3 但 status 仍是 pending（防御）→ 同样短路
U2 = "https://example.com/ac2-attempts3"
seed_article(U2, "AC2已达上限", "pending", 3)
spawn_calls.clear()
jb = job_count()
r = client.post("/api/articles/summarize", json={"url": U2, "title": "AC2已达上限"})
j = r.get_json()
check("TC2a attempts>=3 shortcircuit failed", (j.get("data") or {}).get("status") == "failed", str(j))
check("TC2b no spawn", len(spawn_calls) == 0 and job_count() == jb)
# TC3: pending 文章 → 派发前 set_dispatching（status=summarizing, attempts+1）
U3 = "https://example.com/ac3-pending"
seed_article(U3, "AC3正常", "pending", 0)
spawn_calls.clear()
r = client.post("/api/articles/summarize", json={"url": U3, "title": "AC3正常"})
j = r.get_json()
check("TC3a started + spawn called", (j.get("data") or {}).get("status") == "started" and len(spawn_calls) == 1,
f"{j} spawn={len(spawn_calls)}")
st = attempts_of(U3)
check("TC3b dispatching -> summarizing attempts=1", st[1] == "summarizing" and st[0] == 1, str(st))
# TC4: cache 命中 → 先返回 'cached'，不计 attempts、不派发
U4 = "https://example.com/ac4-cached"
seed_article(U4, "AC4已缓存", "pending", 0)
conn = sqlite3.connect(TMP)
conn.execute("INSERT INTO daily_reports (chatroom_id,chatroom_name,report_date,content,dedup_key,article_url)"
" VALUES (?,?,?,?,?,?)",
("g", "G", "summary", "y" * 80, U4, U4))
conn.commit()
conn.close()
spawn_calls.clear()
r = client.post("/api/articles/summarize", json={"url": U4, "title": "AC4已缓存"})
j = r.get_json()
check("TC4a cache hit", (j.get("data") or {}).get("status") == "cached", str(j))
check("TC4b cache hit no attempts/no spawn",
attempts_of(U4)[0] == 0 and len(spawn_calls) == 0, f"{attempts_of(U4)} spawn={len(spawn_calls)}")
print()
if FAILS:
print(f"RESULT: {len(FAILS)} FAILED -> {FAILS}")
sys.exit(1)
print("RESULT: ALL PASS")