Coverage for tests/test_ping.py: 99%
190 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Offline readiness diagnostics and schema visibility."""
3from dataclasses import replace
4from datetime import datetime, timezone
6import httpx
7import pytest
8from test_data_manager import Clock
10from data_api.api import api as api_module
11from data_api.core.config import PREFETCH_ENDPOINTS
12from data_api.data.manager import nthudata
13from data_api.data.nthudata import Freshness, RefreshError
15EXPECTED_DATASETS = {
16 "/announcements.json",
17 "/announcements_list.json",
18 "/buses.json",
19 "/calendars.json",
20 "/courses/latest.json",
21 "/courses/sources.json",
22 "/dining.json",
23 "/directory.json",
24 "/libraries/calendars.json",
25 "/libraries/rss.json",
26 "/maps.json",
27 "/newsletters.json",
28}
31@pytest.fixture
32async def ping_runtime(monkeypatch):
33 clock = Clock()
34 calls = []
35 payloads = {
36 path: {} if path in {"/buses.json", "/maps.json", "/libraries/rss.json"} else []
37 for path in EXPECTED_DATASETS
38 }
39 payloads["/courses/sources.json"] = {"semesters": []}
41 async def handler(self, request):
42 path = request.url.path
43 calls.append(path)
44 if path == "/file_details.json":
45 return httpx.Response(
46 200,
47 json={
48 "file_details": {
49 "/": [{"name": name.lstrip("/"), "version": "fixture"} for name in payloads]
50 }
51 },
52 )
53 if path == "/libraries.json":
54 return httpx.Response(
55 200,
56 content=b"<!doctype html><html><body>Index</body></html>",
57 headers={"Content-Type": "text/html"},
58 )
59 return httpx.Response(200, json=payloads[path])
61 monkeypatch.setattr(nthudata, "clock", clock)
62 monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", handler)
63 async with api_module.lifespan(api_module.app):
64 async with httpx.AsyncClient(
65 transport=httpx.ASGITransport(app=api_module.app), base_url="http://test"
66 ) as client:
67 yield client, clock, calls
70async def test_ping_ready_after_startup_including_empty_datasets(ping_runtime):
71 client, _, calls = ping_runtime
72 assert "/libraries.json" not in calls
73 assert {f"/{path}" for path in PREFETCH_ENDPOINTS} == EXPECTED_DATASETS
74 assert EXPECTED_DATASETS <= set(calls)
75 calls_before = list(calls)
77 response = await client.get("/ping", headers={"Origin": "https://example.com"})
79 assert response.status_code == 200
80 assert response.headers["Cache-Control"] == "no-store"
81 assert response.headers["Access-Control-Allow-Origin"] == "*"
82 assert float(response.headers["X-Process-Time"]) >= 0
83 report = response.json()
84 assert report["status"] == "ok"
85 assert report["ready"] is True
86 assert datetime.fromisoformat(report["checked_at"]).utcoffset().total_seconds() == 0
87 assert set(report["datasets"]) == EXPECTED_DATASETS
88 for path, health in report["datasets"].items():
89 state = nthudata.states[path]
90 assert health == {
91 "usable": True,
92 "freshness": "current",
93 "check_due": False,
94 "version": "fixture",
95 "loaded_at": state.snapshot.loaded_at.isoformat().replace("+00:00", "Z"),
96 "published_at": None,
97 "last_checked_at": state.last_checked_at.isoformat().replace("+00:00", "Z"),
98 "last_refresh_attempt_at": (
99 state.last_refresh_attempt_at.isoformat().replace("+00:00", "Z")
100 ),
101 "last_refresh_success_at": (
102 state.last_refresh_success_at.isoformat().replace("+00:00", "Z")
103 ),
104 "last_error": None,
105 }
106 assert calls == calls_before
109@pytest.mark.parametrize(
110 "condition,expected_status",
111 [
112 ("ready", 200),
113 ("stale", 200),
114 ("unverified", 200),
115 ("expired", 200),
116 ("missing", 503),
117 ("unregistered", 503),
118 ("cold", 503),
119 ],
120)
121async def test_ping_head_matches_readiness_without_refresh(
122 ping_runtime, monkeypatch, condition, expected_status
123):
124 client, clock, calls = ping_runtime
125 path = "/libraries/rss.json"
126 state = nthudata.states[path]
127 if condition in {"stale", "unverified"}:
128 state.freshness = Freshness(condition)
129 elif condition == "expired":
130 clock.advance(nthudata.ttl)
131 elif condition == "missing":
132 state.snapshot = None
133 state.freshness = Freshness.UNAVAILABLE
134 elif condition == "unregistered":
135 monkeypatch.delitem(nthudata.states, path)
136 elif condition == "cold":
137 monkeypatch.setattr(nthudata, "states", {})
138 calls_before = list(calls)
139 states_before = {
140 path: (state.snapshot, state.freshness, state.next_check)
141 for path, state in nthudata.states.items()
142 }
144 response = await client.head("/ping", headers={"Origin": "https://example.com"})
145 get_response = await client.get("/ping")
147 assert response.status_code == get_response.status_code == expected_status
148 assert response.content == b""
149 assert response.headers["Cache-Control"] == "no-store"
150 assert response.headers["Content-Type"] == get_response.headers["Content-Type"]
151 assert "Content-Length" not in response.headers
152 assert response.headers["Access-Control-Allow-Origin"] == "*"
153 assert float(response.headers["X-Process-Time"]) >= 0
154 assert calls == calls_before
155 assert {
156 path: (state.snapshot, state.freshness, state.next_check)
157 for path, state in nthudata.states.items()
158 } == states_before
161async def test_ping_head_sends_no_asgi_body(ping_runtime):
162 sent = []
164 async def receive():
165 return {"type": "http.request", "body": b"", "more_body": False}
167 async def send(message):
168 sent.append(message)
170 await api_module.app(
171 {
172 "type": "http",
173 "asgi": {"version": "3.0"},
174 "http_version": "1.1",
175 "method": "HEAD",
176 "scheme": "http",
177 "path": "/ping",
178 "raw_path": b"/ping",
179 "query_string": b"",
180 "headers": [],
181 "server": ("test", 80),
182 "client": ("test", 123),
183 },
184 receive,
185 send,
186 )
188 assert sent[0]["type"] == "http.response.start"
189 assert sent[0]["status"] == 200
190 bodies = [message for message in sent if message["type"] == "http.response.body"]
191 assert bodies
192 assert all(message.get("body", b"") == b"" for message in bodies)
195async def test_ping_head_cors_preflight(ping_runtime):
196 client, _, calls = ping_runtime
197 calls_before = list(calls)
199 response = await client.options(
200 "/ping",
201 headers={
202 "Origin": "https://example.com",
203 "Access-Control-Request-Method": "HEAD",
204 },
205 )
207 assert response.status_code == 200
208 assert "HEAD" in response.headers["Access-Control-Allow-Methods"]
209 assert response.headers["Access-Control-Allow-Origin"] == "*"
210 assert calls == calls_before
213@pytest.mark.parametrize("freshness", [Freshness.STALE, Freshness.UNVERIFIED])
214async def test_ping_keeps_usable_fallback_ready(ping_runtime, freshness):
215 client, _, calls = ping_runtime
216 state = nthudata.states["/courses/latest.json"]
217 now = datetime.now(timezone.utc)
218 state.freshness = freshness
219 state.last_error = RefreshError("http_status", now, 502)
220 state.snapshot = replace(state.snapshot, published_at="2026-09-29T08:00:00+08:00")
221 calls_before = list(calls)
223 response = await client.get("/ping")
225 assert response.status_code == 200
226 assert response.json()["ready"] is True
227 assert response.json()["status"] == "degraded"
228 health = response.json()["datasets"]["/courses/latest.json"]
229 assert health["freshness"] == freshness
230 assert health["usable"] is True
231 assert health["published_at"] == "2026-09-29T08:00:00+08:00"
232 assert health["last_error"] == {
233 "category": "http_status",
234 "at": now.isoformat().replace("+00:00", "Z"),
235 "status_code": 502,
236 }
237 assert state.freshness == freshness
238 assert calls == calls_before
241async def test_ping_expired_check_does_not_refresh_or_mutate(ping_runtime):
242 client, clock, calls = ping_runtime
243 clock.advance(nthudata.ttl)
244 calls_before = list(calls)
245 snapshots = {path: state.snapshot for path, state in nthudata.states.items()}
247 for _ in range(2):
248 response = await client.get("/ping")
249 assert response.status_code == 200
250 assert response.json()["status"] == "degraded"
251 for health in response.json()["datasets"].values():
252 assert health["freshness"] == "current"
253 assert health["check_due"] is True
254 assert calls == calls_before
255 assert all(state.snapshot is snapshots[path] for path, state in nthudata.states.items())
258@pytest.mark.parametrize("never_registered", [True, False])
259async def test_ping_missing_dataset_is_unavailable(ping_runtime, monkeypatch, never_registered):
260 client, _, calls = ping_runtime
261 path = "/libraries/rss.json"
262 state = nthudata.states[path]
263 if never_registered:
264 monkeypatch.delitem(nthudata.states, path)
265 else:
266 state.snapshot = None
267 state.freshness = Freshness.UNAVAILABLE
268 state.last_error = RefreshError("timeout", datetime.now(timezone.utc))
269 calls_before = list(calls)
271 response = await client.get("/ping")
273 assert response.status_code == 503
274 assert response.headers["Cache-Control"] == "no-store"
275 report = response.json()
276 assert report["status"] == "unavailable"
277 assert report["ready"] is False
278 health = report["datasets"][path]
279 assert health["usable"] is False
280 assert health["freshness"] == "unavailable"
281 assert health["version"] is None
282 assert health["loaded_at"] is None
283 if never_registered:
284 assert health["check_due"] is True
285 assert health["last_checked_at"] is None
286 assert health["last_error"] is None
287 assert path not in nthudata.states
288 else:
289 assert health["last_error"]["category"] == "timeout"
290 assert health["last_error"]["status_code"] is None
291 assert calls == calls_before
294async def test_ping_reports_additional_registered_datasets(ping_runtime, monkeypatch):
295 client, _, _ = ping_runtime
296 monkeypatch.setattr(nthudata, "states", dict(nthudata.states))
297 nthudata.state_for("extra.json")
299 response = await client.get("/ping")
301 assert response.status_code == 503
302 assert response.json()["datasets"]["/extra.json"]["usable"] is False
305async def test_ping_cold_registry_is_not_ready(monkeypatch):
306 monkeypatch.setattr(nthudata, "states", {})
307 async with httpx.AsyncClient(
308 transport=httpx.ASGITransport(app=api_module.app), base_url="http://test"
309 ) as client:
310 response = await client.get("/ping")
311 assert response.status_code == 503
312 assert set(response.json()["datasets"]) == EXPECTED_DATASETS
313 assert all(not health["usable"] for health in response.json()["datasets"].values())
314 assert nthudata.states == {}
317async def test_ping_hidden_from_openapi_and_swagger(ping_runtime):
318 client, _, _ = ping_runtime
319 schema = await client.get("/openapi.json")
320 assert schema.status_code == 200
321 assert "/ping" not in schema.json()["paths"]
322 assert "/courses" in schema.json()["paths"]
323 assert "PingResponse" not in schema.json()["components"]["schemas"]
324 docs = await client.get("/docs")
325 assert docs.status_code == 200
326 assert "/openapi.json" in docs.text
327 assert "/ping" not in docs.text
328 assert (await client.get("/ping")).status_code == 200
329 assert (await client.head("/ping")).status_code == 200