Coverage for tests/test_calendars.py: 100%
253 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Campus calendar queries and published snapshot lifecycle."""
3import hashlib
4import json
5from copy import deepcopy
7import httpx
8import pytest
9from test_data_manager import Clock, Publisher
11from data_api.api.api import app
12from data_api.api.schemas.calendars import Calendar
13from data_api.data.manager import nthudata
14from data_api.data.nthudata import Freshness
17def make_event(event_id, title, start, end, all_day=True, description=None):
18 return {
19 "id": event_id,
20 "title": title,
21 "description": description,
22 "start": start,
23 "end": end,
24 "all_day": all_day,
25 }
28EVENTS = [
29 make_event("last", "Semester ends", "2026-10-31", "2026-11-01"),
30 make_event(
31 "overnight", "Workshop", "2026-10-03T22:00:00+08:00", "2026-10-04T00:00:00+08:00", False
32 ),
33 make_event("b", "Registration", "2026-10-01", "2026-10-02", description="Select C++ [AI]"),
34 make_event("spanning", "Selection period", "2026-09-28", "2026-10-03"),
35 make_event("before", "Holiday", "2026-09-30", "2026-10-01"),
36 make_event("a", "Classes begin", "2026-10-01", "2026-10-02"),
37 make_event(
38 "crossing", "Night event", "2026-10-05T22:00:00+08:00", "2026-10-06T01:00:00+08:00", False
39 ),
40]
41CALENDARS = [
42 {
43 "id": "academic",
44 "name": "Academic calendar",
45 "description": None,
46 "timezone": "Asia/Taipei",
47 "events": EVENTS,
48 },
49 {
50 "id": "future-calendar",
51 "name": "Additional calendar",
52 "description": None,
53 "timezone": "Asia/Taipei",
54 "events": [],
55 },
56]
57PATHS = [
58 "/calendars",
59 "/calendars/academic",
60 "/calendars/academic/events",
61 "/calendars/academic/events/a",
62]
63EVENTS_PATH = "/calendars/academic/events"
66@pytest.fixture
67def library_publisher():
68 return Publisher([], "libraries/calendars.json")
71@pytest.fixture
72async def runtime(monkeypatch, library_publisher):
73 clock, publisher = Clock(), Publisher(deepcopy(CALENDARS), "calendars.json")
74 monkeypatch.setattr(nthudata, "clock", clock)
76 async def handler(request):
77 if request.url.path == "/libraries/calendars.json":
78 return await library_publisher(request)
79 response = await publisher(request)
80 if request.url.path == "/file_details.json" and response.status_code == 200:
81 manifest = response.json()
82 manifest["file_details"]["/"].extend(library_publisher.manifest()["file_details"]["/"])
83 return httpx.Response(200, json=manifest)
84 return response
86 async with nthudata.lifespan(httpx.AsyncClient(transport=httpx.MockTransport(handler))):
87 async with httpx.AsyncClient(
88 transport=httpx.ASGITransport(app=app), base_url="http://test"
89 ) as client:
90 yield client, publisher, clock
93async def test_calendar_metadata_and_dynamic_ids(runtime):
94 client, publisher, _ = runtime
95 response = await client.get("/calendars")
96 assert response.status_code == 200
97 assert response.json() == [
98 Calendar.model_validate(calendar).model_dump(mode="json") for calendar in CALENDARS
99 ]
100 assert (
101 response.headers["X-Data-Commit-Hash"]
102 == hashlib.sha256(json.dumps(["a", "a"]).encode()).hexdigest()
103 )
104 for metadata in response.json():
105 detail = await client.get(f"/calendars/{metadata['id']}")
106 assert detail.status_code == 200
107 assert detail.json() == metadata
108 assert detail.headers["X-Data-Commit-Hash"] == "a"
109 assert publisher.calls.count("/calendars.json") == 1
112@pytest.mark.parametrize(
113 "params,expected",
114 [
115 ({}, ["spanning", "before", "a", "b", "overnight", "crossing", "last"]),
116 ({"start": "2026-10-01", "end": "2026-10-01"}, ["spanning", "a", "b"]),
117 ({"start": "2026-10-02", "end": "2026-10-02"}, ["spanning"]),
118 ({"start": "2026-10-03", "end": "2026-10-03"}, ["overnight"]),
119 ({"start": "2026-10-04", "end": "2026-10-04"}, []),
120 ({"start": "2026-10-06", "end": "2026-10-06"}, ["crossing"]),
121 ({"start": "2026-10-07"}, ["last"]),
122 ({"end": "2026-09-30"}, ["spanning", "before"]),
123 ({"keyword": " REGISTRATION "}, ["b"]),
124 ({"keyword": "c++ [ai]"}, ["b"]),
125 ({"keyword": ".*"}, []),
126 ({"keyword": "select"}, ["spanning", "b"]),
127 ({"start": "2026-10-02", "end": "2026-10-02", "keyword": "select"}, ["spanning"]),
128 ({"start": "2026-10-07", "keyword": "select"}, []),
129 ],
130)
131async def test_event_filters(runtime, params, expected):
132 client, publisher, _ = runtime
133 response = await client.get(EVENTS_PATH, params=params)
134 assert response.status_code == 200
135 assert [event["id"] for event in response.json()] == expected
136 assert response.headers["X-Total-Count"] == str(len(expected))
137 assert response.headers["X-Data-Commit-Hash"] == "a"
138 assert publisher.payload == CALENDARS
141async def test_pagination_after_filtering_and_sorting(runtime):
142 client, _, _ = runtime
143 response = await client.get(
144 EVENTS_PATH,
145 params={"start": "2026-10-01", "end": "2026-10-03", "limit": 2, "offset": 1},
146 )
147 assert response.status_code == 200
148 assert [event["id"] for event in response.json()] == ["a", "b"]
149 assert response.headers["X-Total-Count"] == "4"
150 response = await client.get(EVENTS_PATH, params={"offset": 1000})
151 assert response.json() == []
152 assert response.headers["X-Total-Count"] == "7"
155async def test_default_limit_and_blank_keyword(runtime):
156 client, publisher, _ = runtime
157 publisher.payload[0]["events"] = [
158 make_event(f"{index:03}", "Event", "2026-10-01", "2026-10-02") for index in range(101)
159 ]
160 response = await client.get(EVENTS_PATH, params={"keyword": " "})
161 assert len(response.json()) == 100
162 assert response.headers["X-Total-Count"] == "101"
163 response = await client.get(EVENTS_PATH, params={"limit": 1000})
164 assert len(response.json()) == 101
167@pytest.mark.parametrize(
168 "params,status",
169 [
170 ({"start": "2026-11-01", "end": "2026-10-01"}, 400),
171 ({"start": "soon"}, 422),
172 ({"end": "2026-02-30"}, 422),
173 ({"limit": 0}, 422),
174 ({"limit": 1001}, 422),
175 ({"offset": -1}, 422),
176 ],
177)
178async def test_invalid_queries_do_not_fetch(runtime, params, status):
179 client, publisher, _ = runtime
180 response = await client.get(EVENTS_PATH, params=params)
181 assert response.status_code == status
182 assert publisher.calls == []
185async def test_single_event_preserves_source_shape(runtime):
186 client, _, _ = runtime
187 response = await client.get(f"{EVENTS_PATH}/b")
188 assert response.status_code == 200
189 assert response.json() == EVENTS[2]
190 assert response.headers["X-Data-Commit-Hash"] == "a"
193@pytest.mark.parametrize(
194 "path",
195 [
196 "/calendars/missing",
197 "/calendars/missing/events",
198 "/calendars/missing/events/a",
199 "/calendars/future-calendar/events/a",
200 f"{EVENTS_PATH}/missing",
201 ],
202)
203async def test_missing_calendar_or_event(runtime, path):
204 client, _, _ = runtime
205 assert (await client.get(path)).status_code == 404
208async def test_empty_calendar_and_empty_dataset(runtime):
209 client, publisher, clock = runtime
210 response = await client.get("/calendars/future-calendar/events")
211 assert response.status_code == 200
212 assert response.json() == []
213 assert response.headers["X-Total-Count"] == "0"
214 publisher.payload, publisher.version = [], "empty"
215 clock.advance()
216 response = await client.get("/calendars")
217 assert response.status_code == 200
218 assert response.json() == []
219 assert (
220 response.headers["X-Data-Commit-Hash"]
221 == hashlib.sha256(json.dumps(["empty", "a"]).encode()).hexdigest()
222 )
223 assert (await client.get(EVENTS_PATH)).status_code == 404
226@pytest.mark.parametrize("path", PATHS)
227async def test_unavailable_dataset(runtime, path):
228 client, publisher, _ = runtime
229 publisher.data_error = 503
230 response = await client.get(path)
231 assert response.status_code == 503
232 assert response.json() == {"detail": "Service temporarily unavailable"}
235@pytest.mark.parametrize("path", PATHS)
236async def test_unknown_version_header_is_omitted(runtime, path):
237 client, publisher, _ = runtime
238 publisher.version = None
239 response = await client.get(path)
240 assert response.status_code == 200
241 assert "X-Data-Commit-Hash" not in response.headers
244@pytest.mark.parametrize(
245 "invalid",
246 [
247 {},
248 [{"id": "academic"}],
249 [{**CALENDARS[0], "events": None}],
250 [{**CALENDARS[0], "events": [{**EVENTS[0], "start": "invalid"}]}],
251 [{**CALENDARS[0], "events": [{**EVENTS[0], "end": "2026-09-01"}]}],
252 [{**CALENDARS[0], "events": [{**EVENTS[0], "all_day": "true"}]}],
253 [{**CALENDARS[0], "id": "library-main"}],
254 [CALENDARS[0], CALENDARS[0]],
255 ],
256)
257@pytest.mark.parametrize("initial_loaded", [True, False])
258async def test_invalid_payload_preserves_snapshot_and_fails_cold_start(
259 runtime, invalid, initial_loaded
260):
261 client, publisher, clock = runtime
262 if initial_loaded:
263 original = (await client.get(EVENTS_PATH)).json()
264 publisher.payload, publisher.version = invalid, "invalid"
265 clock.advance()
266 response = await client.get(EVENTS_PATH)
267 if initial_loaded:
268 assert response.status_code == 200
269 assert response.json() == original
270 assert response.headers["X-Data-Commit-Hash"] == "a"
271 assert nthudata.states["/calendars.json"].freshness == Freshness.STALE
272 else:
273 assert response.status_code == 503
274 assert nthudata.states["/calendars.json"].freshness == Freshness.UNAVAILABLE
277async def test_runtime_refresh_and_upstream_failure(runtime):
278 client, publisher, clock = runtime
279 assert (await client.get(f"{EVENTS_PATH}/b")).json()["title"] == "Registration"
280 publisher.version = "b"
281 publisher.payload[0]["events"][2]["title"] = "Updated registration"
282 clock.advance()
283 response = await client.get(f"{EVENTS_PATH}/b")
284 assert response.json()["title"] == "Updated registration"
285 assert response.headers["X-Data-Commit-Hash"] == "b"
286 publisher.data_error = 503
287 clock.advance()
288 response = await client.get(f"{EVENTS_PATH}/b")
289 assert response.status_code == 200
290 assert response.json()["title"] == "Updated registration"
291 assert response.headers["X-Data-Commit-Hash"] == "b"
294def test_openapi_calendar_contract():
295 schema = app.openapi()
296 assert not any(path.startswith("/libraries/calendars") for path in schema["paths"])
297 assert not any(name.startswith("LibraryCalendar") for name in schema["components"]["schemas"])
298 assert schema["paths"]["/calendars"]["get"]["operationId"] == "getAllCalendars"
299 search = schema["paths"][EVENTS_PATH.replace("academic", "{calendar_id}")]["get"]
300 assert search["operationId"] == "searchCalendarEvents"
301 assert {parameter["name"] for parameter in search["parameters"]} == {
302 "calendar_id",
303 "start",
304 "end",
305 "keyword",
306 "limit",
307 "offset",
308 }
309 for path, operations in schema["paths"].items():
310 if "{calendar_id}" not in path:
311 continue
312 parameter = next(
313 parameter
314 for parameter in operations["get"]["parameters"]
315 if parameter["name"] == "calendar_id"
316 )
317 assert "/calendars " in parameter["description"]
318 assert "/calendars/" not in parameter["description"]
321async def test_unified_list_and_all_library_branches(runtime, library_publisher):
322 client, publisher, _ = runtime
323 library_publisher.version = "library-version"
324 library_publisher.payload = [
325 {
326 "id": branch,
327 "name": f"{branch} opening hours",
328 "url": f"https://calendar.google.com/calendar/embed?src={branch}",
329 "events": deepcopy(EVENTS),
330 }
331 for branch in ["main", "hss", "nanda"]
332 ]
333 original = deepcopy(library_publisher.payload)
334 response = await client.get("/calendars")
335 assert response.status_code == 200
336 assert [calendar["id"] for calendar in response.json()] == [
337 "academic",
338 "future-calendar",
339 "library-main",
340 "library-hss",
341 "library-nanda",
342 ]
343 for metadata in response.json()[2:]:
344 assert metadata["category"] == "library"
345 assert metadata["source"] == "NTHU Library"
346 assert metadata["description"] is None
347 assert "events" not in metadata
348 path = f"/calendars/{metadata['id']}"
349 detail = await client.get(path)
350 assert detail.json() == metadata
351 assert detail.headers["X-Data-Commit-Hash"] == "library-version"
352 events = await client.get(
353 f"{path}/events",
354 params={"start": "2026-10-01", "end": "2026-10-03", "limit": 2, "offset": 1},
355 )
356 assert events.status_code == 200
357 assert events.headers["X-Total-Count"] == "4"
358 assert events.headers["X-Data-Commit-Hash"] == "library-version"
359 assert [event["id"] for event in events.json()] == ["a", "b"]
360 keyword = await client.get(f"{path}/events", params={"keyword": "c++ [ai]"})
361 assert [event["id"] for event in keyword.json()] == ["b"]
362 event = await client.get(f"{path}/events/b")
363 assert event.json() == EVENTS[2]
364 assert event.headers["X-Data-Commit-Hash"] == "library-version"
365 assert (await client.get(f"{path}/events/missing")).status_code == 404
366 assert publisher.payload == CALENDARS
367 assert library_publisher.payload == original
368 assert library_publisher.calls.count("/libraries/calendars.json") == 1
371@pytest.mark.parametrize("unavailable", ["campus", "library"])
372async def test_individual_calendars_do_not_depend_on_other_dataset(
373 runtime, library_publisher, unavailable
374):
375 client, publisher, _ = runtime
376 library_publisher.payload = [
377 {"id": "main", "url": "https://example.com/calendar", "events": deepcopy(EVENTS)}
378 ]
379 if unavailable == "campus":
380 publisher.data_error = 503
381 path = "/calendars/library-main"
382 else:
383 library_publisher.data_error = 503
384 path = "/calendars/academic"
385 for suffix in ["", "/events", "/events/b"]:
386 assert (await client.get(f"{path}{suffix}")).status_code == 200
387 assert (await client.get("/calendars")).status_code == 503
390async def test_library_refresh_versions_and_stale_snapshot(runtime, library_publisher):
391 client, publisher, clock = runtime
392 library_publisher.payload = [
393 {"id": "main", "url": "https://example.com/calendar", "events": deepcopy(EVENTS)}
394 ]
395 first = await client.get("/calendars")
396 library_publisher.version = "b"
397 library_publisher.payload[0]["events"][2]["title"] = "Updated library event"
398 clock.advance()
399 second = await client.get("/calendars")
400 assert first.headers["X-Data-Commit-Hash"] != second.headers["X-Data-Commit-Hash"]
401 event = await client.get("/calendars/library-main/events/b")
402 assert event.json()["title"] == "Updated library event"
403 assert event.headers["X-Data-Commit-Hash"] == "b"
404 assert publisher.version == "a"
405 publisher.version = "campus-b"
406 clock.advance()
407 third = await client.get("/calendars")
408 assert second.headers["X-Data-Commit-Hash"] != third.headers["X-Data-Commit-Hash"]
409 library_publisher.version = "c"
410 library_publisher.data_error = 503
411 clock.advance()
412 event = await client.get("/calendars/library-main/events/b")
413 assert event.status_code == 200
414 assert event.json()["title"] == "Updated library event"
415 assert event.headers["X-Data-Commit-Hash"] == "b"
416 assert nthudata.states["/libraries/calendars.json"].freshness == Freshness.STALE
419@pytest.mark.parametrize(
420 "invalid",
421 [
422 {"id": "main", "events": []},
423 {"id": "main", "url": "invalid", "events": []},
424 {"id": "main", "url": "https://example.com", "events": None},
425 {
426 "id": "main",
427 "url": "https://example.com",
428 "events": [{**EVENTS[0], "all_day": "true"}],
429 },
430 {
431 "id": "main",
432 "url": "https://example.com",
433 "events": [{**EVENTS[0], "end": "2026-01-01"}],
434 },
435 ],
436)
437@pytest.mark.parametrize("initial_loaded", [False, True])
438async def test_invalid_library_payload_preserves_snapshot(
439 runtime, library_publisher, invalid, initial_loaded
440):
441 client, _, clock = runtime
442 library_publisher.payload = [
443 {"id": "main", "url": "https://example.com", "events": deepcopy(EVENTS)}
444 ]
445 path = "/calendars/library-main/events"
446 if initial_loaded:
447 original = (await client.get(path)).json()
448 library_publisher.payload, library_publisher.version = [invalid], "invalid"
449 clock.advance()
450 response = await client.get(path)
451 if initial_loaded:
452 assert response.status_code == 200
453 assert response.json() == original
454 assert response.headers["X-Data-Commit-Hash"] == "a"
455 assert nthudata.states["/libraries/calendars.json"].freshness == Freshness.STALE
456 else:
457 assert response.status_code == 503
458 assert nthudata.states["/libraries/calendars.json"].freshness == Freshness.UNAVAILABLE
461async def test_unknown_library_version_omits_aggregate_header(runtime, library_publisher):
462 client, _, _ = runtime
463 library_publisher.version = None
464 response = await client.get("/calendars")
465 assert response.status_code == 200
466 assert "X-Data-Commit-Hash" not in response.headers