Coverage for tests/test_logs.py: 100%
216 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-11-09 00:25 +0000
« prev ^ index » next coverage.py v7.16.0, created at 2026-11-09 00:25 +0000
1import json 1abcdefg
2from pathlib import Path 1abcdefg
3from unittest.mock import patch 1abcdefg
5import httpx 1abcdefg
6import pytest 1abcdefg
7import respx 1abcdefg
8from typer.testing import CliRunner 1abcdefg
10from fastapi_cloud_cli.api import TooManyRetriesError 1abcdefg
11from fastapi_cloud_cli.cli import cloud_app as app 1abcdefg
12from tests.conftest import ConfiguredApp 1abcdefg
13from tests.utils import changing_dir 1abcdefg
15runner = CliRunner() 1abcdefg
18def test_shows_message_if_not_logged_in(logged_out_cli: None) -> None: 1abcdefg
19 result = runner.invoke(app, ["logs"]) 2/b:b;b=b?b@b[b
21 assert result.exit_code == 1 2/b:b;b=b?b@b[b
22 assert "No credentials found" in result.output 2/b:b;b=b?b@b[b
25def test_shows_message_if_app_not_configured(logged_in_cli: None) -> None: 1abcdefg
26 result = runner.invoke(app, ["logs"]) 2]b^b_b`b{b|b}b
28 assert result.exit_code == 1 2]b^b_b`b{b|b}b
29 assert "App ID is required." in result.output 2]b^b_b`b{b|b}b
32@pytest.mark.respx 1abcdefg
33def test_displays_logs( 1abcdefg
34 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
35) -> None:
36 log_lines = [ 1hijklmn
37 json.dumps(
38 {
39 "timestamp": "2025-12-05T14:32:01.123000Z",
40 "message": "Application startup complete",
41 "level": "info",
42 }
43 ),
44 json.dumps(
45 {
46 "timestamp": "2025-12-05T14:32:05.456000Z",
47 "message": "GET /health 200",
48 "level": "info",
49 }
50 ),
51 ]
52 response_content = "\n".join(log_lines) 1hijklmn
54 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1hijklmn
55 return_value=httpx.Response(200, content=response_content)
56 )
58 with changing_dir(configured_app.path): 1hijklmn
59 result = runner.invoke(app, ["logs", "--no-follow"]) 1hijklmn
61 assert result.exit_code == 0 1hijklmn
62 assert "Fetching logs" in result.output 1hijklmn
63 assert configured_app.app_id in result.output 1hijklmn
64 assert "Application startup complete" in result.output 1hijklmn
65 assert "GET /health 200" in result.output 1hijklmn
68@pytest.mark.respx 1abcdefg
69def test_displays_logs_with_explicit_app_id( 1abcdefg
70 logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
71) -> None:
72 app_id = "explicit-app" 1vwxyzAB
73 response_content = json.dumps( 1vwxyzAB
74 {
75 "timestamp": "2025-12-05T14:32:01.123000Z",
76 "message": "Application startup complete",
77 "level": "info",
78 }
79 )
81 respx_mock.get(url__regex=rf"/apps/{app_id}/logs/stream.*").mock( 1vwxyzAB
82 return_value=httpx.Response(200, content=response_content)
83 )
85 with changing_dir(tmp_path): 1vwxyzAB
86 result = runner.invoke(app, ["logs", "--no-follow", "--app-id", app_id]) 1vwxyzAB
88 assert result.exit_code == 0 1vwxyzAB
89 assert "Fetching logs" in result.output 1vwxyzAB
90 assert app_id in result.output 1vwxyzAB
91 assert "Application startup complete" in result.output 1vwxyzAB
94@pytest.mark.respx 1abcdefg
95def test_apps_logs_displays_logs( 1abcdefg
96 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
97) -> None:
98 response_content = json.dumps( 1CDEFGHI
99 {
100 "timestamp": "2025-12-05T14:32:01.123000Z",
101 "message": "Application startup complete",
102 "level": "info",
103 }
104 )
106 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1CDEFGHI
107 return_value=httpx.Response(200, content=response_content)
108 )
110 with changing_dir(configured_app.path): 1CDEFGHI
111 result = runner.invoke(app, ["apps", "logs", "--no-follow"]) 1CDEFGHI
113 assert result.exit_code == 0 1CDEFGHI
114 assert "Fetching logs" in result.output 1CDEFGHI
115 assert configured_app.app_id in result.output 1CDEFGHI
116 assert "Application startup complete" in result.output 1CDEFGHI
119@pytest.mark.respx 1abcdefg
120def test_apps_logs_no_follow_json_outputs_envelope( 1abcdefg
121 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
122) -> None:
123 logs = [ 1JKLMNOP
124 {
125 "timestamp": "2025-12-05T14:32:01.123000Z",
126 "message": "Application startup complete",
127 "level": "info",
128 },
129 {
130 "timestamp": "2025-12-05T14:32:05.456000Z",
131 "message": "GET /health 200",
132 "level": "info",
133 },
134 ]
135 response_content = "\n".join(json.dumps(log) for log in logs) 1JKLMNOP
137 respx_mock.get( 1JKLMNOP
138 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
139 params={"follow": "false"},
140 ).mock(return_value=httpx.Response(200, content=response_content))
142 with changing_dir(configured_app.path): 1JKLMNOP
143 result = runner.invoke(app, ["apps", "logs", "--no-follow", "--json"]) 1JKLMNOP
145 assert result.exit_code == 0 1JKLMNOP
146 assert json.loads(result.stdout) == { 1JKLMNOP
147 "data": {
148 "app_id": configured_app.app_id,
149 "logs": logs,
150 }
151 }
152 assert result.stderr == "" 1JKLMNOP
155@pytest.mark.respx 1abcdefg
156def test_apps_logs_follow_json_outputs_ndjson( 1abcdefg
157 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
158) -> None:
159 logs = [ 1QRSTUVW
160 {
161 "timestamp": "2025-12-05T14:32:01.123000Z",
162 "message": "Application startup complete",
163 "level": "info",
164 },
165 {
166 "timestamp": "2025-12-05T14:32:05.456000Z",
167 "message": "GET /health 200",
168 "level": "info",
169 },
170 ]
171 response_content = "\n".join(json.dumps(log) for log in logs) 1QRSTUVW
173 respx_mock.get( 1QRSTUVW
174 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
175 params={"follow": "true"},
176 ).mock(return_value=httpx.Response(200, content=response_content))
178 with changing_dir(configured_app.path): 1QRSTUVW
179 result = runner.invoke(app, ["apps", "logs", "--json"]) 1QRSTUVW
181 assert result.exit_code == 0 1QRSTUVW
182 assert [json.loads(line) for line in result.stdout.splitlines()] == [ 1QRSTUVW
183 {"type": "log", "app_id": configured_app.app_id, **log} for log in logs
184 ]
185 assert result.stderr == "" 1QRSTUVW
188@pytest.mark.respx 1abcdefg
189def test_passes_default_params( 1abcdefg
190 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
191) -> None:
192 respx_mock.get( 2cbdbebfbgbhbib
193 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
194 params={"follow": "true", "tail": "100", "since": "5m"},
195 ).mock(return_value=httpx.Response(200, content=""))
197 with changing_dir(configured_app.path): 2cbdbebfbgbhbib
198 result = runner.invoke(app, ["logs"]) 2cbdbebfbgbhbib
200 assert result.exit_code == 0 2cbdbebfbgbhbib
201 assert "Streaming logs" in result.output 2cbdbebfbgbhbib
202 assert configured_app.app_id in result.output 2cbdbebfbgbhbib
205@pytest.mark.respx 1abcdefg
206def test_streams_logs_in_human_output( 1abcdefg
207 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
208) -> None:
209 response_content = json.dumps( 1#$%'()*
210 {
211 "timestamp": "2025-12-05T14:32:01.123000Z",
212 "message": "Application startup complete",
213 "level": "info",
214 }
215 )
217 respx_mock.get( 1#$%'()*
218 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
219 params={"follow": "true"},
220 ).mock(return_value=httpx.Response(200, content=response_content))
222 with changing_dir(configured_app.path): 1#$%'()*
223 result = runner.invoke(app, ["logs"]) 1#$%'()*
225 assert result.exit_code == 0 1#$%'()*
226 assert "Streaming logs" in result.output 1#$%'()*
227 assert "Application startup complete" in result.output 1#$%'()*
230@pytest.mark.respx 1abcdefg
231def test_passes_custom_params( 1abcdefg
232 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
233) -> None:
234 respx_mock.get( 2SbTbUbVbWbXbYb
235 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
236 params={"tail": "50", "since": "1h", "follow": "false"},
237 ).mock(return_value=httpx.Response(200, content=""))
239 with changing_dir(configured_app.path): 2SbTbUbVbWbXbYb
240 result = runner.invoke( 2SbTbUbVbWbXbYb
241 app, ["logs", "--no-follow", "--tail", "50", "--since", "1h"]
242 )
244 assert result.exit_code == 0 2SbTbUbVbWbXbYb
247@pytest.mark.respx 1abcdefg
248def test_displays_all_log_levels( 1abcdefg
249 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
250) -> None:
251 log_lines = [ 1opqrstu
252 json.dumps(
253 {
254 "timestamp": "2025-12-05T14:32:01.123000Z",
255 "message": "Debug message",
256 "level": "debug",
257 }
258 ),
259 json.dumps(
260 {
261 "timestamp": "2025-12-05T14:32:02.123000Z",
262 "message": "Info message",
263 "level": "info",
264 }
265 ),
266 json.dumps(
267 {
268 "timestamp": "2025-12-05T14:32:03.123000Z",
269 "message": "Warning message",
270 "level": "warning",
271 }
272 ),
273 json.dumps(
274 {
275 "timestamp": "2025-12-05T14:32:04.123000Z",
276 "message": "Error message",
277 "level": "error",
278 }
279 ),
280 ]
281 response_content = "\n".join(log_lines) 1opqrstu
283 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1opqrstu
284 return_value=httpx.Response(200, content=response_content)
285 )
287 with changing_dir(configured_app.path): 1opqrstu
288 result = runner.invoke(app, ["logs", "--no-follow"]) 1opqrstu
290 assert result.exit_code == 0 1opqrstu
291 assert "Debug message" in result.output 1opqrstu
292 assert "Info message" in result.output 1opqrstu
293 assert "Warning message" in result.output 1opqrstu
294 assert "Error message" in result.output 1opqrstu
297@pytest.mark.respx 1abcdefg
298def test_handles_401_unauthorized( 1abcdefg
299 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
300) -> None:
301 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2qbrbsbtbubvbwb
302 return_value=httpx.Response(401)
303 )
305 with changing_dir(configured_app.path): 2qbrbsbtbubvbwb
306 result = runner.invoke(app, ["logs", "--no-follow"]) 2qbrbsbtbubvbwb
308 assert result.exit_code == 1 2qbrbsbtbubvbwb
309 assert "token is not valid" in result.output 2qbrbsbtbubvbwb
312@pytest.mark.respx 1abcdefg
313def test_handles_403_forbidden( 1abcdefg
314 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
315) -> None:
316 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2xbybzbAbBbCbDb
317 return_value=httpx.Response(403)
318 )
320 with changing_dir(configured_app.path): 2xbybzbAbBbCbDb
321 result = runner.invoke(app, ["logs", "--no-follow"]) 2xbybzbAbBbCbDb
323 assert result.exit_code == 1 2xbybzbAbBbCbDb
324 assert "You don't have permissions for this resource" in result.output 2xbybzbAbBbCbDb
327@pytest.mark.respx 1abcdefg
328def test_handles_404( 1abcdefg
329 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
330) -> None:
331 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2EbFbGbHbIbJbKb
332 return_value=httpx.Response(404)
333 )
335 with changing_dir(configured_app.path): 2EbFbGbHbIbJbKb
336 result = runner.invoke(app, ["logs", "--no-follow"]) 2EbFbGbHbIbJbKb
338 assert result.exit_code == 1 2EbFbGbHbIbJbKb
339 assert "App not found" in result.output 2EbFbGbHbIbJbKb
342@pytest.mark.respx 1abcdefg
343def test_handles_retention_limit_error( 1abcdefg
344 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
345) -> None:
346 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2jbkblbmbnbobpb
347 return_value=httpx.Response(
348 400, json={"detail": "Cannot fetch logs older than 14 days"}
349 )
350 )
352 with changing_dir(configured_app.path): 2jbkblbmbnbobpb
353 result = runner.invoke(app, ["logs", "--no-follow", "--since", "30d"]) 2jbkblbmbnbobpb
355 assert result.exit_code == 1 2jbkblbmbnbobpb
356 assert "Cannot fetch logs older than 14 days" in result.output 2jbkblbmbnbobpb
357 assert "Try a shorter time range" in result.output 2jbkblbmbnbobpb
360@pytest.mark.respx 1abcdefg
361def test_handles_retention_limit_error_json( 1abcdefg
362 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
363) -> None:
364 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1XYZ0123
365 return_value=httpx.Response(
366 400, json={"detail": "Cannot fetch logs older than 14 days"}
367 )
368 )
370 with changing_dir(configured_app.path): 1XYZ0123
371 result = runner.invoke(app, ["logs", "--no-follow", "--since", "30d", "--json"]) 1XYZ0123
373 assert result.exit_code == 1 1XYZ0123
374 error = json.loads(result.stdout)["error"] 1XYZ0123
375 assert error["code"] == "invalid_input" 1XYZ0123
376 assert error["message"] == "Cannot fetch logs older than 14 days" 1XYZ0123
377 assert "Try a shorter time range" in error["hint"] 1XYZ0123
380@pytest.mark.respx 1abcdefg
381def test_shows_message_when_no_logs_found( 1abcdefg
382 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
383) -> None:
384 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2LbMbNbObPbQbRb
385 return_value=httpx.Response(200, content="")
386 )
388 with changing_dir(configured_app.path): 2LbMbNbObPbQbRb
389 result = runner.invoke(app, ["logs", "--no-follow"]) 2LbMbNbObPbQbRb
391 assert result.exit_code == 0 2LbMbNbObPbQbRb
392 assert "No logs found" in result.output 2LbMbNbObPbQbRb
395@pytest.mark.respx 1abcdefg
396def test_handles_server_error_message( 1abcdefg
397 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
398) -> None:
399 log_lines = [ 1+,-./:;
400 json.dumps({"type": "error", "message": "Log storage unavailable"}),
401 ]
402 response_content = "\n".join(log_lines) 1+,-./:;
404 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1+,-./:;
405 return_value=httpx.Response(200, content=response_content)
406 )
408 with changing_dir(configured_app.path): 1+,-./:;
409 result = runner.invoke(app, ["logs", "--no-follow"]) 1+,-./:;
411 assert result.exit_code == 1 1+,-./:;
412 assert "Log storage unavailable" in result.output 1+,-./:;
415@pytest.mark.respx 1abcdefg
416def test_handles_unknown_log_level( 1abcdefg
417 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
418) -> None:
419 log_lines = [ 1=?@[]^_
420 json.dumps(
421 {
422 "timestamp": "2025-12-05T14:32:01.123000Z",
423 "message": "Unknown level message",
424 "level": "custom_level",
425 }
426 ),
427 ]
428 response_content = "\n".join(log_lines) 1=?@[]^_
430 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1=?@[]^_
431 return_value=httpx.Response(200, content=response_content)
432 )
434 with changing_dir(configured_app.path): 1=?@[]^_
435 result = runner.invoke(app, ["logs", "--no-follow"]) 1=?@[]^_
437 assert result.exit_code == 0 1=?@[]^_
438 assert "Unknown level message" in result.output 1=?@[]^_
441@pytest.mark.respx 1abcdefg
442def test_skips_invalid_json_lines( 1abcdefg
443 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
444) -> None:
445 log_lines = [ 2` { | } ~ abbb
446 "not valid json",
447 json.dumps(
448 {
449 "timestamp": "2025-12-05T14:32:01.123000Z",
450 "message": "Valid log message",
451 "level": "info",
452 }
453 ),
454 ]
455 response_content = "\n".join(log_lines) 2` { | } ~ abbb
457 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 2` { | } ~ abbb
458 return_value=httpx.Response(200, content=response_content)
459 )
461 with changing_dir(configured_app.path): 2` { | } ~ abbb
462 result = runner.invoke(app, ["logs", "--no-follow"]) 2` { | } ~ abbb
464 assert result.exit_code == 0 2` { | } ~ abbb
465 assert "Valid log message" in result.output 2` { | } ~ abbb
468@pytest.mark.respx 1abcdefg
469def test_skips_heartbeat_messages( 1abcdefg
470 logged_in_cli: None, respx_mock: respx.MockRouter, configured_app: ConfiguredApp
471) -> None:
472 log_lines = [ 1456789!
473 json.dumps({"type": "heartbeat"}),
474 json.dumps(
475 {
476 "timestamp": "2025-12-05T14:32:01.123000Z",
477 "message": "Real log message",
478 "level": "info",
479 }
480 ),
481 ]
482 response_content = "\n".join(log_lines) 1456789!
484 respx_mock.get(url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*").mock( 1456789!
485 return_value=httpx.Response(200, content=response_content)
486 )
488 with changing_dir(configured_app.path): 1456789!
489 result = runner.invoke(app, ["logs", "--no-follow"]) 1456789!
491 assert result.exit_code == 0 1456789!
492 assert "Real log message" in result.output 1456789!
493 assert "heartbeat" not in result.output.lower() 1456789!
496@pytest.mark.parametrize( 1abcdefg
497 "error",
498 [TooManyRetriesError, TimeoutError],
499)
500def test_handles_connection_loss( 1abcdefg
501 logged_in_cli: None,
502 configured_app: ConfiguredApp,
503 error: type[Exception],
504) -> None:
505 with ( 2Zb0b
506 changing_dir(configured_app.path),
507 patch(
508 "fastapi_cloud_cli.api.APIClient.stream_app_logs",
509 side_effect=error("Connection lost"),
510 ),
511 ):
512 result = runner.invoke(app, ["logs", "--no-follow"]) 2Zb~bacbc0bccdc
514 assert result.exit_code == 1 2Zb~bacbc0bccdc
515 assert "Lost connection to log stream" in result.output 2Zb~bacbc0bccdc
518@pytest.mark.parametrize( 1abcdefg
519 "invalid_since",
520 [
521 "5", # missing unit
522 "m", # missing number
523 "5x", # invalid unit
524 "5min", # invalid unit (should be 'm')
525 "1hour", # invalid unit (should be 'h')
526 "5 m", # space not allowed
527 "-5m", # negative not allowed
528 "", # empty string
529 ],
530)
531def test_rejects_invalid_since_format( 1abcdefg
532 logged_in_cli: None,
533 configured_app: ConfiguredApp,
534 invalid_since: str,
535) -> None:
536 with changing_dir(configured_app.path): 21b2b3b4b5b6b7b
537 result = runner.invoke(app, ["logs", "--since", invalid_since]) 21b2b3b4b5b6b7b
539 assert result.exit_code == 2 21b2b3b4b5b6b7b
540 assert "Invalid format" in result.output 21b2b3b4b5b6b7b
543@pytest.mark.parametrize("invalid_tail", [0, 1001]) 1abcdefg
544def test_rejects_invalid_tail( 1abcdefg
545 logged_in_cli: None,
546 configured_app: ConfiguredApp,
547 invalid_tail: int,
548) -> None:
549 with changing_dir(configured_app.path): 28b9b!b#b$b%b'b
550 result = runner.invoke(app, ["logs", "--tail", str(invalid_tail)]) 28b9b!b#b$b%b'b
552 assert result.exit_code == 2 28b9b!b#b$b%b'b
553 assert "between 1 and 1000" in result.output 28b9b!b#b$b%b'b
556@pytest.mark.respx 1abcdefg
557@pytest.mark.parametrize( 1abcdefg
558 "valid_since",
559 [
560 "30s", # seconds
561 "5m", # minutes
562 "1h", # hours
563 "2d", # days
564 "100m", # larger numbers
565 ],
566)
567def test_accepts_valid_since_format( 1abcdefg
568 logged_in_cli: None,
569 respx_mock: respx.MockRouter,
570 configured_app: ConfiguredApp,
571 valid_since: str,
572) -> None:
573 respx_mock.get( 2(b)b*b+b,b-b.b
574 url__regex=rf"/apps/{configured_app.app_id}/logs/stream.*",
575 params={"since": valid_since},
576 ).mock(return_value=httpx.Response(200, content=""))
578 with changing_dir(configured_app.path): 2(b)b*b+b,b-b.b
579 result = runner.invoke(app, ["logs", "--no-follow", "--since", valid_since]) 2(b)b*b+b,b-b.b
581 assert result.exit_code == 0 2(b)b*b+b,b-b.b