Coverage for tests/test_api_client.py: 100%

1from datetime import timedelta 7 ctx1abcdefg

2from unittest.mock import patch 7 ctx1abcdefg

3 

4import httpx 7 ctx1abcdefg

5import pytest 7 ctx1abcdefg

6import respx 7 ctx1abcdefg

7from agent_detector import DetectionResult 7 ctx1abcdefg

8from httpx import Response 7 ctx1abcdefg

9 

10from fastapi_cloud_cli import __version__ 7 ctx1abcdefg

11from fastapi_cloud_cli.api import ( 7 ctx1abcdefg

12 STREAM_LOGS_MAX_RETRIES, 

13 APIClient, 

14 BuildLogLineMessage, 

15 DeploymentStatus, 

16 StreamLogError, 

17 TooManyRetriesError, 

18 get_http_error_code, 

19 get_http_error_hint, 

20) 

21from tests.utils import build_logs_response 7 ctx1abcdefg

22 

23 

24@pytest.fixture 7 ctx1abcdefg

25def client() -> httpx.Client: 7 ctx1abcdefg

26 """Create an HTTP client for testing.""" 

27 return APIClient() 7 ctx1abcdefg

28 

29 

30@pytest.fixture 7 ctx1abcdefg

31def deployment_id() -> str: 7 ctx1abcdefg

32 return "test-deployment-123" 7 ctx1abcdefg

33 

34 

35def test_api_client_reports_high_confidence_agent() -> None: 7 ctx1abcdefg

36 detection = DetectionResult("codex", "high", "environment", "CODEX_THREAD_ID") 7 ctx2qbzbAbBbrbCbDb

37 

38 with ( 2 ctx2qbrb

39 patch( 

40 "fastapi_cloud_cli.api.client.detect_agent", return_value=detection 

41 ) as mock_detect_agent, 

42 APIClient() as client, 

43 ): 

44 assert client.headers["User-Agent"] == ( 7 ctx2qbzbAbBbrbCbDb

45 f"fastapi-cloud-cli/{__version__} AI-Agent/codex" 

46 ) 

47 

48 mock_detect_agent.assert_called_once_with(minimum_confidence="high") 7 ctx2qbzbAbBbrbCbDb

49 

50 

51def test_api_client_omits_unattributed_agent() -> None: 7 ctx1abcdefg

52 with ( 2 ctx26b7b

53 patch("fastapi_cloud_cli.api.client.detect_agent", return_value=None), 

54 APIClient() as client, 

55 ): 

56 assert client.headers["User-Agent"] == f"fastapi-cloud-cli/{__version__}" 7 ctx26b!b#b$b7b%b'b

57 

58 

59@pytest.fixture 7 ctx1abcdefg

60def logs_route(respx_mock: respx.MockRouter, deployment_id: str) -> respx.Route: 7 ctx1abcdefg

61 return respx_mock.get(f"/deployments/{deployment_id}/build-logs") 7 ctx1abcdefg

62 

63 

64def test_stream_build_logs_successful( 7 ctx1abcdefg

65 logs_route: respx.Route, 

66 client: APIClient, 

67 deployment_id: str, 

68) -> None: 

69 logs_route.mock( 7 ctx1vwxyzAB

70 return_value=Response( 

71 200, 

72 content=build_logs_response( 

73 {"type": "message", "message": "Building...", "id": "1"}, 

74 {"type": "message", "message": "Done!", "id": "2"}, 

75 {"type": "complete", "id": "3"}, 

76 ), 

77 ) 

78 ) 

79 

80 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1vwxyzAB

81 

82 assert len(logs) == 3 7 ctx1vwxyzAB

83 

84 assert logs[0].type == "message" 7 ctx1vwxyzAB

85 assert logs[0].message == "Building..." 7 ctx1vwxyzAB

86 

87 assert logs[1].type == "message" 7 ctx1vwxyzAB

88 assert logs[1].message == "Done!" 7 ctx1vwxyzAB

89 

90 assert logs[2].type == "complete" 7 ctx1vwxyzAB

91 

92 

93def test_get_http_error_code_returns_network_error() -> None: 7 ctx1abcdefg

94 assert ( 7 ctx2(b)b*b+b,b-b.b

95 get_http_error_code(httpx.NetworkError("Connection failed")) == "network_error" 

96 ) 

97 

98 

99def test_get_http_error_hint_for_invalid_deploy_token() -> None: 7 ctx1abcdefg

100 assert get_http_error_hint("invalid_token", auth_mode="token") == ( 7 ctx2/b:b;b=b?b@b[b

101 "Make sure FASTAPI_CLOUD_TOKEN contains a valid token." 

102 ) 

103 

104 

105def test_stream_build_logs_failed( 7 ctx1abcdefg

106 logs_route: respx.Route, client: APIClient, deployment_id: str 

107) -> None: 

108 logs_route.mock( 7 ctx1456789!

109 return_value=Response( 

110 200, 

111 content=build_logs_response( 

112 {"type": "message", "message": "Error occurred", "id": "1"}, 

113 {"type": "failed", "id": "2"}, 

114 ), 

115 ) 

116 ) 

117 

118 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1456789!

119 

120 assert len(logs) == 2 7 ctx1456789!

121 assert logs[0].type == "message" 7 ctx1456789!

122 assert logs[1].type == "failed" 7 ctx1456789!

123 

124 

125@pytest.mark.parametrize("terminal_type", ["complete", "failed"]) 7 ctx1abcdefg

126def test_stream_build_logs_stop_after_terminal_state( 7 ctx1abcdefg

127 logs_route: respx.Route, 

128 client: APIClient, 

129 terminal_type: str, 

130 deployment_id: str, 

131) -> None: 

132 logs_route.mock( 7 ctx1#$%'()*

133 return_value=Response( 

134 200, 

135 content=build_logs_response( 

136 {"type": "message", "message": "Step 1", "id": "1"}, 

137 {"type": terminal_type, "id": "2"}, 

138 {"type": "message", "message": "This should not appear", "id": "3"}, 

139 ), 

140 ) 

141 ) 

142 

143 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1#$%'()*

144 

145 assert len(logs) == 2 7 ctx1#$%'()*

146 assert logs[0].type == "message" 7 ctx1#$%'()*

147 assert logs[1].type == terminal_type 7 ctx1#$%'()*

148 

149 

150def test_stream_build_logs_internal_messages_are_skipped( 7 ctx1abcdefg

151 logs_route: respx.Route, 

152 client: APIClient, 

153 deployment_id: str, 

154) -> None: 

155 logs_route.mock( 7 ctx1+,-./:;

156 return_value=Response( 

157 200, 

158 content=build_logs_response( 

159 {"type": "heartbeat", "id": "1"}, 

160 {"type": "message", "message": "Continuing...", "id": "2"}, 

161 {"type": "complete", "id": "3"}, 

162 ), 

163 ) 

164 ) 

165 

166 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1+,-./:;

167 

168 assert len(logs) == 2 7 ctx1+,-./:;

169 assert logs[0].type == "message" 7 ctx1+,-./:;

170 assert logs[1].type == "complete" 7 ctx1+,-./:;

171 

172 

173def test_stream_build_logs_malformed_json_is_skipped( 7 ctx1abcdefg

174 logs_route: respx.Route, client: APIClient, deployment_id: str 

175) -> None: 

176 content = "\n".join( 7 ctx1EFGHIJK

177 [ 

178 '{"type": "message", "message": "Valid", "id": "1"}', 

179 "not valid json", 

180 '{"type": "complete", "id": "2"}', 

181 ] 

182 ) 

183 

184 logs_route.mock(return_value=Response(200, content=content)) 7 ctx1EFGHIJK

185 

186 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1EFGHIJK

187 

188 assert len(logs) == 2 7 ctx1EFGHIJK

189 assert logs[0].type == "message" 7 ctx1EFGHIJK

190 assert logs[1].type == "complete" 7 ctx1EFGHIJK

191 

192 

193def test_stream_build_logs_unknown_log_type_is_skipped( 7 ctx1abcdefg

194 logs_route: respx.Route, client: APIClient, deployment_id: str 

195) -> None: 

196 logs_route.mock( 7 ctx1=?@[]^_

197 return_value=Response( 

198 200, 

199 content=build_logs_response( 

200 {"type": "unknown_future_type", "id": "1"}, 

201 {"type": "message", "message": "Valid", "id": "2"}, 

202 {"type": "complete", "id": "3"}, 

203 ), 

204 ) 

205 ) 

206 

207 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1=?@[]^_

208 

209 # Unknown type should be filtered out 

210 assert len(logs) == 2 7 ctx1=?@[]^_

211 assert logs[0].type == "message" 7 ctx1=?@[]^_

212 assert logs[1].type == "complete" 7 ctx1=?@[]^_

213 

214 

215@pytest.mark.parametrize( 7 ctx1abcdefg

216 "network_error", 

217 [httpx.NetworkError, httpx.TimeoutException, httpx.RemoteProtocolError], 

218) 

219def test_stream_build_logs_network_error_retry( 7 ctx1abcdefg

220 logs_route: respx.Route, 

221 client: APIClient, 

222 network_error: Exception, 

223 deployment_id: str, 

224) -> None: 

225 # First call fails, second succeeds 

226 logs_route.side_effect = [ 7 ctx1LMNOPQR

227 network_error, 

228 network_error, 

229 Response( 

230 200, 

231 content=build_logs_response( 

232 {"type": "message", "message": "Success after retry", "id": "1"}, 

233 {"type": "complete", "id": "2"}, 

234 ), 

235 ), 

236 ] 

237 

238 with patch("time.sleep"): 7 ctx1LMNOPQR

239 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1LMNOPQR

240 

241 assert len(logs) == 2 7 ctx1LMNOPQR

242 assert logs[0].type == "message" 7 ctx1LMNOPQR

243 assert logs[0].message == "Success after retry" 7 ctx1LMNOPQR

244 

245 

246def test_stream_build_logs_server_error_retry( 7 ctx1abcdefg

247 logs_route: respx.Route, client: APIClient, deployment_id: str 

248) -> None: 

249 logs_route.side_effect = [ 7 ctx2` { | } ~ abbb

250 Response(500, text="Internal Server Error"), 

251 Response( 

252 200, 

253 content=build_logs_response( 

254 {"type": "complete", "id": "1"}, 

255 ), 

256 ), 

257 ] 

258 

259 with patch("time.sleep"): 7 ctx2` { | } ~ abbb

260 logs = list(client.stream_build_logs(deployment_id)) 7 ctx2` { | } ~ abbb

261 

262 assert len(logs) == 1 7 ctx2` { | } ~ abbb

263 assert logs[0].type == "complete" 7 ctx2` { | } ~ abbb

264 

265 

266def test_stream_build_logs_client_error_raises_immediately( 7 ctx1abcdefg

267 logs_route: respx.Route, client: APIClient, deployment_id: str 

268) -> None: 

269 logs_route.mock(return_value=Response(404, text="Not Found")) 7 ctx2EbFbGbHbIbJbKb

270 

271 with pytest.raises(StreamLogError, match="HTTP 404"): 7 ctx2EbFbGbHbIbJbKb

272 list(client.stream_build_logs(deployment_id)) 7 ctx2EbFbGbHbIbJbKb

273 

274 

275def test_stream_build_logs_max_retries_exceeded( 7 ctx1abcdefg

276 logs_route: respx.Route, client: APIClient, deployment_id: str 

277) -> None: 

278 logs_route.side_effect = httpx.NetworkError("Connection failed") 7 ctx2sbtbubvbwbxbyb

279 

280 with patch("time.sleep"): 7 ctx2sbtbubvbwbxbyb

281 with pytest.raises( 7 ctx2sbtbubvbwbxbyb

282 TooManyRetriesError, 

283 match=f"Failed after {STREAM_LOGS_MAX_RETRIES} attempts", 

284 ): 

285 list(client.stream_build_logs(deployment_id)) 7 ctx2sbtbubvbwbxbyb

286 

287 

288def test_stream_build_logs_empty_lines_are_skipped( 7 ctx1abcdefg

289 logs_route: respx.Route, client: APIClient, deployment_id: str 

290) -> None: 

291 content = "\n".join( 7 ctx1STUVWXY

292 [ 

293 "", 

294 '{"type": "message", "message": "Valid", "id": "1"}', 

295 " ", 

296 '{"type": "complete", "id": "2"}', 

297 "", 

298 ] 

299 ) 

300 

301 logs_route.mock(return_value=Response(200, content=content)) 7 ctx1STUVWXY

302 

303 logs = list(client.stream_build_logs(deployment_id)) 7 ctx1STUVWXY

304 

305 assert len(logs) == 2 7 ctx1STUVWXY

306 assert logs[0].type == "message" 7 ctx1STUVWXY

307 assert logs[1].type == "complete" 7 ctx1STUVWXY

308 

309 

310def test_stream_build_logs_continue_after_timeout( 7 ctx1abcdefg

311 respx_mock: respx.MockRouter, 

312 client: APIClient, 

313 deployment_id: str, 

314) -> None: 

315 for id, last_id in enumerate([None, "1", "2"], start=1): 7 ctx1hijklmn

316 params = {"last_id": last_id} if last_id else {} 7 ctx1hijklmn

317 message = f"message {id}" 7 ctx1hijklmn

318 

319 respx_mock.get( 7 ctx1hijklmn

320 f"/deployments/{deployment_id}/build-logs", params__eq=params 

321 ).mock( 

322 return_value=Response( 

323 200, 

324 content=build_logs_response( 

325 {"type": "message", "message": message, "id": str(id)}, 

326 {"type": "timeout"}, 

327 ), 

328 ) 

329 ) 

330 

331 respx_mock.get( 7 ctx1hijklmn

332 f"/deployments/{deployment_id}/build-logs", params__eq={"last_id": "3"} 

333 ).mock( 

334 return_value=Response( 

335 200, 

336 content=build_logs_response( 

337 {"type": "message", "message": "message 4", "id": "4"}, 

338 {"type": "complete", "id": "5"}, 

339 ), 

340 ) 

341 ) 

342 

343 logs = client.stream_build_logs(deployment_id) 7 ctx1hijklmn

344 

345 with patch("time.sleep"): 7 ctx1hijklmn

346 assert next(logs) == BuildLogLineMessage(message="message 1", id="1") 7 ctx1hijklmn

347 assert next(logs) == BuildLogLineMessage(message="message 2", id="2") 7 ctx1hijklmn

348 assert next(logs) == BuildLogLineMessage(message="message 3", id="3") 7 ctx1hijklmn

349 assert next(logs) == BuildLogLineMessage(message="message 4", id="4") 7 ctx1hijklmn

350 assert next(logs).type == "complete" 7 ctx1hijklmn

351 

352 

353def test_stream_build_logs_no_follow_stops_after_timeout( 7 ctx1abcdefg

354 logs_route: respx.Route, client: APIClient, deployment_id: str 

355) -> None: 

356 logs_route.mock( 7 ctx2LbMbNbObPbQbRb

357 return_value=Response( 

358 200, 

359 content=build_logs_response( 

360 {"type": "message", "message": "Step 1", "id": "1"}, 

361 {"type": "timeout"}, 

362 ), 

363 ) 

364 ) 

365 

366 logs = list(client.stream_build_logs(deployment_id, follow=False)) 7 ctx2LbMbNbObPbQbRb

367 

368 assert logs == [BuildLogLineMessage(message="Step 1", id="1")] 7 ctx2LbMbNbObPbQbRb

369 

370 

371def test_stream_build_logs_connection_closed_without_complete_failed_or_timeout( 7 ctx1abcdefg

372 logs_route: respx.Route, client: APIClient, deployment_id: str 

373) -> None: 

374 logs_route.mock( 7 ctx2cbdbebfbgbhbib

375 return_value=Response( 

376 200, 

377 content=build_logs_response( 

378 {"type": "message", "message": "hello", "id": "1"}, 

379 ), 

380 ) 

381 ) 

382 

383 logs = client.stream_build_logs(deployment_id) 7 ctx2cbdbebfbgbhbib

384 

385 with patch("time.sleep"), pytest.raises(TooManyRetriesError, match="Failed after"): 7 ctx2cbdbebfbgbhbib

386 for _ in range(STREAM_LOGS_MAX_RETRIES + 1): 7 ctx2cbdbebfbgbhbib

387 next(logs) 7 ctx2cbdbebfbgbhbib

388 

389 

390def test_stream_build_logs_retry_timeout( 7 ctx1abcdefg

391 logs_route: respx.Route, 

392 client: APIClient, 

393 deployment_id: str, 

394) -> None: 

395 clock = [0.0] # Container with a value to use as a result of time.monotonic() mock 7 ctx1CZ01D23

396 

397 def responses(request: httpx.Request, route: respx.Route) -> Response: 7 ctx1CZ01D23

398 clock[0] += timedelta(hours=1).total_seconds() # Simulate time passing 7 ctx1CZ01D23

399 

400 return Response( 7 ctx1CZ01D23

401 200, 

402 content=build_logs_response( 

403 {"type": "message", "message": "First", "id": "1"}, 

404 ), 

405 ) 

406 

407 logs_route.mock(side_effect=responses) 7 ctx1CZ01D23

408 

409 with ( 2 ctx1CD

410 patch("time.monotonic", side_effect=lambda: clock[0]), 

411 patch("time.sleep"), 

412 pytest.raises(TimeoutError, match="timed out"), 

413 ): 

414 list(client.stream_build_logs(deployment_id)) 7 ctx1CZ01D23

415 

416 

417@pytest.fixture 7 ctx1abcdefg

418def poll_route(respx_mock: respx.MockRouter, deployment_id: str) -> respx.Route: 7 ctx1abcdefg

419 return respx_mock.get(f"/deployments/{deployment_id}") 7 ctx1abcdefg

420 

421 

422def test_poll_deployment_status_recovers_from_transient_errors( 7 ctx1abcdefg

423 poll_route: respx.Route, client: APIClient, deployment_id: str 

424) -> None: 

425 call_count = 0 7 ctx1opqrstu

426 

427 def handler(request: httpx.Request, route: respx.Route) -> Response: 7 ctx1opqrstu

428 nonlocal call_count 

429 call_count += 1 7 ctx1opqrstu

430 if call_count <= 2: 7 ctx1opqrstu

431 return Response(500) 7 ctx1opqrstu

432 return Response(200, json={"status": "success"}) 7 ctx1opqrstu

433 

434 poll_route.mock(side_effect=handler) 7 ctx1opqrstu

435 

436 with patch("time.sleep"): 7 ctx1opqrstu

437 status = client.poll_deployment_status(deployment_id) 7 ctx1opqrstu

438 

439 assert status == DeploymentStatus.success 7 ctx1opqrstu

440 assert call_count == 3 7 ctx1opqrstu

441 

442 

443def test_poll_deployment_status_keeps_polling_on_unknown_status( 7 ctx1abcdefg

444 poll_route: respx.Route, client: APIClient, deployment_id: str 

445) -> None: 

446 poll_route.mock( 7 ctx2jbkblbmbnbobpb

447 side_effect=[ 

448 Response(200, json={"status": "some_new_status"}), 

449 Response(200, json={"status": "success"}), 

450 ] 

451 ) 

452 

453 with patch("time.sleep"): 7 ctx2jbkblbmbnbobpb

454 status = client.poll_deployment_status(deployment_id) 7 ctx2jbkblbmbnbobpb

455 

456 assert status == DeploymentStatus.success 7 ctx2jbkblbmbnbobpb

457 assert poll_route.call_count == 2 7 ctx2jbkblbmbnbobpb

458 

459 

460def test_poll_deployment_status_raises_after_max_consecutive_errors( 7 ctx1abcdefg

461 poll_route: respx.Route, client: APIClient, deployment_id: str 

462) -> None: 

463 poll_route.mock(return_value=Response(500)) 7 ctx2SbTbUbVbWbXbYb

464 

465 with patch("time.sleep"), pytest.raises(TooManyRetriesError): 7 ctx2SbTbUbVbWbXbYb

466 client.poll_deployment_status(deployment_id) 7 ctx2SbTbUbVbWbXbYb

467 

468 

469def test_poll_deployment_status_timeout(client: APIClient, deployment_id: str) -> None: 7 ctx1abcdefg

470 with ( 2 ctx28b9b

471 patch("fastapi_cloud_cli.api.client.POLL_TIMEOUT", timedelta(seconds=-1)), 

472 pytest.raises(TimeoutError, match="timed out"), 

473 ): 

474 client.poll_deployment_status(deployment_id) 7 ctx28b]b^b_b9b`b{b

475 

476 

477def test_get_deployment_with_persisted_failure( 7 ctx1abcdefg

478 respx_mock: respx.MockRouter, 

479 client: APIClient, 

480 deployment_id: str, 

481) -> None: 

482 deployment = { 7 ctx2Zb0b1b2b3b4b5b

483 "id": deployment_id, 

484 "app_id": "123", 

485 "slug": "demo-build", 

486 "created_at": "2026-09-10T12:00:00Z", 

487 "status": "building_image_failed", 

488 "url": "https://demo.fastapicloud.app", 

489 "dashboard_url": "https://dashboard.fastapicloud.com/demo-build", 

490 "failure": { 

491 "error_code": "future_build_error", 

492 "error_title": "A new build error", 

493 "error_message": "Guidance supplied by the backend.", 

494 "error_hint": "", 

495 }, 

496 } 

497 respx_mock.get(f"/deployments/{deployment_id}").respond(200, json=deployment) 7 ctx2Zb0b1b2b3b4b5b

498 

499 assert client.get_deployment(deployment_id).model_dump(mode="json") == deployment 7 ctx2Zb0b1b2b3b4b5b