test_s1_cron_route.py
1.59 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
#!/usr/bin/env python3
"""S1 RED test: Verify /cron/<job_id> route behavior."""
import subprocess, json, sys, os
os.chdir(os.path.expanduser("~/wechat-viewer"))
BASE = "http://127.0.0.1:5000"
FAIL = 0
def check(desc, cmd, expect):
global FAIL
r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
ok = expect(r)
status = "PASS" if ok else "FAIL"
if not ok:
FAIL += 1
print(f"[{status}] {desc}")
if not ok:
print(f" stdout[:200]: {r.stdout[:200]}")
print(f" stderr[:200]: {r.stderr[:200]}")
# 1. Get job IDs from API
print("=== Fetching job IDs ===")
r = subprocess.run(["curl", "-s", f"{BASE}/api/cron/list"], capture_output=True, text=True, timeout=10)
jobs_data = json.loads(r.stdout)
jobs = jobs_data.get("data", [])
print(f"Found {len(jobs)} jobs")
if not jobs:
print("SKIP: No cron jobs found, cannot test real job route")
sys.exit(0)
job_id = jobs[0]["id"]
print(f"Testing with job: {jobs[0]['name']} (id={job_id})")
# 2. Test existing job → 200
check(f"GET /cron/{job_id} returns 200",
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", f"{BASE}/cron/{job_id}"],
lambda r: r.stdout.strip() == "200")
# 3. Test nonexistent job → 404
check("GET /cron/nonexistent returns 404",
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", f"{BASE}/cron/nonexistent"],
lambda r: r.stdout.strip() == "404")
# 4. Test response contains job name
check(f"GET /cron/{job_id} contains job name",
["curl", "-s", f"{BASE}/cron/{job_id}"],
lambda r: jobs[0]["name"][:10] in r.stdout)
sys.exit(min(FAIL, 1))