test_pipeline.py
1.45 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
#!/usr/bin/env python3
"""Test Agent: pipeline test — verify learn/articles/cron data pipeline endpoints."""
import json
import subprocess
import pytest
BASE_URL = "http://127.0.0.1:5001"
def _curl_status_body(path):
r = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", f"{BASE_URL}{path}"],
capture_output=True, text=True, timeout=10,
)
status = int(r.stdout.strip() or 0)
r2 = subprocess.run(
["curl", "-s", f"{BASE_URL}{path}"],
capture_output=True, text=True, timeout=10,
)
try:
body = json.loads(r2.stdout) if r2.stdout.strip() else {}
except Exception:
body = {"_raw": r2.stdout[:200]}
return status, body
ENDPOINTS = [
"/api/cron/list",
"/api/learn-queue",
"/api/articles",
"/api/reports",
"/api/learn-results",
]
@pytest.mark.parametrize("ep", ENDPOINTS)
def test_pipeline_endpoint_returns_200(ep):
status, _ = _curl_status_body(ep)
assert status == 200, f"{ep} status={status}"
@pytest.mark.parametrize("ep", ENDPOINTS)
def test_pipeline_endpoint_returns_ok_true(ep):
_, data = _curl_status_body(ep)
assert data.get("ok") is True, f"{ep} body={data}"
@pytest.mark.parametrize("ep", ENDPOINTS)
def test_pipeline_endpoint_returns_list(ep):
_, data = _curl_status_body(ep)
assert isinstance(data.get("data"), list), f"{ep} data type={type(data.get('data'))}"
if __name__ == "__main__":
pytest.main([__file__, "-v"])