test_cron.py
3.94 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
#!/usr/bin/env python3
"""Test S1: /api/cron/list and /api/cron/logs routes."""
import sys, os, json, tempfile, shutil
# Add app dir to path
sys.path.insert(0, os.path.expanduser("~/wechat-viewer"))
# ── Setup: ensure Flask is running ──
import subprocess
def check_flask():
r = subprocess.run(["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"http://127.0.0.1:5000/"],
capture_output=True, text=True, timeout=5)
if r.stdout.strip() != "200":
print("SKIP: Flask not running (HTTP {})".format(r.stdout.strip()))
return False
return True
def curl_get(path):
r = subprocess.run(["curl", "-s", "http://127.0.0.1:5000" + path],
capture_output=True, text=True, timeout=10)
try:
return r.returncode, json.loads(r.stdout)
except:
return r.returncode, {"_raw": r.stdout[:200]}
passed = 0
failed = 0
def test(name, condition, detail=""):
global passed, failed
if condition:
passed += 1
print(f" PASS: {name}")
else:
failed += 1
print(f" FAIL: {name} {detail}")
# ── Run tests ──
if not check_flask():
sys.exit(0)
print("\n=== S1: Cron API Tests ===\n")
# ── Test 1: /api/cron/list returns valid JSON ──
code, data = curl_get("/api/cron/list")
test("GET /api/cron/list returns 200", code == 0 and data.get("ok"), f"got {code}, {data}")
test("GET /api/cron/list has data array", data.get("ok") and isinstance(data.get("data"), list),
f"data={data}")
if data.get("ok") and isinstance(data.get("data"), list):
items = data["data"]
test("GET /api/cron/list has items", len(items) > 0, f"count={len(items)}")
# Check field structure on first item
if items:
item = items[0]
required = ["id", "name", "schedule", "enabled", "last_run", "last_status", "next_run"]
for fld in required:
test(f"GET /api/cron/list item has '{fld}'", fld in item, str(list(item.keys())[:10]))
# ── Test 2: /api/cron/logs/<job_id> ──
# Find an active job with output
active_jobs = []
if data.get("ok") and isinstance(data.get("data"), list):
active_jobs = [j for j in data["data"] if j.get("last_run")]
if active_jobs:
job_id = active_jobs[0]["id"]
job_name = active_jobs[0]["name"]
code2, data2 = curl_get(f"/api/cron/logs/{job_id}")
test(f"GET /api/cron/logs/{job_id} returns 200", code2 == 0 and data2.get("ok"),
f"got {code2}, {str(data2)[:200]}")
if data2.get("ok") and isinstance(data2.get("data"), list):
logs = data2["data"]
test(f"GET /api/cron/logs/{job_id} has items", len(logs) > 0,
f"count={len(logs)}, job={job_name}")
if logs:
log = logs[0]
test("Log entry has timestamp", "timestamp" in log, str(list(log.keys())[:10]))
test("Log entry has content (not empty)", bool(log.get("content")),
f"content={log.get('content','')[:50]}")
else:
test("Has at least one active job", len(active_jobs) > 0, "No jobs with last_run!")
if not active_jobs:
# Try any job_id from jobs.json
jobs_file = os.path.expanduser("~/pixelphone/cron/jobs.json")
if os.path.exists(jobs_file):
with open(jobs_file) as f:
jd = json.load(f)
if jd.get("jobs"):
job_id = jd["jobs"][0]["id"]
code2, data2 = curl_get(f"/api/cron/logs/{job_id}")
test(f"GET /api/cron/logs/{job_id} returns 200", code2 == 0 and data2.get("ok"),
f"got {code2}, {str(data2)[:200]}")
# ── Test 3: Invalid job_id → graceful error ──
code3, data3 = curl_get("/api/cron/logs/nonexistent123")
test("GET /api/cron/logs/nonexistent → ok=False or 404",
(code3 != 0) or (not data3.get("ok")),
f"got code={code3}, {str(data3)[:200]}")
# ── Summary ──
print(f"\n=== Result: {passed} passed, {failed} failed ===")
sys.exit(0 if failed == 0 else 1)