test_learn.py
5.06 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
"""Learn Pipeline - Integration Tests (TDD)"""
import sys, os, json, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import requests
BASE = "http://127.0.0.1:5000"
FAILURES = []
def check(name, condition, detail=""):
if condition:
print(f" PASS {name}")
else:
print(f" FAIL {name}: {detail}")
FAILURES.append(f"{name}: {detail}")
# ============================================================
# Test 1: DB layer - learn_queue CRUD
# ============================================================
def test_learn_table_exists():
"""learn_queue table should exist after init_db"""
import local_db
conn = local_db.get_conn()
rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='learn_queue'").fetchall()
conn.close()
check("learn_queue table exists", len(rows) > 0, "Table not found")
def test_learn_add_and_list():
"""Add items and list them"""
import local_db
# Clean up any test data
local_db.learn_add(
article_url="https://mp.weixin.qq.com/s/test_learn_1",
title="Test Article 1: New AI Tool",
dedup_key="test|1|1|abc",
learn_type="skill",
learn_target="test-skill",
learn_content="# Test Skill\n\nThis is a test skill",
confidence="high"
)
local_db.learn_add(
article_url="https://mp.weixin.qq.com/s/test_learn_2",
title="Test Article 2: Market Trend",
dedup_key="test|2|2|def",
learn_type="memory",
learn_target="test_memory",
learn_content="Important market data",
confidence="medium"
)
items = local_db.learn_list("pending_review")
# Filter to our test items
test_items = [i for i in items if "test_learn" in i.get("article_url", "")]
check("learn items listed", len(test_items) >= 2, f"Got {len(test_items)}")
def test_learn_update_status():
"""Approve and reject update status"""
import local_db
items = local_db.learn_list("pending_review")
test_items = [i for i in items if "test_learn" in i.get("article_url", "")]
if len(test_items) >= 2:
id1 = test_items[0]["id"]
local_db.learn_update_status(id1, "approved")
item = local_db.learn_get(id1)
check("status updated to approved", item["status"] == "approved")
check("reviewed_at set", item["reviewed_at"] is not None)
id2 = test_items[1]["id"]
local_db.learn_update_status(id2, "rejected")
item = local_db.learn_get(id2)
check("status updated to rejected", item["status"] == "rejected")
# ============================================================
# Test 2: API layer
# ============================================================
def test_api_learn_queue():
"""GET /api/learn-queue returns pending items"""
r = requests.get(f"{BASE}/api/learn-queue")
check("HTTP 200", r.status_code == 200, f"Got {r.status_code}")
j = r.json()
check("ok is true", j.get("ok") == True)
check("data is list", isinstance(j.get("data"), list))
def test_api_learn_classify():
"""POST /api/learn/classify spawns classification for new summaries"""
r = requests.post(f"{BASE}/api/learn/classify")
check("HTTP 200", r.status_code == 200, f"Got {r.status_code}")
j = r.json()
check("ok is true", j.get("ok") == True, f"Error: {j.get('error','')}")
def test_api_learn_approve_reject():
"""Approve and reject endpoints work"""
# Create fresh test items via API
import local_db
local_db.learn_add("https://test/approve1", "Approve Test", "t|a|1|x", "memory", "test_mem", "Test content", "high")
local_db.learn_add("https://test/reject1", "Reject Test", "t|r|1|y", "todo", "test_todo", "Test content", "low")
items = requests.get(f"{BASE}/api/learn-queue").json()["data"]
test_items = [i for i in items if "test" in i.get("article_url", "")]
if len(test_items) < 2:
print(" SKIP (not enough test items)")
return
# Approve
item_id = test_items[0]["id"]
r = requests.post(f"{BASE}/api/learn/{item_id}/approve")
check("approve HTTP 200", r.status_code == 200)
# Reject
if len(items) > 1:
item_id2 = items[1]["id"]
r = requests.post(f"{BASE}/api/learn/{item_id2}/reject")
check("reject HTTP 200", r.status_code == 200)
# ============================================================
def test_learn_page():
"""The /learn page renders"""
r = requests.get(f"{BASE}/learn")
check("page HTTP 200", r.status_code == 200, f"Got {r.status_code}")
check("has 学习审核", "学习审核" in r.text)
# ============================================================
print("=" * 60)
print("Learn Pipeline — Integration Tests")
print("=" * 60)
test_learn_table_exists()
test_learn_add_and_list()
test_learn_update_status()
test_api_learn_queue()
test_api_learn_classify()
test_api_learn_approve_reject()
test_learn_page()
print("\n" + "=" * 60)
if FAILURES:
print(f"FAILED: {len(FAILURES)} test(s)")
for f in FAILURES:
print(f" ❌ {f}")
sys.exit(1)
else:
print("ALL TESTS PASSED ✅")