test_jobs_integration.py
7.6 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
"""Hermes Jobs - Frontend Integration Test (TDD: RED → GREEN)
Simulates what a browser does: fetch HTML → extract JS → run logic → verify.
"""
import sys, os, json, re
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}")
# ============================================================
# RED Phase: These capture what the BROWSER actually does
# ============================================================
def test_api_jobs_returns_valid_json():
"""GET /api/jobs should return {ok: true, data: [...]}"""
r = requests.get(f"{BASE}/api/jobs")
check("HTTP 200", r.status_code == 200, f"got {r.status_code}")
j = r.json()
check("ok is true", j.get("ok") == True, f"got {j.get('ok')}")
check("data is list", isinstance(j.get("data"), list), f"got {type(j.get('data'))}")
return j["data"]
def test_job_data_has_required_fields(data):
"""Each job must have task_id, job_type, job_name, status, started_at"""
if not data:
print(" SKIP (no jobs) test_job_data_fields")
return
j = data[0]
for field in ["task_id", "job_type", "job_name", "status", "started_at"]:
check(f"field '{field}' present", field in j, str(list(j.keys())))
check("task_id is string", isinstance(j.get("task_id"), str))
check("job_type is string", isinstance(j.get("job_type"), str))
def test_job_log_endpoint(task_id):
"""GET /api/job/<id>/log should return log content"""
r = requests.get(f"{BASE}/api/job/{task_id}/log")
check("log HTTP 200", r.status_code == 200, f"got {r.status_code}")
j = r.json()
check("log ok is true", j.get("ok") == True, f"got {j.get('ok')}")
if j.get("ok"):
d = j["data"]
check("log has content", "content" in d)
check("log has status", "status" in d)
if d.get("content"):
check("log content not empty", len(d["content"]) > 0, f"size={len(d['content'])}")
print(f" log status={d.get('status')}, size={len(d.get('content',''))}")
def test_jobs_page_html():
"""The jobs page HTML should be usable by a browser"""
r = requests.get(f"{BASE}/jobs")
check("page HTTP 200", r.status_code == 200, f"got {r.status_code}")
html = r.text
# Check key elements exist
check("has '任务监控' in title or body", "任务监控" in html)
check("has list div", 'id="list"' in html)
check("has refresh function", "refresh" in html and "function refresh" in html)
# ============================================================
# RED TEST: Check if the JS has syntax issues
# ============================================================
# Extract the jobs <script> block
start = html.find("function loadJobs()")
end = html.find("</script>", start)
js_block = html[max(0, start-50):end]
# Check for problematic Jinja2-escaping in JS
# Template source uses: class=\\\" which renders as class=\\\" in HTML
# In JS single-quoted strings, \" should just be " (or at worst literal \"")
# Let's check if the rendered JS has weird escaping
check("JS has no double-backslash in class attr",
'class=\\\\\"' not in js_block,
"Found double-escaped quotes in JS — Jinja2 issue")
# Check if the innerHTML assignment has correct quoting
inner_html_line = [l for l in js_block.split('\n') if 'innerHTML' in l]
if inner_html_line:
line = inner_html_line[0].strip()
# Should look like: el.innerHTML='<div class="...">...'
check("innerHTML uses single quotes for JS string",
"innerHTML='" in line or 'innerHTML="' in line,
f"Line: {line[:80]}")
print(f"\n JS block length: {len(js_block)} chars")
def test_js_can_parse_job_data():
"""Simulate what the browser's JS engine does with the API response"""
r = requests.get(f"{BASE}/api/jobs")
data = r.json()["data"]
if not data:
print(" SKIP (no jobs)")
return
# Simulate: var t = data[i]
t = data[0]
# Simulate: TYPE_NAMES[t.job_type] || t.job_type
TYPE_NAMES = {
"daily_report": "日报生成", "report_image": "日报图片",
"article_summary": "文章摘要", "article_learn": "文章学习",
"article_image": "文章图片"
}
type_name = TYPE_NAMES.get(t["job_type"], t["job_type"])
check("type_name resolved", type_name != t["job_type"] or t["job_type"] not in TYPE_NAMES,
f"Unrecognized type: {t['job_type']}")
# Simulate: STATUS_NAMES[t.status] || t.status
STATUS_NAMES = {"running": "运行中", "completed": "已完成", "failed": "失败"}
status_name = STATUS_NAMES.get(t["status"], t["status"])
check("status_name resolved", isinstance(status_name, str))
# Simulate: esc(t.job_name)
# The esc function uses textContent for safe HTML insertion
name = t.get("job_name", "未命名任务")
check("job_name is valid string", isinstance(name, str) and len(name) > 0)
# Simulate building the href
href = f'/job/{t["task_id"]}'
check("task_id in href", t["task_id"] in href)
def test_trigger_and_track():
"""Integration: trigger a task, verify it appears in job list"""
import time
unique_id = str(int(time.time() * 1000))[-8:]
# Trigger
r = requests.post(f"{BASE}/api/articles/summarize", json={
"url": f"https://mp.weixin.qq.com/s/tdd_test_{unique_id}",
"title": f"TDD Test {unique_id}"
})
check("trigger HTTP 200", r.status_code == 200)
j = r.json()
check("trigger returns ok", j.get("ok") == True)
task_id = j.get("data", {}).get("task_id", "")
check("got task_id", len(task_id) > 0, f"got: {task_id}")
if task_id:
# Wait briefly for DB write
import time; time.sleep(0.5)
# Check job list includes it
r2 = requests.get(f"{BASE}/api/jobs")
jobs = r2.json()["data"]
found = [x for x in jobs if x["task_id"] == task_id]
check("task appears in job list", len(found) > 0,
f"task_id={task_id} not in {[j['task_id'] for j in jobs[:5]]}")
if found:
job = found[0]
check("job status is running", job["status"] == "running",
f"got {job['status']}")
check("job_name matches", unique_id in job.get("job_name", ""),
f"got {job.get('job_name')}")
# Check log endpoint
r3 = requests.get(f"{BASE}/api/job/{task_id}/log")
check("log API HTTP 200", r3.status_code == 200)
if r3.status_code == 200:
log_data = r3.json()
# Log file may be empty if task just started; that's OK
log_content = log_data.get("data", {}).get("content", "")
check("log content present", isinstance(log_content, str),
f"type={type(log_content)}")
# ============================================================
print("=" * 60)
print("Hermes Jobs — Frontend Integration Tests")
print("=" * 60)
data = test_api_jobs_returns_valid_json()
test_job_data_has_required_fields(data)
if data:
test_job_log_endpoint(data[0]["task_id"])
test_jobs_page_html()
test_js_can_parse_job_data()
print("\n--- Integration Test ---")
test_trigger_and_track()
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 ✅")