Coverage for tests/test_app_lifespan.py: 98%

1"""Tests for the exported app's middleware, REST/MCP routes, and lifespans. 

2 

3The data layer is stubbed to keep tests offline. Most tests also stub the MCP 

4lifespan: its real session manager can only start once per app. One integration 

5test exercises the real manager and HTTP transport through the exported app. 

6""" 

7 

8import asyncio 

9import json 

10import logging 

11from contextlib import asynccontextmanager 

12from types import SimpleNamespace 

13 

14import pytest 

15from conftest import course_payload 

16from fastapi import HTTPException 

17from fastapi.exceptions import ResponseValidationError 

18from fastapi.testclient import TestClient 

19from httpx import ASGITransport, AsyncClient 

20 

21from data_api.api import api as api_module 

22from data_api.api.api import app 

23from data_api.core import config 

24from data_api.core.exceptions import ( 

25 DataNotAvailableException, 

26 UpstreamResponseException, 

27 UpstreamTimeoutException, 

28) 

29from data_api.data.manager import nthudata 

30from data_api.domain.buses import services as buses_services 

31from data_api.domain.courses import services as courses_services 

32 

33 

34@asynccontextmanager 

35async def _noop_lifespan(app): 

36 """Stand in for the MCP lifespan, which cannot be started twice.""" 

37 yield 

38 

39 

40class TestCombinedLifespan: 

41 """Tests for the startup tasks chained into the combined lifespan.""" 

42 

43 async def test_runs_startup_tasks_in_order(self, monkeypatch): 

44 """Test the combined lifespan pre-fetches data and updates both services.""" 

45 calls = [] 

46 

47 async def fake_prefetch(endpoints): 

48 calls.append(("prefetch", list(endpoints))) 

49 return {endpoint: True for endpoint in endpoints} 

50 

51 async def fake_buses_update(): 

52 calls.append(("buses", None)) 

53 

54 async def fake_courses_update(): 

55 calls.append(("courses", None)) 

56 

57 monkeypatch.setattr(api_module, "mcp_app", SimpleNamespace(lifespan=_noop_lifespan)) 

58 monkeypatch.setattr(nthudata, "prefetch", fake_prefetch) 

59 monkeypatch.setattr(buses_services.buses_service, "update_data", fake_buses_update) 

60 monkeypatch.setattr(courses_services.courses_service, "get_snapshot", fake_courses_update) 

61 

62 async with api_module.combined_lifespan(app): 

63 pass 

64 

65 assert [name for name, _ in calls] == ["prefetch", "buses", "courses"] 

66 assert calls[0][1] == config.PREFETCH_ENDPOINTS 

67 

68 async def test_reports_failed_endpoints_as_ascii(self, monkeypatch, caplog): 

69 """Lifecycle logs stay ASCII and summarize partial availability.""" 

70 

71 async def fake_prefetch(endpoints): 

72 return {endpoint: index == 0 for index, endpoint in enumerate(endpoints)} 

73 

74 async def noop(): 

75 return None 

76 

77 monkeypatch.setattr(api_module, "mcp_app", SimpleNamespace(lifespan=_noop_lifespan)) 

78 monkeypatch.setattr(nthudata, "prefetch", fake_prefetch) 

79 monkeypatch.setattr(buses_services.buses_service, "update_data", noop) 

80 monkeypatch.setattr(courses_services.courses_service, "get_snapshot", noop) 

81 

82 with caplog.at_level(logging.INFO): 

83 async with api_module.combined_lifespan(app): 

84 pass 

85 assert f"Data startup: 1/{len(config.PREFETCH_ENDPOINTS)} datasets usable" in caplog.text 

86 caplog.text.encode("ascii") 

87 

88 @pytest.mark.parametrize("startup_fails", [False, True]) 

89 async def test_exported_lifespan_orders_and_unwinds_contexts(self, monkeypatch, startup_fails): 

90 calls = [] 

91 

92 @asynccontextmanager 

93 async def fake_mcp_lifespan(application): 

94 assert application is app 

95 calls.append("mcp startup") 

96 try: 

97 yield 

98 finally: 

99 calls.append("mcp shutdown") 

100 

101 @asynccontextmanager 

102 async def fake_data_lifespan(application): 

103 assert application is app 

104 calls.append("data startup") 

105 if startup_fails: 

106 raise RuntimeError("startup failed") 

107 try: 

108 yield 

109 finally: 

110 calls.append("data shutdown") 

111 

112 monkeypatch.setattr(api_module, "mcp_app", SimpleNamespace(lifespan=fake_mcp_lifespan)) 

113 monkeypatch.setattr(api_module, "lifespan", fake_data_lifespan) 

114 

115 if startup_fails: 

116 with pytest.raises(RuntimeError, match="startup failed"): 

117 async with app.router.lifespan_context(app): 

118 pass 

119 assert calls == ["mcp startup", "data startup", "mcp shutdown"] 

120 else: 

121 async with app.router.lifespan_context(app): 

122 assert calls == ["mcp startup", "data startup"] 

123 assert calls == ["mcp startup", "data startup", "data shutdown", "mcp shutdown"] 

124 

125 

126FAKE_COURSE = course_payload(chinese_title="測試課程") 

127 

128 

129async def _fake_fetch(url: str, sha256=None): 

130 """Serve in-memory data in place of a request to data.nthusa.tw.""" 

131 if url.endswith("file_details.json"): 

132 return { 

133 "file_details": { 

134 "/": [ 

135 {"name": name, "version": "testcommithash"} 

136 for name in config.PREFETCH_ENDPOINTS 

137 ] 

138 } 

139 } 

140 if url.endswith("courses/latest.json"): 

141 return [FAKE_COURSE] 

142 if url.endswith("courses/sources.json"): 

143 return {"semesters": ["11510"]} 

144 if url.endswith("buses.json"): 

145 return {} 

146 return [] 

147 

148 

149class TestCoursesAfterStartup: 

150 """Tests that the app's own lifespan is attached to the app that gets served.""" 

151 

152 async def test_courses_endpoint_returns_data_after_startup(self, monkeypatch): 

153 """Test /courses/ is populated once the app's own lifespan has run. 

154 

155 Exercise actual validation and conversion against a fixed upstream response. 

156 """ 

157 monkeypatch.setattr(api_module, "mcp_app", SimpleNamespace(lifespan=_noop_lifespan)) 

158 monkeypatch.setattr(nthudata.fetcher, "fetch_json", _fake_fetch) 

159 

160 async with app.router.lifespan_context(app): 

161 assert courses_services.courses_service.state.usable, "courses data was not loaded" 

162 

163 async with AsyncClient( 

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

165 ) as client: 

166 response = await client.get("/courses") 

167 

168 assert response.status_code == 200 

169 assert len(response.json()) == 1 

170 assert response.json()[0]["chinese_title"] == "測試課程" 

171 

172 

173@pytest.mark.parametrize("failure", ["dependency", "response_validation"]) 

174async def test_unexpected_pre_response_errors_keep_safe_body_and_headers( 

175 monkeypatch, caplog, failure 

176): 

177 private = "Private bus failure at https://upstream.example/credentials" 

178 

179 async def update_data(): 

180 if failure == "dependency": 

181 raise RuntimeError(private) 

182 

183 monkeypatch.setattr(buses_services.buses_service, "update_data", update_data) 

184 if failure == "response_validation": 

185 monkeypatch.setattr( 

186 buses_services.buses_service, 

187 "get_route_info", 

188 lambda *_: [{"direction": private}], 

189 ) 

190 

191 async with AsyncClient( 

192 transport=ASGITransport(app=app, raise_app_exceptions=True), base_url="http://test" 

193 ) as client: 

194 response = await client.get("/buses/routes", headers={"Origin": "https://example.com"}) 

195 

196 assert response.status_code == 500 

197 assert response.json() == {"detail": "Internal server error"} 

198 assert "Private" not in response.text 

199 assert "https://upstream.example" not in response.text 

200 assert response.headers["Access-Control-Allow-Origin"] == "*" 

201 assert float(response.headers["X-Process-Time"]) >= 0 

202 assert "x-process-time" in response.headers["Access-Control-Expose-Headers"].lower() 

203 assert private in caplog.text 

204 assert any( 

205 record.message == "Unexpected request failure" and record.exc_info 

206 for record in caplog.records 

207 ) 

208 if failure == "response_validation": 

209 assert any( 

210 isinstance(record.exc_info[1], ResponseValidationError) 

211 for record in caplog.records 

212 if record.exc_info 

213 ) 

214 

215 

216@pytest.mark.parametrize( 

217 "error,status,detail", 

218 [ 

219 pytest.param( 

220 DataNotAvailableException("Private dataset state"), 

221 503, 

222 "Service temporarily unavailable", 

223 id="snapshot", 

224 ), 

225 pytest.param( 

226 UpstreamTimeoutException("Private upstream timeout"), 

227 504, 

228 "Upstream request timed out", 

229 id="upstream-timeout", 

230 ), 

231 pytest.param( 

232 UpstreamResponseException("Private upstream response"), 

233 502, 

234 "Invalid response from upstream service", 

235 id="upstream-response", 

236 ), 

237 pytest.param( 

238 HTTPException(status_code=409, detail="Expected conflict"), 

239 409, 

240 "Expected conflict", 

241 id="http-exception", 

242 ), 

243 ], 

244) 

245async def test_expected_dependency_errors_keep_status_and_headers( 

246 monkeypatch, error, status, detail 

247): 

248 async def update_data(): 

249 raise error 

250 

251 monkeypatch.setattr(buses_services.buses_service, "update_data", update_data) 

252 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: 

253 response = await client.get("/buses/routes", headers={"Origin": "https://example.com"}) 

254 

255 assert response.status_code == status 

256 assert response.json() == {"detail": detail} 

257 assert response.headers["Access-Control-Allow-Origin"] == "*" 

258 assert float(response.headers["X-Process-Time"]) >= 0 

259 

260 

261async def test_error_after_stream_starts_is_not_replaced(): 

262 sent = [] 

263 

264 async def fail_after_start(scope, receive, send): 

265 await send({"type": "http.response.start", "status": 200, "headers": []}) 

266 await send({"type": "http.response.body", "body": b"partial", "more_body": True}) 

267 raise RuntimeError("Late stream failure") 

268 

269 async def send(message): 

270 sent.append(message) 

271 

272 async def receive(): 

273 return {"type": "http.request", "body": b"", "more_body": False} 

274 

275 middleware = api_module._SafeUnexpectedErrorsMiddleware(fail_after_start) 

276 with pytest.raises(RuntimeError, match="Late stream failure"): 

277 await middleware({"type": "http"}, receive, send) 

278 

279 assert [message["type"] for message in sent] == ["http.response.start", "http.response.body"] 

280 

281 

282async def test_cancelled_request_is_not_formatted(): 

283 sent = [] 

284 

285 async def cancel(scope, receive, send): 

286 raise asyncio.CancelledError() 

287 

288 async def send(message): 

289 sent.append(message) 

290 

291 async def receive(): 

292 return {"type": "http.request", "body": b"", "more_body": False} 

293 

294 middleware = api_module._SafeUnexpectedErrorsMiddleware(cancel) 

295 with pytest.raises(asyncio.CancelledError): 

296 await middleware({"type": "http"}, receive, send) 

297 

298 assert not sent 

299 

300 

301def test_exported_app_serves_rest_and_mcp_with_middleware(monkeypatch): 

302 """Start the real ASGI app and MCP manager without external data access.""" 

303 

304 async def fake_prefetch(endpoints): 

305 return {endpoint: True for endpoint in endpoints} 

306 

307 async def noop(): 

308 return None 

309 

310 monkeypatch.setattr(nthudata, "prefetch", fake_prefetch) 

311 monkeypatch.setattr(nthudata.fetcher, "fetch_json", _fake_fetch) 

312 monkeypatch.setattr(buses_services.buses_service, "update_data", noop) 

313 origin = "https://example.com" 

314 

315 with TestClient(app) as client: 

316 response = client.get("/courses", headers={"Origin": origin}) 

317 assert response.status_code == 200 

318 assert [course["id"] for course in response.json()] == [FAKE_COURSE["id"]] 

319 assert float(response.headers["X-Process-Time"]) >= 0 

320 assert response.headers["Access-Control-Allow-Origin"] == "*" 

321 assert response.headers["X-Total-Count"] == "1" 

322 assert response.headers["X-Data-Commit-Hash"] == "testcommithash" 

323 exposed = { 

324 header.strip().lower() 

325 for header in response.headers["Access-Control-Expose-Headers"].split(",") 

326 } 

327 assert {"x-total-count", "x-data-commit-hash", "x-process-time"} <= exposed 

328 

329 for path, method in [("/courses", "GET"), ("/mcp", "POST")]: 

330 response = client.options( 

331 path, 

332 headers={ 

333 "Origin": origin, 

334 "Access-Control-Request-Method": method, 

335 "Access-Control-Request-Headers": "Content-Type, MCP-Protocol-Version", 

336 }, 

337 ) 

338 assert response.status_code == 200 

339 assert response.headers["Access-Control-Allow-Origin"] == "*" 

340 assert method in response.headers["Access-Control-Allow-Methods"] 

341 assert ( 

342 "mcp-protocol-version" in response.headers["Access-Control-Allow-Headers"].lower() 

343 ) 

344 assert float(response.headers["X-Process-Time"]) >= 0 

345 

346 response = client.get("/openapi.json") 

347 assert response.status_code == 200 

348 assert "/courses" in response.json()["paths"] 

349 

350 for method, params in [ 

351 ( 

352 "initialize", 

353 { 

354 "protocolVersion": "2025-03-26", 

355 "capabilities": {}, 

356 "clientInfo": {"name": "phase-a-test", "version": "1.0"}, 

357 }, 

358 ), 

359 ("tools/list", {}), 

360 ("tools/call", {"name": "search_courses", "arguments": {"limit": 1}}), 

361 ]: 

362 response = client.post( 

363 "/mcp", 

364 headers={ 

365 "Origin": origin, 

366 "Accept": "application/json, text/event-stream", 

367 }, 

368 json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params}, 

369 ) 

370 assert response.status_code == 200 

371 assert float(response.headers["X-Process-Time"]) >= 0 

372 assert response.headers["Access-Control-Allow-Origin"] == "*" 

373 messages = [ 

374 json.loads(line.removeprefix("data: ")) 

375 for line in response.text.splitlines() 

376 if line.startswith("data: ") 

377 ] 

378 assert len(messages) == 1 

379 assert "error" not in messages[0] 

380 result = messages[0]["result"] 

381 if method == "initialize": 

382 assert "serverInfo" in result 

383 elif method == "tools/list": 

384 assert "search_courses" in {tool["name"] for tool in result["tools"]} 

385 else: 

386 assert not result.get("isError", False) 

387 content = json.loads(result["content"][0]["text"]) 

388 assert content["count"] == 1 

389 assert content["courses"][0]["id"] == FAKE_COURSE["id"]