test_summary_status.py
4.02 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
#!/usr/bin/env python3
"""Dev Agent RED/GREEN 测试：文章摘要状态机 (local_db)。
运行: python3 test_summary_status.py
实现前应先失败（AttributeError / 断言失败）= RED；实现后 ALL PASS = GREEN。
不污染生产 DB：改写 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="summary_status_"), "test.db")
config.LOCAL_DB = TMP
import local_db # noqa: E402
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 state_of(url):
return local_db.article_summary_get_state(None, url)
# TC1: init_db 幂等 + 新增 3 列存在
local_db.init_db()
local_db.init_db()
cols = [r[1] for r in sqlite3.connect(TMP).execute("PRAGMA table_info(articles)")]
check("TC1a summary_status column", "summary_status" in cols, str(cols))
check("TC1b summary_attempts column", "summary_attempts" in cols)
check("TC1c summary_updated_at column", "summary_updated_at" in cols)
# TC2: 回填——已有摘要的文章标 ok；未摘要的保持 pending
conn = sqlite3.connect(TMP)
conn.execute("INSERT INTO articles (url,title,dedup_key) VALUES (?,?,?)",
("https://mp.weixin.qq.com/s?__biz=A&mid=1&idx=1&sn=abc", "已摘要", "A|1|1|abc"))
NOURL = "https://example.com/nosum"
conn.execute("INSERT INTO articles (url,title,dedup_key) VALUES (?,?,?)",
(NOURL, "未摘要", NOURL))
conn.execute(
"INSERT INTO daily_reports (chatroom_id,chatroom_name,report_date,content,dedup_key,article_url)"
" VALUES (?,?,?,?,?,?)",
("g", "G", "summary", "x" * 60, "A|1|1|abc", "https://mp.weixin.qq.com/s?__biz=A&mid=1&idx=1&sn=abc"))
conn.commit()
conn.close()
local_db.init_db() # 触发回填
st = local_db.article_summary_get_state("A|1|1|abc")
check("TC2a backfill summarized -> ok", bool(st) and st["summary_status"] == "ok", str(st))
check("TC2b unsummarized stays pending", state_of(NOURL)["summary_status"] == "pending")
# TC3: 状态转移 pending -> summarizing -> (3 次失败) -> failed
local_db.article_summary_set_dispatching(None, NOURL)
s = state_of(NOURL)
check("TC3a dispatching -> summarizing attempts=1",
s["summary_status"] == "summarizing" and s["summary_attempts"] == 1, str(s))
local_db.article_summary_mark_failed(None, NOURL)
check("TC3b fail #1 -> pending", state_of(NOURL)["summary_status"] == "pending")
local_db.article_summary_set_dispatching(None, NOURL)
local_db.article_summary_mark_failed(None, NOURL)
local_db.article_summary_set_dispatching(None, NOURL)
local_db.article_summary_mark_failed(None, NOURL)
s = state_of(NOURL)
check("TC3c fail #3 -> failed attempts=3",
s["summary_status"] == "failed" and s["summary_attempts"] == 3, str(s))
# TC3d: failed 后 set_dispatching 不再自增（短路保护）
before = state_of(NOURL)["summary_attempts"]
local_db.article_summary_set_dispatching(None, NOURL)
check("TC3d failed not re-dispatched", state_of(NOURL)["summary_attempts"] == before)
# TC4: mark_ok
local_db.article_summary_mark_ok("A|1|1|abc")
check("TC4 mark_ok", local_db.article_summary_get_state("A|1|1|abc")["summary_status"] == "ok")
# TC5: state_counts
c = local_db.article_summary_state_counts()
check("TC5 state_counts ok/failed/total",
c.get("ok") == 1 and c.get("failed") == 1 and c.get("total") == 2, str(c))
# TC6: failed_titles
t = local_db.article_summary_failed_titles()
check("TC6 failed_titles contains 未摘要", "未摘要" in t, str(t))
# TC7: 返回语义不变——article_get_unsummarized_new 仍可用（不因新列改变行为）
try:
rows = local_db.article_get_unsummarized_new(limit=5)
check("TC7 unsummarized_new callable", isinstance(rows, list), f"{len(rows)} rows")
except Exception as e:
check("TC7 unsummarized_new callable", False, repr(e))
print()
if FAILS:
print(f"RESULT: {len(FAILS)} FAILED -> {FAILS}")
sys.exit(1)
print("RESULT: ALL PASS")