Improve EOF error messaging during CLI login flows

changelog.d/20241024_113828_chris_command_line_login_flow_manager_catch_eof.rst

@@ -0,0 +1,5 @@
+Changed
+~~~~~~~
+
+- Improved error messaging around EOF errors when prompting for code during a command
+  line login flow (:pr:`NUMBER`)

docs/authorization/login_flows.rst

@@ -52,6 +52,9 @@ Example Code:
     :member-order: bysource
     :show-inheritance:
 
+
+.. autoexception:: CommandLineLoginFlowEOFError
+
 Local Server
 ------------
 

src/globus_sdk/login_flows/__init__.py

@@ -1,4 +1,7 @@
-from .command_line_login_flow_manager import CommandLineLoginFlowManager
+from .command_line_login_flow_manager import (
+    CommandLineLoginFlowEOFError,
+    CommandLineLoginFlowManager,
+)
 from .local_server_login_flow_manager import (
     LocalServerEnvironmentalLoginError,
     LocalServerLoginError,
@@ -8,6 +11,7 @@
 
 __all__ = (
     "CommandLineLoginFlowManager",
+    "CommandLineLoginFlowEOFError",
     "LocalServerLoginError",
     "LocalServerEnvironmentalLoginError",
     "LocalServerLoginFlowManager",

src/globus_sdk/login_flows/command_line_login_flow_manager.py

@@ -2,8 +2,10 @@
 
 import textwrap
 import typing as t
+from contextlib import contextmanager
 
 import globus_sdk
+from globus_sdk.exc.base import GlobusError
 from globus_sdk.gare import GlobusAuthorizationParameters
 from globus_sdk.utils import get_nice_hostname
 
@@ -13,6 +15,13 @@
     from globus_sdk.globus_app import GlobusAppConfig
 
 
+class CommandLineLoginFlowEOFError(GlobusError, EOFError):
+    """
+    An error raised when a CommandLineLoginFlowManager reads an EOF when prompting for
+    a code.
+    """
+
+
 class CommandLineLoginFlowManager(LoginFlowManager):
     """
     A login flow manager which drives authorization-code token grants through the
@@ -128,4 +137,16 @@ def prompt_for_code(self) -> str:
         :returns: The authorization code entered by the user.
         """
         code_prompt = "Enter the resulting Authorization Code here: "
-        return input(code_prompt).strip()
+        with self._handle_input_errors():
+            return input(code_prompt).strip()
+
+    @contextmanager
+    def _handle_input_errors(self) -> t.Iterator[None]:
+        try:
+            yield
+        except EOFError as e:
+            msg = (
+                "An EOF was read when an authorization code was expected."
+                " (Are you running this in an interactive terminal?)"
+            )
+            raise CommandLineLoginFlowEOFError(msg) from e

tests/functional/login_flows/test_login_flow_manager.py

@@ -1,12 +1,17 @@
 from unittest.mock import Mock, patch
 
+import pytest
+
 from globus_sdk import ConfidentialAppAuthClient, NativeAppAuthClient
 from globus_sdk._testing import load_response
 from globus_sdk.gare import GlobusAuthorizationParameters
 from globus_sdk.login_flows import (
     CommandLineLoginFlowManager,
     LocalServerLoginFlowManager,
 )
+from globus_sdk.login_flows.command_line_login_flow_manager import (
+    CommandLineLoginFlowEOFError,
+)
 
 
 def _mock_input(s):
@@ -96,6 +101,28 @@ def test_command_line_login_flower_manager_confidential(monkeypatch, capsys):
     assert "&session_required_single_domain=org.edu" in captured_output
 
 
+def test_command_line_login_flow_manager_eof_error(monkeypatch, capsys):
+    """
+    test CommandLineLoginFlowManager with a NativeAppAuthClient
+    """
+    login_client = NativeAppAuthClient("mock_client_id")
+    load_response(login_client.oauth2_exchange_code_for_tokens)
+
+    def broken_input(_s):
+        raise EOFError()
+
+    monkeypatch.setattr("builtins.input", broken_input)
+
+    login_flow_manager = CommandLineLoginFlowManager(login_client)
+    auth_params = GlobusAuthorizationParameters(
+        required_scopes=["urn:globus:auth:scope:transfer.api.globus.org:all"],
+        session_required_identities=["user@org.edu"],
+    )
+
+    with pytest.raises(CommandLineLoginFlowEOFError):
+        login_flow_manager.run_login_flow(auth_params)
+
+
 class MockRedirectServer:
     def __init__(self, *args, **kwargs):
         self.socket = Mock()