Coverage for tests/test_libraries.py: 99%

1"""Tests for libraries endpoints.""" 

2 

3from datetime import date 

4 

5import pytest 

6from httpx import ASGITransport, AsyncClient, Response 

7 

8from data_api.api.api import app 

9from data_api.core.exceptions import DataNotAvailableException 

10from data_api.domain.libraries import services 

11from data_api.utils.calendars import filter_calendar_events, get_event_date_range 

12 

13RSS_DATA = { 

14 "news": [ 

15 { 

16 "guid": "1", 

17 "category": "最新消息", 

18 "title": "Library news", 

19 "link": "https://www.lib.nthu.edu.tw/news/1", 

20 "pubDate": "Mon, 22 Sep 2025 10:00:00 +0800", 

21 "description": "hello", 

22 "author": "NTHU Library", 

23 "image": { 

24 "url": "https://www.lib.nthu.edu.tw/image/news/1.jpg", 

25 "title": "cover", 

26 "link": "https://www.lib.nthu.edu.tw/", 

27 }, 

28 } 

29 ], 

30 # The scraper emits null for missing RSS tags, including items without an image. 

31 "eresources": [ 

32 { 

33 "guid": None, 

34 "category": None, 

35 "title": "No image", 

36 "link": None, 

37 "pubDate": None, 

38 "description": "", 

39 "author": None, 

40 "image": None, 

41 } 

42 ], 

43} 

44 

45 

46def make_event(event_id, title, start, end, all_day, description=None): 

47 return { 

48 "id": event_id, 

49 "title": title, 

50 "description": description, 

51 "start": start, 

52 "end": end, 

53 "all_day": all_day, 

54 } 

55 

56 

57EVENTS = [ 

58 make_event( 

59 "a1", "08:00-22:00", "2025-09-24T08:00:00+08:00", "2025-09-24T22:00:00+08:00", False 

60 ), 

61 make_event("a2", "休館", "2025-09-25", "2025-09-26", True, "中秋節"), 

62 make_event("a3", "寒假閉館", "2026-01-20", "2026-01-23", True), 

63 make_event("a4", "Overnight", "2026-02-01T22:00:00+08:00", "2026-02-02T00:00:00+08:00", False), 

64] 

65 

66CALENDARS_DATA = [ 

67 { 

68 "id": "main", 

69 "name": "總圖書館開館時間", 

70 "description": None, 

71 "timezone": "Asia/Taipei", 

72 "url": "https://calendar.google.com/calendar/embed?src=main", 

73 "events": EVENTS, 

74 }, 

75 { 

76 "id": "hss", 

77 "name": "人社分館開館時間", 

78 "description": None, 

79 "timezone": "Asia/Taipei", 

80 "url": "https://calendar.google.com/calendar/embed?src=hss", 

81 "events": [], 

82 }, 

83] 

84 

85FAKE_DATA = { 

86 services.RSS_JSON_PATH: RSS_DATA, 

87 "libraries/calendars.json": CALENDARS_DATA, 

88 "calendars.json": [], 

89} 

90 

91 

92@pytest.fixture 

93async def client(): 

94 """Create async test client.""" 

95 async with AsyncClient( 

96 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True 

97 ) as client: 

98 yield client 

99 

100 

101@pytest.fixture 

102def fake_data(monkeypatch): 

103 """Serve the scraper JSON files from memory instead of data.nthusa.tw.""" 

104 

105 async def fake_get(endpoint_name): 

106 if endpoint_name in FAKE_DATA: 

107 return "fakehash", FAKE_DATA[endpoint_name] 

108 raise DataNotAvailableException("Dataset temporarily unavailable") 

109 

110 monkeypatch.setattr(services.nthudata, "get", fake_get) 

111 

112 

113@pytest.fixture 

114def no_data(monkeypatch): 

115 """Simulate data.nthusa.tw being unavailable.""" 

116 

117 async def fake_get(endpoint_name): 

118 raise DataNotAvailableException("Dataset temporarily unavailable") 

119 

120 monkeypatch.setattr(services.nthudata, "get", fake_get) 

121 

122 

123class TestEventFiltering: 

124 """Tests for event date handling and filtering.""" 

125 

126 def test_all_day_end_is_exclusive(self): 

127 assert get_event_date_range(EVENTS[2]) == (date(2026, 1, 20), date(2026, 1, 22)) 

128 

129 def test_timed_event(self): 

130 assert get_event_date_range(EVENTS[0]) == (date(2025, 9, 24), date(2025, 9, 24)) 

131 

132 def test_timed_event_ending_at_midnight(self): 

133 assert get_event_date_range(EVENTS[3]) == (date(2026, 2, 1), date(2026, 2, 1)) 

134 

135 def test_filter_by_date_range(self): 

136 events = filter_calendar_events(EVENTS, start=date(2025, 9, 25), end=date(2026, 1, 21)) 

137 assert [e["id"] for e in events] == ["a2", "a3"] 

138 

139 def test_filter_by_keyword_matches_description(self): 

140 events = filter_calendar_events(EVENTS, keyword="中秋") 

141 assert [e["id"] for e in events] == ["a2"] 

142 

143 def test_filter_keyword_is_case_insensitive(self): 

144 events = filter_calendar_events(EVENTS, keyword="overnight") 

145 assert [e["id"] for e in events] == ["a4"] 

146 

147 

148class TestLibraryRss: 

149 """Tests for the RSS endpoint backed by the scraper JSON.""" 

150 

151 async def test_get_rss(self, client: AsyncClient, fake_data): 

152 response = await client.get("/libraries/rss/news") 

153 assert response.status_code == 200 

154 assert response.headers["X-Data-Commit-Hash"] == "fakehash" 

155 data = response.json() 

156 assert len(data) == 1 

157 assert data[0]["title"] == "Library news" 

158 

159 async def test_nullable_fields(self, client: AsyncClient, fake_data): 

160 response = await client.get("/libraries/rss/eresources") 

161 assert response.status_code == 200 

162 assert response.json()[0]["image"] is None 

163 

164 async def test_missing_feed(self, client: AsyncClient, fake_data): 

165 response = await client.get("/libraries/rss/exhibit") 

166 assert response.status_code == 404 

167 

168 async def test_invalid_type(self, client: AsyncClient, fake_data): 

169 response = await client.get("/libraries/rss/unknown") 

170 assert response.status_code == 422 

171 

172 async def test_data_unavailable(self, client: AsyncClient, no_data): 

173 response = await client.get("/libraries/rss/news") 

174 assert response.status_code == 503 

175 

176 

177class TestLibraryCalendars: 

178 """Tests for the calendar endpoints.""" 

179 

180 async def test_list_calendars_without_events(self, client: AsyncClient, fake_data): 

181 response = await client.get("/calendars") 

182 assert response.status_code == 200 

183 data = response.json() 

184 assert [c["id"] for c in data] == ["library-main", "library-hss"] 

185 assert all(c["category"] == "library" and c["source"] == "NTHU Library" for c in data) 

186 assert all("events" not in c for c in data) 

187 

188 async def test_get_calendar(self, client: AsyncClient, fake_data): 

189 response = await client.get("/calendars/library-main") 

190 assert response.status_code == 200 

191 data = response.json() 

192 assert data["name"] == "總圖書館開館時間" 

193 assert data["url"] == CALENDARS_DATA[0]["url"] 

194 assert response.headers["X-Data-Commit-Hash"] == "fakehash" 

195 assert "events" not in data 

196 

197 async def test_calendar_missing_from_data(self, client: AsyncClient, fake_data): 

198 response = await client.get("/calendars/library-nanda") 

199 assert response.status_code == 404 

200 

201 async def test_invalid_calendar_id(self, client: AsyncClient, fake_data): 

202 response = await client.get("/calendars/library-unknown") 

203 assert response.status_code == 404 

204 

205 async def test_list_events(self, client: AsyncClient, fake_data): 

206 response = await client.get("/calendars/library-main/events") 

207 assert response.status_code == 200 

208 assert response.headers["X-Total-Count"] == "4" 

209 assert [e["id"] for e in response.json()] == ["a1", "a2", "a3", "a4"] 

210 

211 async def test_events_on_a_single_day(self, client: AsyncClient, fake_data): 

212 response = await client.get( 

213 "/calendars/library-main/events", params={"start": "2025-09-25", "end": "2025-09-25"} 

214 ) 

215 assert response.status_code == 200 

216 assert [e["id"] for e in response.json()] == ["a2"] 

217 

218 async def test_events_by_keyword(self, client: AsyncClient, fake_data): 

219 response = await client.get("/calendars/library-main/events", params={"keyword": "閉館"}) 

220 assert [e["id"] for e in response.json()] == ["a3"] 

221 

222 async def test_events_pagination(self, client: AsyncClient, fake_data): 

223 response = await client.get( 

224 "/calendars/library-main/events", params={"limit": 2, "offset": 1} 

225 ) 

226 assert response.headers["X-Total-Count"] == "4" 

227 assert [e["id"] for e in response.json()] == ["a2", "a3"] 

228 

229 async def test_events_invalid_range(self, client: AsyncClient, fake_data): 

230 response = await client.get( 

231 "/calendars/library-main/events", params={"start": "2025-10-01", "end": "2025-09-01"} 

232 ) 

233 assert response.status_code == 400 

234 

235 async def test_events_invalid_date(self, client: AsyncClient, fake_data): 

236 response = await client.get("/calendars/library-main/events", params={"start": "soon"}) 

237 assert response.status_code == 422 

238 

239 async def test_events_of_missing_calendar(self, client: AsyncClient, fake_data): 

240 response = await client.get("/calendars/library-nanda/events") 

241 assert response.status_code == 404 

242 

243 async def test_get_event(self, client: AsyncClient, fake_data): 

244 response = await client.get("/calendars/library-main/events/a2") 

245 assert response.status_code == 200 

246 assert response.json()["title"] == "休館" 

247 

248 async def test_get_missing_event(self, client: AsyncClient, fake_data): 

249 response = await client.get("/calendars/library-main/events/nope") 

250 assert response.status_code == 404 

251 

252 @pytest.mark.parametrize( 

253 "url", 

254 [ 

255 "/calendars", 

256 "/calendars/library-main", 

257 "/calendars/library-main/events", 

258 "/calendars/library-main/events/a1", 

259 ], 

260 ) 

261 async def test_data_unavailable(self, client: AsyncClient, no_data, url: str): 

262 response = await client.get(url) 

263 assert response.status_code == 503 

264 

265 @pytest.mark.parametrize( 

266 "url", 

267 [ 

268 "/libraries/calendars", 

269 "/libraries/calendars/main", 

270 "/libraries/calendars/main/events", 

271 "/libraries/calendars/main/events/a1", 

272 ], 

273 ) 

274 async def test_old_endpoints_removed(self, client, url): 

275 response = await client.get(url) 

276 assert response.status_code == 404 

277 assert "location" not in response.headers 

278 

279 

280class TestLibrariesLiveEndpoints: 

281 """Endpoints that still hit the library website directly.""" 

282 

283 @pytest.mark.parametrize( 

284 "url", 

285 [ 

286 "/libraries/spaces", 

287 "/libraries/lost-and-found", 

288 ], 

289 ) 

290 async def test_libraries_endpoints(self, client: AsyncClient, url: str): 

291 response = await client.get(url) 

292 assert response.status_code == 502 

293 assert response.json() == {"detail": "Upstream service unavailable"} 

294 

295 @pytest.mark.parametrize("empty", [False, True]) 

296 @pytest.mark.parametrize("path", ["/libraries/spaces"]) 

297 async def test_library_space( 

298 self, client: AsyncClient, mock_upstream, library_space_payload, empty, path 

299 ): 

300 if empty: 

301 library_space_payload["rows"] = [] 

302 mock_upstream(lambda request: Response(200, json=library_space_payload)) 

303 response = await client.get(path) 

304 assert response.status_code == 200 

305 assert not response.history 

306 assert "location" not in response.headers 

307 assert response.json() == ( 

308 [] 

309 if empty 

310 else [ 

311 { 

312 **library_space_payload["rows"][0], 

313 "spacetype": 1, 

314 "count": 12, 

315 } 

316 ] 

317 ) 

318 

319 @pytest.mark.parametrize("path", ["/libraries/lost-and-found"]) 

320 async def test_lost_items(self, client: AsyncClient, mock_upstream, lost_items_html, path): 

321 def handler(request): 

322 assert request.method == "POST" 

323 return Response(200, text=lost_items_html()) 

324 

325 mock_upstream(handler) 

326 response = await client.get(path) 

327 assert response.status_code == 200 

328 assert response.json() == [ 

329 { 

330 "序號": str(index), 

331 "拾獲時間": "2026-09-30", 

332 "拾獲地點": "Main library", 

333 "描述": "Book title", 

334 } 

335 for index in range(12) 

336 ] 

337 

338 @pytest.mark.parametrize("empty_page", ["table", "message"]) 

339 @pytest.mark.parametrize("path", ["/libraries/lost-and-found"]) 

340 async def test_lost_items_empty( 

341 self, 

342 client: AsyncClient, 

343 mock_upstream, 

344 lost_items_html, 

345 lost_items_empty_page, 

346 empty_page, 

347 path, 

348 ): 

349 html = lost_items_html(0) if empty_page == "table" else lost_items_empty_page 

350 mock_upstream(lambda request: Response(200, text=html)) 

351 response = await client.get(path) 

352 assert response.status_code == 200 

353 assert response.json() == []