test_agent_cron.py
1.89 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
#!/usr/bin/env python3
"""Test Agent: verify cron API surface (/cron page, /api/cron/list, /api/cron/logs/<id>)."""
import json
import os
import subprocess
import pytest
BASE_URL = "http://127.0.0.1:5001"
def _curl_status(path):
r = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", f"{BASE_URL}{path}"],
capture_output=True, text=True, timeout=10,
)
try:
return int(r.stdout.strip())
except ValueError:
return 0
def _curl_json(path):
r = subprocess.run(
["curl", "-s", f"{BASE_URL}{path}"],
capture_output=True, text=True, timeout=10,
)
try:
return json.loads(r.stdout) if r.stdout.strip() else {}
except Exception:
return {"_raw": r.stdout[:200]}
def test_cron_page_reachable():
assert _curl_status("/cron") == 200
def test_cron_list_returns_ok():
status = _curl_status("/api/cron/list")
data = _curl_json("/api/cron/list")
assert status == 200
assert data.get("ok") is True
assert isinstance(data.get("data"), list)
def test_cron_list_items_have_required_fields():
data = _curl_json("/api/cron/list")
jobs = data.get("data", [])
assert isinstance(jobs, list)
if jobs:
item = jobs[0]
required = ["id", "name", "schedule", "enabled", "last_run", "last_status", "next_run"]
for fld in required:
assert fld in item, f"missing {fld}: {list(item.keys())}"
def test_cron_logs_endpoint_exists():
data = _curl_json("/api/cron/list")
jobs = data.get("data", [])
if jobs:
job_id = jobs[0]["id"]
status = _curl_status(f"/api/cron/logs/{job_id}")
assert status in (200, 404)
def test_jobs_json_exists():
cron_jobs_file = os.path.expanduser("~/pixelphone/cron/jobs.json")
assert os.path.exists(cron_jobs_file), cron_jobs_file
if __name__ == "__main__":
pytest.main([__file__, "-v"])