Coverage for tests/test_upstream_errors.py: 100%

1"""REST and MCP contracts for live upstream failures.""" 

2 

3import json 

4 

5import httpx 

6import pytest 

7from fastmcp import Client 

8 

9from data_api.api.api import app, mcp_app 

10from data_api.domain.energy.services import energy_service 

11from data_api.domain.libraries.services import libraries_service 

12from data_api.mcp.server import mcp 

13 

14LIVE_CALLS = [ 

15 pytest.param( 

16 "/energy/electricity", 

17 "get_energy_usage", 

18 {}, 

19 energy_service, 

20 "get_realtime_electricity_usage", 

21 id="energy-current", 

22 ), 

23 pytest.param( 

24 "/libraries/spaces", 

25 "get_library_info", 

26 {"info_type": "space"}, 

27 libraries_service, 

28 "get_space_availability", 

29 id="spaces-current", 

30 ), 

31 pytest.param( 

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

33 "get_library_info", 

34 {"info_type": "lost_and_found"}, 

35 libraries_service, 

36 "get_lost_and_found_items", 

37 id="lost-items-current", 

38 ), 

39] 

40CALL_PARAMETERS = "path,tool,arguments,service,method" 

41LOST_ITEM_HEADERS = ( 

42 "\u5e8f\u865f", 

43 "\u62fe\u7372\u6642\u9593", 

44 "\u62fe\u7372\u5730\u9ede", 

45 "\u63cf\u8ff0", 

46) 

47 

48 

49@pytest.fixture 

50async def rest(): 

51 async with httpx.AsyncClient( 

52 transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), 

53 base_url="http://test", 

54 ) as client: 

55 yield client 

56 

57 

58@pytest.mark.parametrize(CALL_PARAMETERS, LIVE_CALLS) 

59@pytest.mark.parametrize( 

60 "failure", 

61 [ 

62 httpx.ConnectTimeout, 

63 httpx.ReadTimeout, 

64 httpx.WriteTimeout, 

65 httpx.PoolTimeout, 

66 httpx.ConnectError, 

67 httpx.ReadError, 

68 httpx.RequestError, 

69 403, 

70 404, 

71 500, 

72 503, 

73 ], 

74) 

75async def test_live_http_failures( 

76 rest, mock_upstream, caplog, path, tool, arguments, service, method, failure 

77): 

78 def handler(request): 

79 if isinstance(failure, int): 

80 return httpx.Response(failure, text="Private upstream response") 

81 raise failure("Private upstream exception", request=request) 

82 

83 mock_upstream(handler) 

84 timed_out = not isinstance(failure, int) and issubclass(failure, httpx.TimeoutException) 

85 status = 504 if timed_out else 502 

86 detail = "Upstream request timed out" if timed_out else "Upstream service unavailable" 

87 

88 response = await rest.get(path, headers={"Origin": "https://example.com"}) 

89 assert response.status_code == status 

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

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

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

93 async with Client(mcp) as client: 

94 result = await client.call_tool(tool, arguments, raise_on_error=False) 

95 assert result.is_error 

96 assert result.content[0].text == detail 

97 assert result.data is None 

98 assert any(record.exc_info for record in caplog.records) 

99 assert "Traceback" in caplog.text 

100 assert "http" in caplog.text if isinstance(failure, int) else "Private upstream" in caplog.text 

101 

102 

103@pytest.mark.parametrize(CALL_PARAMETERS, LIVE_CALLS) 

104async def test_unexpected_live_failures_are_safe( 

105 rest, monkeypatch, caplog, path, tool, arguments, service, method 

106): 

107 async def fail(): 

108 raise RuntimeError("Private internal failure at https://upstream.example") 

109 

110 monkeypatch.setattr(service, method, fail) 

111 response = await rest.get(path, headers={"Origin": "https://example.com"}) 

112 assert response.status_code == 500 

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

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

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

116 async with Client(mcp) as client: 

117 result = await client.call_tool(tool, arguments, raise_on_error=False) 

118 assert result.is_error 

119 assert result.content[0].text == "Internal server error" 

120 assert "Private internal failure" in caplog.text 

121 

122 

123@pytest.mark.parametrize( 

124 "path,tool,arguments,body", 

125 [ 

126 ("/energy/electricity", "get_energy_usage", {}, ""), 

127 ("/energy/electricity", "get_energy_usage", {}, '<img alt="kW: --">'), 

128 ("/libraries/spaces", "get_library_info", {"info_type": "space"}, "Private invalid JSON"), 

129 ("/libraries/spaces", "get_library_info", {"info_type": "space"}, b'"\xff"'), 

130 ("/libraries/spaces", "get_library_info", {"info_type": "space"}, "[]"), 

131 ("/libraries/spaces", "get_library_info", {"info_type": "space"}, "null"), 

132 ("/libraries/spaces", "get_library_info", {"info_type": "space"}, "{}"), 

133 ( 

134 "/libraries/spaces", 

135 "get_library_info", 

136 {"info_type": "space"}, 

137 json.dumps({"resmsg": "\u6210\u529f"}), 

138 ), 

139 ( 

140 "/libraries/spaces", 

141 "get_library_info", 

142 {"info_type": "space"}, 

143 json.dumps({"resmsg": "\u6210\u529f", "rows": None}), 

144 ), 

145 ( 

146 "/libraries/spaces", 

147 "get_library_info", 

148 {"info_type": "space"}, 

149 json.dumps({"resmsg": "\u6210\u529f", "rows": [{}]}), 

150 ), 

151 ( 

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

153 "get_library_info", 

154 {"info_type": "lost_and_found"}, 

155 "<html>Private upstream login page</html>", 

156 ), 

157 ( 

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

159 "get_library_info", 

160 {"info_type": "lost_and_found"}, 

161 "<table></table>", 

162 ), 

163 ( 

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

165 "get_library_info", 

166 {"info_type": "lost_and_found"}, 

167 "<table><tr><td>Unexpected column</td></tr></table>", 

168 ), 

169 ( 

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

171 "get_library_info", 

172 {"info_type": "lost_and_found"}, 

173 "<p>\u76ee\u524d\u7121\u8cc7\u6599 !!</p>", 

174 ), 

175 ], 

176) 

177async def test_invalid_live_responses(rest, mock_upstream, path, tool, arguments, body): 

178 mock_upstream(lambda request: httpx.Response(200, content=body)) 

179 response = await rest.get(path) 

180 assert response.status_code == 502 

181 assert response.json() == {"detail": "Invalid response from upstream service"} 

182 async with Client(mcp) as client: 

183 result = await client.call_tool(tool, arguments, raise_on_error=False) 

184 assert result.is_error 

185 assert result.content[0].text == "Invalid response from upstream service" 

186 

187 

188@pytest.mark.parametrize("invalid_row_position", [0, 11]) 

189async def test_malformed_lost_rows_are_not_skipped( 

190 rest, mock_upstream, lost_items_html, invalid_row_position 

191): 

192 html = lost_items_html().replace( 

193 f"<td>{invalid_row_position}</td>", "<td>incomplete</td><td>extra cell</td>" 

194 ) 

195 mock_upstream(lambda request: httpx.Response(200, text=html)) 

196 response = await rest.get("/libraries/lost-and-found") 

197 assert response.status_code == 502 

198 assert response.json() == {"detail": "Invalid response from upstream service"} 

199 async with Client(mcp) as client: 

200 result = await client.call_tool( 

201 "get_library_info", {"info_type": "lost_and_found"}, raise_on_error=False 

202 ) 

203 assert result.is_error 

204 assert result.content[0].text == "Invalid response from upstream service" 

205 

206 

207@pytest.mark.parametrize( 

208 "headers", 

209 [ 

210 pytest.param((*LOST_ITEM_HEADERS, "Internal note"), id="extra"), 

211 pytest.param((*LOST_ITEM_HEADERS, LOST_ITEM_HEADERS[-1]), id="duplicate"), 

212 pytest.param(LOST_ITEM_HEADERS[:-1], id="missing"), 

213 ], 

214) 

215async def test_lost_items_reject_invalid_headers(rest, mock_upstream, headers): 

216 header = "".join(f"<th>{title}</th>" for title in headers) 

217 cells = "".join(f"<td>{index}</td>" for index in range(len(headers))) 

218 html = f"<table><tr>{header}</tr><tr>{cells}</tr></table>" 

219 mock_upstream(lambda request: httpx.Response(200, text=html)) 

220 

221 response = await rest.get("/libraries/lost-and-found") 

222 assert response.status_code == 502 

223 assert response.json() == {"detail": "Invalid response from upstream service"} 

224 async with Client(mcp) as client: 

225 result = await client.call_tool( 

226 "get_library_info", {"info_type": "lost_and_found"}, raise_on_error=False 

227 ) 

228 assert result.is_error 

229 assert result.content[0].text == "Invalid response from upstream service" 

230 

231 

232@pytest.mark.parametrize("count", [0, 12]) 

233async def test_lost_items_accept_reordered_headers(rest, mock_upstream, count): 

234 serial, found_at, location, description = LOST_ITEM_HEADERS 

235 headers = (description, location, serial, found_at) 

236 html = "<table><tr>" + "".join(f"<th> {title} </th>" for title in headers) + "</tr>" 

237 items = [] 

238 for index in range(count): 

239 values = { 

240 serial: str(index), 

241 found_at: "2026-09-30", 

242 location: "Main library", 

243 description: " Book\n title ", 

244 } 

245 items.append({**values, description: "Book title"}) 

246 html += "<tr>" + "".join(f"<td>{values[title]}</td>" for title in headers) + "</tr>" 

247 html += "</table>" 

248 mock_upstream(lambda request: httpx.Response(200, text=html)) 

249 

250 response = await rest.get("/libraries/lost-and-found") 

251 assert response.status_code == 200 

252 assert response.json() == items 

253 async with Client(mcp) as client: 

254 result = await client.call_tool("get_library_info", {"info_type": "lost_and_found"}) 

255 assert not result.is_error 

256 assert result.data == {"items": items[:10]} 

257 

258 

259@pytest.mark.parametrize( 

260 "html", 

261 [ 

262 pytest.param( 

263 "<h1>Lost and Found System</h1>\u76ee\u524d\u7121\u8cc7\u6599 !!", 

264 id="missing-content", 

265 ), 

266 pytest.param( 

267 '<div id="content">\u76ee\u524d\u7121\u8cc7\u6599 !!</div>', id="missing-heading" 

268 ), 

269 pytest.param( 

270 '<div id="content"><h1>Library News</h1>\u76ee\u524d\u7121\u8cc7\u6599 !!</div>', 

271 id="wrong-heading", 

272 ), 

273 pytest.param( 

274 '<div id="content"><h1>Lost and Found System</h1>Results pending</div>', 

275 id="missing-empty-marker", 

276 ), 

277 ], 

278) 

279async def test_lost_items_reject_invalid_empty_pages(rest, mock_upstream, html): 

280 mock_upstream(lambda request: httpx.Response(200, text=html)) 

281 

282 response = await rest.get("/libraries/lost-and-found") 

283 assert response.status_code == 502 

284 assert response.json() == {"detail": "Invalid response from upstream service"} 

285 async with Client(mcp) as client: 

286 result = await client.call_tool( 

287 "get_library_info", {"info_type": "lost_and_found"}, raise_on_error=False 

288 ) 

289 assert result.is_error 

290 assert result.content[0].text == "Invalid response from upstream service" 

291 

292 

293async def test_library_space_application_failure(rest, mock_upstream): 

294 mock_upstream( 

295 lambda request: httpx.Response( 

296 200, json={"resmsg": "Private upstream rejection", "rows": []} 

297 ) 

298 ) 

299 response = await rest.get("/libraries/spaces") 

300 assert response.status_code == 502 

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

302 async with Client(mcp) as client: 

303 result = await client.call_tool( 

304 "get_library_info", {"info_type": "space"}, raise_on_error=False 

305 ) 

306 assert result.is_error 

307 assert result.content[0].text == "Upstream service unavailable" 

308 

309 

310@pytest.mark.parametrize(CALL_PARAMETERS, LIVE_CALLS) 

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

312async def test_live_protocol_success( 

313 rest, 

314 mock_upstream, 

315 library_space_payload, 

316 lost_items_html, 

317 lost_items_empty_page, 

318 path, 

319 tool, 

320 arguments, 

321 service, 

322 method, 

323 empty_page, 

324): 

325 def handler(request): 

326 if request.url.host == "140.114.188.86": 

327 return httpx.Response(200, text='<img alt="kW: 0">') 

328 if request.url.host == "libsms.lib.nthu.edu.tw": 

329 return httpx.Response(200, json={**library_space_payload, "rows": []}) 

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

331 return httpx.Response(200, text=html) 

332 

333 mock_upstream(handler) 

334 response = await rest.get(path) 

335 assert response.status_code == 200 

336 async with Client(mcp) as client: 

337 result = await client.call_tool(tool, arguments) 

338 assert not result.is_error 

339 if tool == "get_energy_usage": 

340 assert len(result.data["zones"]) == len(response.json()) == 3 

341 assert all(zone["usage_kw"] == 0 for zone in result.data["zones"]) 

342 else: 

343 assert response.json() == [] 

344 assert result.data == {"spaces" if arguments["info_type"] == "space" else "items": []} 

345 

346 

347@pytest.mark.parametrize(CALL_PARAMETERS, LIVE_CALLS) 

348async def test_mcp_http_failure_is_tool_error( 

349 rest, mock_upstream, path, tool, arguments, service, method 

350): 

351 def handler(request): 

352 raise httpx.ReadTimeout("Private upstream exception", request=request) 

353 

354 mock_upstream(handler) 

355 async with mcp_app.lifespan(app): 

356 response = await rest.post( 

357 "/mcp", 

358 headers={"Accept": "application/json, text/event-stream"}, 

359 json={ 

360 "jsonrpc": "2.0", 

361 "id": 1, 

362 "method": "tools/call", 

363 "params": {"name": tool, "arguments": arguments}, 

364 }, 

365 ) 

366 assert response.status_code == 200 

367 messages = [ 

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

369 for line in response.text.splitlines() 

370 if line.startswith("data: ") 

371 ] 

372 assert len(messages) == 1 

373 assert "error" not in messages[0] 

374 result = messages[0]["result"] 

375 assert result["isError"] is True 

376 assert result["content"][0]["text"] == "Upstream request timed out"