From 2a7dac356b0230ab6ece904459af40310d5cf43b Mon Sep 17 00:00:00 2001 From: Patrick Arminio Date: Thu, 5 Nov 2026 11:44:34 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Handle=20declined=20OAuth=20devi?= =?UTF-8?q?ce=20authorization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/fastapi_cloud_cli/commands/_flow.py | 13 +++ tests/test_cli_login.py | 127 +++++++++++++++++++++++- 2 files changed, 138 insertions(+), 2 deletions(-) diff --git a/src/fastapi_cloud_cli/commands/_flow.py b/src/fastapi_cloud_cli/commands/_flow.py index 92038655..1ef28dde 100644 --- a/src/fastapi_cloud_cli/commands/_flow.py +++ b/src/fastapi_cloud_cli/commands/_flow.py @@ -45,6 +45,10 @@ class LoginTimeoutError(Exception): pass +class LoginDeniedError(Exception): + pass + + def render_login_output(data: LoginOutput, toolkit: RichToolkit) -> None: toolkit.print("Now you are logged in! 🚀") @@ -105,6 +109,9 @@ def fetch_access_token( error = data.get("error") logger.debug(f"Token response error: {error}") + if error == "access_denied": + raise LoginDeniedError + if error != "authorization_pending": response.raise_for_status() @@ -139,6 +146,12 @@ def complete_device_login( try: with client.handle_http_errors(progress, toolkit=toolkit): access_token = fetch_access_token(client, device_code, interval, timeout) + except LoginDeniedError: + toolkit.fail( + "permission_denied", + "Login authorization was declined.", + hint="Run `fastapi cloud login` again and approve the authorization request.", + ) except LoginTimeoutError: message = "Login timed out before authorization completed." toolkit.fail( diff --git a/tests/test_cli_login.py b/tests/test_cli_login.py index 079e4045..33114128 100644 --- a/tests/test_cli_login.py +++ b/tests/test_cli_login.py @@ -6,10 +6,12 @@ import pytest import respx from httpx import Response +from inline_snapshot import snapshot from typer.testing import CliRunner from fastapi_cloud_cli.cli import app from fastapi_cloud_cli.config import Settings +from tests.utils import SnapshotCliRunner runner = CliRunner() @@ -398,6 +400,127 @@ def test_auth_wait_json_cancel_returns_error_envelope( assert not temp_auth_config.exists() +@pytest.mark.respx +@pytest.mark.parametrize( + "command", [["login"], ["cloud", "login"], ["cloud", "auth", "login"]] +) +def test_login_shows_authorization_declined( + command: list[str], + respx_mock: respx.MockRouter, + temp_auth_config: Path, + settings: Settings, +) -> None: + respx_mock.post( + "/login/device/authorization", data={"client_id": settings.client_id} + ).respond( + 200, + json={ + "verification_uri_complete": "http://test.com/device?code=query-code", + "verification_uri": "http://test.com", + "user_code": "manual-code", + "device_code": "5678", + }, + ) + respx_mock.post( + "/login/device/token", + data={ + "device_code": "5678", + "client_id": settings.client_id, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + }, + ).respond(400, json={"error": "access_denied"}) + + result = SnapshotCliRunner().invoke(app, [*command, "--no-open"]) + + assert result.exit_code == 1 + assert result.output == snapshot("""\ +FastAPI Cloud + +🔐 Login to FastAPI Cloud + + Open http://test.com/device?code=query-code + + ✗ error: Login authorization was declined. + + hint: Run `fastapi cloud login` again and approve the authorization request.\ +""") + assert not temp_auth_config.exists() + + +@pytest.mark.respx +def test_auth_wait_shows_authorization_declined( + respx_mock: respx.MockRouter, + temp_auth_config: Path, + settings: Settings, +) -> None: + respx_mock.post( + "/login/device/token", + data={ + "device_code": "5678", + "client_id": settings.client_id, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + }, + ).respond(400, json={"error": "access_denied"}) + + result = SnapshotCliRunner().invoke( + app, ["cloud", "auth", "wait", "--device-code", "5678"] + ) + + assert result.exit_code == 1 + assert result.output == snapshot("""\ +FastAPI Cloud + +🔐 Login to FastAPI Cloud + + ✗ error: Login authorization was declined. + + hint: Run `fastapi cloud login` again and approve the authorization request.\ +""") + assert not temp_auth_config.exists() + + +@pytest.mark.respx +@pytest.mark.parametrize("pending_first", [False, True]) +def test_auth_wait_json_authorization_declined_returns_error_envelope( + pending_first: bool, + respx_mock: respx.MockRouter, + temp_auth_config: Path, + settings: Settings, +) -> None: + responses = [] + if pending_first: + responses.append(Response(400, json={"error": "authorization_pending"})) + responses.append(Response(400, json={"error": "access_denied"})) + respx_mock.post( + "/login/device/token", + data={ + "device_code": "5678", + "client_id": settings.client_id, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + }, + ).mock(side_effect=responses) + + with patch("fastapi_cloud_cli.commands._flow.time.sleep") as mock_sleep: + result = runner.invoke( + app, ["cloud", "auth", "wait", "--device-code", "5678", "--json"] + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "permission_denied", + "message": "Login authorization was declined.", + "hint": "Run `fastapi cloud login` again and approve the authorization request.", + } + } + assert result.stderr == "" + if pending_first: + mock_sleep.assert_called_once_with(5) + else: + mock_sleep.assert_not_called() + assert not temp_auth_config.exists() + + @pytest.mark.respx def test_fetch_access_token_uses_default_timeout( respx_mock: respx.MockRouter, settings: Settings @@ -537,7 +660,7 @@ def test_fetch_access_token_authorization_pending_then_success( @pytest.mark.respx -def test_fetch_access_token_handles_400_error_not_authorization_pending( +def test_fetch_access_token_handles_unknown_oauth_error( respx_mock: respx.MockRouter, settings: Settings, ) -> None: @@ -551,7 +674,7 @@ def test_fetch_access_token_handles_400_error_not_authorization_pending( "client_id": settings.client_id, "grant_type": "urn:ietf:params:oauth:grant-type:device_code", }, - ).mock(return_value=Response(400, json={"error": "access_denied"})) + ).mock(return_value=Response(400, json={"error": "unknown_error"})) with APIClient() as client: with pytest.raises(httpx.HTTPStatusError):