Coverage for tests/test_cli_login.py: 100%
219 statements
« 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 httpx import Response 1abcdefg
9from inline_snapshot import snapshot 1abcdefg
10from typer.testing import CliRunner 1abcdefg
12from fastapi_cloud_cli.cli import app 1abcdefg
13from fastapi_cloud_cli.config import Settings 1abcdefg
14from tests.utils import SnapshotCliRunner 1abcdefg
16runner = CliRunner() 1abcdefg
18assets_path = Path(__file__).parent / "assets" 1abcdefg
21@pytest.mark.respx 1abcdefg
22def test_shows_a_message_if_something_is_wrong( 1abcdefg
23 logged_out_cli: None, respx_mock: respx.MockRouter, settings: Settings
24) -> None:
25 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1-./:;=?
26 respx_mock.post( 1-./:;=?
27 "/login/device/authorization", data={"client_id": settings.client_id}
28 ).mock(return_value=Response(500))
30 result = runner.invoke(app, ["login"]) 1-./:;=?
32 assert result.exit_code == 1 1-./:;=?
33 assert ( 1-./:;=?
34 "Something went wrong while contacting the FastAPI Cloud server."
35 in result.output
36 )
38 assert not mock_open.called 1-./:;=?
41@pytest.mark.respx 1abcdefg
42def test_full_login( 1abcdefg
43 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
44) -> None:
45 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1vwxyzAB
46 respx_mock.post( 1vwxyzAB
47 "/login/device/authorization", data={"client_id": settings.client_id}
48 ).mock(
49 return_value=Response(
50 200,
51 json={
52 "verification_uri_complete": (
53 "http://test.com/device?code=query-code"
54 ),
55 "verification_uri": "http://test.com",
56 "user_code": "manual-code",
57 "device_code": "5678",
58 },
59 )
60 )
61 respx_mock.post( 1vwxyzAB
62 "/login/device/token",
63 data={
64 "device_code": "5678",
65 "client_id": settings.client_id,
66 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
67 },
68 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
70 # Verify no auth file exists before login
71 assert not temp_auth_config.exists() 1vwxyzAB
73 result = runner.invoke(app, ["login"]) 1vwxyzAB
75 assert result.exit_code == 0 1vwxyzAB
76 assert mock_open.called 1vwxyzAB
77 assert mock_open.call_args.args == ("http://test.com/device?code=query-code",) 1vwxyzAB
78 assert "Opening http://test.com/device?code=query-code" in result.output 1vwxyzAB
79 assert "Now you are logged in!" in result.output 1vwxyzAB
81 # Verify auth file was created with correct content
82 assert temp_auth_config.exists() 1vwxyzAB
83 assert '"access_token":"test_token_1234"' in temp_auth_config.read_text() 1vwxyzAB
86@pytest.mark.respx 1abcdefg
87def test_cloud_auth_login_alias_uses_interactive_login( 1abcdefg
88 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
89) -> None:
90 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1CDEFGHI
91 respx_mock.post( 1CDEFGHI
92 "/login/device/authorization", data={"client_id": settings.client_id}
93 ).mock(
94 return_value=Response(
95 200,
96 json={
97 "verification_uri_complete": (
98 "http://test.com/device?code=query-code"
99 ),
100 "verification_uri": "http://test.com",
101 "user_code": "manual-code",
102 "device_code": "5678",
103 },
104 )
105 )
106 respx_mock.post( 1CDEFGHI
107 "/login/device/token",
108 data={
109 "device_code": "5678",
110 "client_id": settings.client_id,
111 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
112 },
113 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
115 result = runner.invoke(app, ["cloud", "auth", "login", "--no-open"]) 1CDEFGHI
117 assert result.exit_code == 0 1CDEFGHI
118 assert "Open http://test.com/device?code=query-code" in result.output 1CDEFGHI
119 assert "enter code" not in result.output 1CDEFGHI
120 assert "manual-code" not in result.output 1CDEFGHI
121 mock_open.assert_not_called() 1CDEFGHI
122 assert temp_auth_config.exists() 1CDEFGHI
123 assert '"access_token":"test_token_1234"' in temp_auth_config.read_text() 1CDEFGHI
126@pytest.mark.respx 1abcdefg
127def test_cloud_login_json_returns_authorization_data( 1abcdefg
128 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
129) -> None:
130 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1XYZ0123
131 respx_mock.post( 1XYZ0123
132 "/login/device/authorization", data={"client_id": settings.client_id}
133 ).mock(
134 return_value=Response(
135 200,
136 json={
137 "verification_uri_complete": (
138 "http://test.com/device?code=query-code"
139 ),
140 "verification_uri": "http://test.com",
141 "user_code": "manual-code",
142 "device_code": "5678",
143 },
144 )
145 )
147 result = runner.invoke(app, ["cloud", "login", "--json"]) 1XYZ0123
149 assert result.exit_code == 0 1XYZ0123
150 assert json.loads(result.stdout) == { 1XYZ0123
151 "data": {
152 "verification_uri": "http://test.com",
153 "verification_uri_complete": ("http://test.com/device?code=query-code"),
154 "user_code": "manual-code",
155 "device_code": "5678",
156 "interval": 5,
157 }
158 }
159 mock_open.assert_not_called() 1XYZ0123
160 assert not temp_auth_config.exists() 1XYZ0123
163@pytest.mark.respx 1abcdefg
164def test_top_level_login_json_returns_authorization_data( 1abcdefg
165 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
166) -> None:
167 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1456789!
168 respx_mock.post( 1456789!
169 "/login/device/authorization", data={"client_id": settings.client_id}
170 ).mock(
171 return_value=Response(
172 200,
173 json={
174 "verification_uri_complete": (
175 "http://test.com/device?code=query-code"
176 ),
177 "verification_uri": "http://test.com",
178 "user_code": "manual-code",
179 "device_code": "5678",
180 },
181 )
182 )
184 result = runner.invoke(app, ["login", "--json"]) 1456789!
186 assert result.exit_code == 0 1456789!
187 assert json.loads(result.stdout) == { 1456789!
188 "data": {
189 "verification_uri": "http://test.com",
190 "verification_uri_complete": ("http://test.com/device?code=query-code"),
191 "user_code": "manual-code",
192 "device_code": "5678",
193 "interval": 5,
194 }
195 }
196 mock_open.assert_not_called() 1456789!
197 assert not temp_auth_config.exists() 1456789!
200def test_auth_start_is_not_supported(logged_in_cli: None) -> None: 1abcdefg
201 result = runner.invoke(app, ["cloud", "auth", "start"]) 2%b'b(b)b*b+b,b
203 assert result.exit_code != 0 2%b'b(b)b*b+b,b
204 assert "No such command 'start'" in result.output 2%b'b(b)b*b+b,b
207@pytest.mark.respx 1abcdefg
208def test_auth_login_as_json_returns_authorization_data( 1abcdefg
209 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
210) -> None:
211 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1#$%'()*
212 respx_mock.post( 1#$%'()*
213 "/login/device/authorization", data={"client_id": settings.client_id}
214 ).mock(
215 return_value=Response(
216 200,
217 json={
218 "verification_uri_complete": (
219 "http://test.com/device?code=query-code"
220 ),
221 "verification_uri": "http://test.com",
222 "user_code": "manual-code",
223 "device_code": "5678",
224 },
225 )
226 )
228 result = runner.invoke(app, ["cloud", "auth", "login", "--json"]) 1#$%'()*
230 assert result.exit_code == 0 1#$%'()*
231 assert json.loads(result.stdout) == { 1#$%'()*
232 "data": {
233 "verification_uri": "http://test.com",
234 "verification_uri_complete": ("http://test.com/device?code=query-code"),
235 "user_code": "manual-code",
236 "device_code": "5678",
237 "interval": 5,
238 }
239 }
240 mock_open.assert_not_called() 1#$%'()*
241 assert not temp_auth_config.exists() 1#$%'()*
244@pytest.mark.respx 1abcdefg
245def test_auth_wait_as_json_writes_auth_config( 1abcdefg
246 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
247) -> None:
248 respx_mock.post( 1@[]^_`{
249 "/login/device/token",
250 data={
251 "device_code": "5678",
252 "client_id": settings.client_id,
253 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
254 },
255 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
257 result = runner.invoke( 1@[]^_`{
258 app, ["cloud", "auth", "wait", "--device-code", "5678", "--json"]
259 )
261 assert result.exit_code == 0 1@[]^_`{
262 assert json.loads(result.stdout) == { 1@[]^_`{
263 "data": {"authenticated": True, "auth_mode": "user"}
264 }
265 assert temp_auth_config.exists() 1@[]^_`{
266 assert '"access_token":"test_token_1234"' in temp_auth_config.read_text() 1@[]^_`{
269@pytest.mark.respx 1abcdefg
270def test_auth_wait_shows_login_title( 1abcdefg
271 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
272) -> None:
273 respx_mock.post( 2LbMbNbObPbQbRb
274 "/login/device/token",
275 data={
276 "device_code": "5678",
277 "client_id": settings.client_id,
278 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
279 },
280 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
282 result = runner.invoke(app, ["cloud", "auth", "wait", "--device-code", "5678"]) 2LbMbNbObPbQbRb
284 assert result.exit_code == 0 2LbMbNbObPbQbRb
285 assert "Login to FastAPI Cloud" in result.output 2LbMbNbObPbQbRb
286 assert temp_auth_config.exists() 2LbMbNbObPbQbRb
289@pytest.mark.respx 1abcdefg
290def test_full_login_with_no_open_shows_manual_authorization_details( 1abcdefg
291 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
292) -> None:
293 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1JKLMNOP
294 respx_mock.post( 1JKLMNOP
295 "/login/device/authorization", data={"client_id": settings.client_id}
296 ).mock(
297 return_value=Response(
298 200,
299 json={
300 "verification_uri_complete": (
301 "http://test.com/device?code=query-code"
302 ),
303 "verification_uri": "http://test.com",
304 "user_code": "manual-code",
305 "device_code": "5678",
306 },
307 )
308 )
309 respx_mock.post( 1JKLMNOP
310 "/login/device/token",
311 data={
312 "device_code": "5678",
313 "client_id": settings.client_id,
314 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
315 },
316 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
318 result = runner.invoke(app, ["cloud", "login", "--no-open"]) 1JKLMNOP
320 assert result.exit_code == 0 1JKLMNOP
321 assert "Open http://test.com/device?code=query-code" in result.output 1JKLMNOP
322 assert "enter code" not in result.output 1JKLMNOP
323 assert "manual-code" not in result.output 1JKLMNOP
324 mock_open.assert_not_called() 1JKLMNOP
325 assert temp_auth_config.exists() 1JKLMNOP
328@pytest.mark.respx 1abcdefg
329def test_auth_wait_json_timeout_returns_error_envelope( 1abcdefg
330 respx_mock: respx.MockRouter,
331 temp_auth_config: Path,
332 settings: Settings,
333) -> None:
334 respx_mock.post( 2| } ~ abbbcbdb
335 "/login/device/token",
336 data={
337 "device_code": "5678",
338 "client_id": settings.client_id,
339 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
340 },
341 ).mock(return_value=Response(400, json={"error": "authorization_pending"}))
343 with patch( 2| } ~ abbbcbdb
344 "fastapi_cloud_cli.commands._flow.time.monotonic",
345 side_effect=[0.0, 11.0],
346 ):
347 result = runner.invoke( 2| } ~ abbbcbdb
348 app,
349 [
350 "cloud",
351 "auth",
352 "wait",
353 "--device-code",
354 "5678",
355 "--timeout",
356 "10",
357 "--json",
358 ],
359 )
361 assert result.exit_code == 1 2| } ~ abbbcbdb
362 assert json.loads(result.stdout) == { 2| } ~ abbbcbdb
363 "error": {
364 "code": "timeout",
365 "message": "Login timed out before authorization completed.",
366 "hint": "Try again with a longer --timeout value.",
367 }
368 }
369 assert not temp_auth_config.exists() 2| } ~ abbbcbdb
372@pytest.mark.respx 1abcdefg
373def test_auth_wait_json_cancel_returns_error_envelope( 1abcdefg
374 temp_auth_config: Path,
375) -> None:
376 with patch( 2SbTbUbVbWbXbYb
377 "fastapi_cloud_cli.commands._flow.fetch_access_token",
378 side_effect=KeyboardInterrupt,
379 ):
380 result = runner.invoke( 2SbTbUbVbWbXbYb
381 app,
382 [
383 "cloud",
384 "auth",
385 "wait",
386 "--device-code",
387 "5678",
388 "--json",
389 ],
390 )
392 assert result.exit_code == 1 2SbTbUbVbWbXbYb
393 assert json.loads(result.stdout) == { 2SbTbUbVbWbXbYb
394 "error": {
395 "code": "cancelled",
396 "message": "Login cancelled before authorization completed.",
397 "hint": "Run `fastapi cloud auth wait --json` again to retry.",
398 }
399 }
400 assert not temp_auth_config.exists() 2SbTbUbVbWbXbYb
403@pytest.mark.respx 1abcdefg
404@pytest.mark.parametrize( 1abcdefg
405 "command", [["login"], ["cloud", "login"], ["cloud", "auth", "login"]]
406)
407def test_login_shows_authorization_declined( 1abcdefg
408 command: list[str],
409 respx_mock: respx.MockRouter,
410 temp_auth_config: Path,
411 settings: Settings,
412) -> None:
413 respx_mock.post( 2ebfbgbhbibjbkb
414 "/login/device/authorization", data={"client_id": settings.client_id}
415 ).respond(
416 200,
417 json={
418 "verification_uri_complete": "http://test.com/device?code=query-code",
419 "verification_uri": "http://test.com",
420 "user_code": "manual-code",
421 "device_code": "5678",
422 },
423 )
424 respx_mock.post( 2ebfbgbhbibjbkb
425 "/login/device/token",
426 data={
427 "device_code": "5678",
428 "client_id": settings.client_id,
429 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
430 },
431 ).respond(400, json={"error": "access_denied"})
433 result = SnapshotCliRunner().invoke(app, [*command, "--no-open"]) 2ebfbgbhbibjbkb
435 assert result.exit_code == 1 2ebfbgbhbibjbkb
436 assert result.output == snapshot("""\ 2ebfbgbhbibjbkb
437FastAPI Cloud
439🔐 Login to FastAPI Cloud
441 Open http://test.com/device?code=query-code
443 ✗ error: Login authorization was declined.
445 hint: Run `fastapi cloud login` again and approve the authorization request.\
446""")
447 assert not temp_auth_config.exists() 2ebfbgbhbibjbkb
450@pytest.mark.respx 1abcdefg
451def test_auth_wait_shows_authorization_declined( 1abcdefg
452 respx_mock: respx.MockRouter,
453 temp_auth_config: Path,
454 settings: Settings,
455) -> None:
456 respx_mock.post( 2Zb0b1b2b3b4b5b
457 "/login/device/token",
458 data={
459 "device_code": "5678",
460 "client_id": settings.client_id,
461 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
462 },
463 ).respond(400, json={"error": "access_denied"})
465 result = SnapshotCliRunner().invoke( 2Zb0b1b2b3b4b5b
466 app, ["cloud", "auth", "wait", "--device-code", "5678"]
467 )
469 assert result.exit_code == 1 2Zb0b1b2b3b4b5b
470 assert result.output == snapshot("""\ 2Zb0b1b2b3b4b5b
471FastAPI Cloud
473🔐 Login to FastAPI Cloud
475 ✗ error: Login authorization was declined.
477 hint: Run `fastapi cloud login` again and approve the authorization request.\
478""")
479 assert not temp_auth_config.exists() 2Zb0b1b2b3b4b5b
482@pytest.mark.respx 1abcdefg
483@pytest.mark.parametrize("pending_first", [False, True]) 1abcdefg
484def test_auth_wait_json_authorization_declined_returns_error_envelope( 1abcdefg
485 pending_first: bool,
486 respx_mock: respx.MockRouter,
487 temp_auth_config: Path,
488 settings: Settings,
489) -> None:
490 responses = [] 1hijklmn
491 if pending_first: 1hijklmn
492 responses.append(Response(400, json={"error": "authorization_pending"})) 1hijklmn
493 responses.append(Response(400, json={"error": "access_denied"})) 1hijklmn
494 respx_mock.post( 1hijklmn
495 "/login/device/token",
496 data={
497 "device_code": "5678",
498 "client_id": settings.client_id,
499 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
500 },
501 ).mock(side_effect=responses)
503 with patch("fastapi_cloud_cli.commands._flow.time.sleep") as mock_sleep: 1hijklmn
504 result = runner.invoke( 1hijklmn
505 app, ["cloud", "auth", "wait", "--device-code", "5678", "--json"]
506 )
508 assert result.exit_code == 1 1hijklmn
509 assert json.loads(result.stdout) == { 1hijklmn
510 "error": {
511 "code": "permission_denied",
512 "message": "Login authorization was declined.",
513 "hint": "Run `fastapi cloud login` again and approve the authorization request.",
514 }
515 }
516 assert result.stderr == "" 1hijklmn
517 if pending_first: 1hijklmn
518 mock_sleep.assert_called_once_with(5) 1hijklmn
519 else:
520 mock_sleep.assert_not_called() 1hijklmn
521 assert not temp_auth_config.exists() 1hijklmn
524@pytest.mark.respx 1abcdefg
525def test_fetch_access_token_uses_default_timeout( 1abcdefg
526 respx_mock: respx.MockRouter, settings: Settings
527) -> None:
528 from fastapi_cloud_cli.api import APIClient 2+ lbmbnb, obpb
529 from fastapi_cloud_cli.commands._flow import ( 2+ lbmbnb, obpb
530 LoginTimeoutError,
531 fetch_access_token,
532 )
534 respx_mock.post( 2+ lbmbnb, obpb
535 "/login/device/token",
536 data={
537 "device_code": "test_device_code",
538 "client_id": settings.client_id,
539 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
540 },
541 ).mock(return_value=Response(400, json={"error": "authorization_pending"}))
543 with ( 1+,
544 patch(
545 "fastapi_cloud_cli.commands._flow.time.monotonic",
546 side_effect=[0.0, 301.0],
547 ),
548 patch(
549 "fastapi_cloud_cli.commands._flow.time.sleep",
550 side_effect=AssertionError("sleep should not run after timeout"),
551 ),
552 ):
553 with APIClient() as client: 2+ lbmbnb, obpb
554 with pytest.raises(LoginTimeoutError): 2+ lbmbnb, obpb
555 fetch_access_token(client, "test_device_code", 5) 2+ lbmbnb, obpb
558@pytest.mark.respx 1abcdefg
559def test_full_login_with_deploy_token_set( 1abcdefg
560 respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
561) -> None:
562 with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open: 1opqrstu
563 respx_mock.post( 1opqrstu
564 "/login/device/authorization", data={"client_id": settings.client_id}
565 ).mock(
566 return_value=Response(
567 200,
568 json={
569 "verification_uri_complete": "http://test.com",
570 "verification_uri": "http://test.com",
571 "user_code": "1234",
572 "device_code": "5678",
573 },
574 )
575 )
576 respx_mock.post( 1opqrstu
577 "/login/device/token",
578 data={
579 "device_code": "5678",
580 "client_id": settings.client_id,
581 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
582 },
583 ).mock(return_value=Response(200, json={"access_token": "test_token_1234"}))
585 # Verify no auth file exists before login
586 assert not temp_auth_config.exists() 1opqrstu
588 result = runner.invoke( 1opqrstu
589 app,
590 ["login"],
591 env={"FASTAPI_CLOUD_TOKEN": "test_deploy_token"}, # Should be ignored
592 )
594 assert result.exit_code == 0 1opqrstu
595 assert mock_open.called 1opqrstu
596 assert mock_open.call_args.args == ("http://test.com",) 1opqrstu
598 # Verify the warning message is shown
599 assert "You have FASTAPI_CLOUD_TOKEN environment variable set." in result.output 1opqrstu
600 assert "This token will take precedence over the user token" in result.output 1opqrstu
602 assert "Now you are logged in!" in result.output 1opqrstu
604 # Verify auth file was created with correct content
605 assert temp_auth_config.exists() 1opqrstu
606 assert '"access_token":"test_token_1234"' in temp_auth_config.read_text() 1opqrstu
609@pytest.mark.respx 1abcdefg
610def test_fetch_access_token_success_immediately( 1abcdefg
611 respx_mock: respx.MockRouter, settings: Settings
612) -> None:
613 from fastapi_cloud_cli.api import APIClient 2qbrbsbtbubvbwb
614 from fastapi_cloud_cli.commands._flow import fetch_access_token 2qbrbsbtbubvbwb
616 respx_mock.post( 2qbrbsbtbubvbwb
617 "/login/device/token",
618 data={
619 "device_code": "test_device_code",
620 "client_id": settings.client_id,
621 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
622 },
623 ).mock(return_value=Response(200, json={"access_token": "test_token_success"}))
625 with APIClient() as client: 2qbrbsbtbubvbwb
626 access_token = fetch_access_token(client, "test_device_code", 5) 2qbrbsbtbubvbwb
628 assert access_token == "test_token_success" 2qbrbsbtbubvbwb
631@pytest.mark.respx 1abcdefg
632def test_fetch_access_token_authorization_pending_then_success( 1abcdefg
633 respx_mock: respx.MockRouter,
634 settings: Settings,
635) -> None:
636 from fastapi_cloud_cli.api import APIClient 1QRSTUVW
637 from fastapi_cloud_cli.commands._flow import fetch_access_token 1QRSTUVW
639 # First call returns authorization pending, second call succeeds
640 respx_mock.post( 1QRSTUVW
641 "/login/device/token",
642 data={
643 "device_code": "test_device_code",
644 "client_id": settings.client_id,
645 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
646 },
647 ).mock(
648 side_effect=[
649 Response(400, json={"error": "authorization_pending"}),
650 Response(200, json={"access_token": "test_token_after_pending"}),
651 ]
652 )
654 with patch("fastapi_cloud_cli.commands._flow.time.sleep") as mock_sleep: 1QRSTUVW
655 with APIClient() as client: 1QRSTUVW
656 access_token = fetch_access_token(client, "test_device_code", 3) 1QRSTUVW
658 assert access_token == "test_token_after_pending" 1QRSTUVW
659 mock_sleep.assert_called_once_with(3) 1QRSTUVW
662@pytest.mark.respx 1abcdefg
663def test_fetch_access_token_handles_unknown_oauth_error( 1abcdefg
664 respx_mock: respx.MockRouter,
665 settings: Settings,
666) -> None:
667 from fastapi_cloud_cli.api import APIClient 2xbybzbAbBbCbDb
668 from fastapi_cloud_cli.commands._flow import fetch_access_token 2xbybzbAbBbCbDb
670 respx_mock.post( 2xbybzbAbBbCbDb
671 "/login/device/token",
672 data={
673 "device_code": "test_device_code",
674 "client_id": settings.client_id,
675 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
676 },
677 ).mock(return_value=Response(400, json={"error": "unknown_error"}))
679 with APIClient() as client: 2xbybzbAbBbCbDb
680 with pytest.raises(httpx.HTTPStatusError): 2xbybzbAbBbCbDb
681 fetch_access_token(client, "test_device_code", 5) 2xbybzbAbBbCbDb
684@pytest.mark.respx 1abcdefg
685def test_fetch_access_token_handles_500_error( 1abcdefg
686 respx_mock: respx.MockRouter, settings: Settings
687) -> None:
688 from fastapi_cloud_cli.api import APIClient 2EbFbGbHbIbJbKb
689 from fastapi_cloud_cli.commands._flow import fetch_access_token 2EbFbGbHbIbJbKb
691 respx_mock.post( 2EbFbGbHbIbJbKb
692 "/login/device/token",
693 data={
694 "device_code": "test_device_code",
695 "client_id": settings.client_id,
696 "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
697 },
698 ).mock(return_value=Response(500))
700 with APIClient() as client: 2EbFbGbHbIbJbKb
701 with pytest.raises(httpx.HTTPStatusError): 2EbFbGbHbIbJbKb
702 fetch_access_token(client, "test_device_code", 5) 2EbFbGbHbIbJbKb
705@pytest.mark.respx 1abcdefg
706def test_notify_already_logged_in_user( 1abcdefg
707 respx_mock: respx.MockRouter, logged_in_cli: None
708) -> None:
709 result = runner.invoke(app, ["login"]) 26b7b8b9b!b#b$b
711 assert result.exit_code == 0 26b7b8b9b!b#b$b
712 assert "You are already logged in." in result.output 26b7b8b9b!b#b$b
713 assert ( 26b7b8b9b!b#b$b
714 "Run fastapi cloud logout first if you want to switch accounts."
715 in result.output
716 )