Better exception handling around self-check

src/pip/_internal/cli/base_command.py

@@ -1,14 +1,13 @@
 """Base Command class, and related routines"""
 
-import functools
 import logging
 import logging.config
 import optparse
 import os
 import sys
 import traceback
 from optparse import Values
-from typing import Any, Callable, List, Optional, Tuple
+from typing import List, Optional, Tuple
 
 from pip._vendor.rich import reconfigure
 from pip._vendor.rich import traceback as rich_traceback
@@ -91,6 +90,63 @@ def handle_pip_version_check(self, options: Values) -> None:
     def run(self, options: Values, args: List[str]) -> int:
         raise NotImplementedError
 
+    def _run_wrapper(self, level_number: int, options: Values, args: List[str]) -> int:
+        def _inner_run() -> int:
+            try:
+                return self.run(options, args)
+            finally:
+                self.handle_pip_version_check(options)
+
+        if options.debug_mode:
+            rich_traceback.install(show_locals=True)
+            return _inner_run()
+
+        try:
+            status = _inner_run()
+            assert isinstance(status, int)
+            return status
+        except DiagnosticPipError as exc:
+            logger.error("%s", exc, extra={"rich": True})
+            logger.debug("Exception information:", exc_info=True)
+
+            return ERROR
+        except PreviousBuildDirError as exc:
+            logger.critical(str(exc))
+            logger.debug("Exception information:", exc_info=True)
+
+            return PREVIOUS_BUILD_DIR_ERROR
+        except (
+            InstallationError,
+            BadCommand,
+            NetworkConnectionError,
+        ) as exc:
+            logger.critical(str(exc))
+            logger.debug("Exception information:", exc_info=True)
+
+            return ERROR
+        except CommandError as exc:
+            logger.critical("%s", exc)
+            logger.debug("Exception information:", exc_info=True)
+
+            return ERROR
+        except BrokenStdoutLoggingError:
+            # Bypass our logger and write any remaining messages to
+            # stderr because stdout no longer works.
+            print("ERROR: Pipe to stdout was broken", file=sys.stderr)
+            if level_number <= logging.DEBUG:
+                traceback.print_exc(file=sys.stderr)
+
+            return ERROR
+        except KeyboardInterrupt:
+            logger.critical("Operation cancelled by user")
+            logger.debug("Exception information:", exc_info=True)
+
+            return ERROR
+        except BaseException:
+            logger.critical("Exception:", exc_info=True)
+
+            return UNKNOWN_ERROR
+
     def parse_args(self, args: List[str]) -> Tuple[Values, List[str]]:
         # factored out for testability
         return self.parser.parse_args(args)
@@ -172,65 +228,4 @@ def _main(self, args: List[str]) -> int:
                 )
                 options.cache_dir = None
 
-        def intercepts_unhandled_exc(
-            run_func: Callable[..., int]
-        ) -> Callable[..., int]:
-            @functools.wraps(run_func)
-            def exc_logging_wrapper(*args: Any) -> int:
-                try:
-                    status = run_func(*args)
-                    assert isinstance(status, int)
-                    return status
-                except DiagnosticPipError as exc:
-                    logger.error("%s", exc, extra={"rich": True})
-                    logger.debug("Exception information:", exc_info=True)
-
-                    return ERROR
-                except PreviousBuildDirError as exc:
-                    logger.critical(str(exc))
-                    logger.debug("Exception information:", exc_info=True)
-
-                    return PREVIOUS_BUILD_DIR_ERROR
-                except (
-                    InstallationError,
-                    BadCommand,
-                    NetworkConnectionError,
-                ) as exc:
-                    logger.critical(str(exc))
-                    logger.debug("Exception information:", exc_info=True)
-
-                    return ERROR
-                except CommandError as exc:
-                    logger.critical("%s", exc)
-                    logger.debug("Exception information:", exc_info=True)
-
-                    return ERROR
-                except BrokenStdoutLoggingError:
-                    # Bypass our logger and write any remaining messages to
-                    # stderr because stdout no longer works.
-                    print("ERROR: Pipe to stdout was broken", file=sys.stderr)
-                    if level_number <= logging.DEBUG:
-                        traceback.print_exc(file=sys.stderr)
-
-                    return ERROR
-                except KeyboardInterrupt:
-                    logger.critical("Operation cancelled by user")
-                    logger.debug("Exception information:", exc_info=True)
-
-                    return ERROR
-                except BaseException:
-                    logger.critical("Exception:", exc_info=True)
-
-                    return UNKNOWN_ERROR
-
-            return exc_logging_wrapper
-
-        try:
-            if not options.debug_mode:
-                run = intercepts_unhandled_exc(self.run)
-            else:
-                run = self.run
-                rich_traceback.install(show_locals=True)
-            return run(options, args)
-        finally:
-            self.handle_pip_version_check(options)
+        return self._run_wrapper(level_number, options, args)

src/pip/_internal/cli/index_command.py

@@ -156,11 +156,15 @@ def handle_pip_version_check(self, options: Values) -> None:
         if options.disable_pip_version_check or options.no_index:
             return
 
-        # Otherwise, check if we're using the latest version of pip available.
-        session = self._build_session(
-            options,
-            retries=0,
-            timeout=min(5, options.timeout),
-        )
-        with session:
-            _pip_self_version_check(session, options)
+        try:
+            # Otherwise, check if we're using the latest version of pip available.
+            session = self._build_session(
+                options,
+                retries=0,
+                timeout=min(5, options.timeout),
+            )
+            with session:
+                _pip_self_version_check(session, options)
+        except Exception:
+            logger.warning("There was an error checking the latest version of pip.")
+            logger.debug("See below for error", exc_info=True)

src/pip/_internal/self_outdated_check.py

@@ -232,17 +232,13 @@ def pip_self_version_check(session: PipSession, options: optparse.Values) -> Non
     if not installed_dist:
         return
 
-    try:
-        upgrade_prompt = _self_version_check_logic(
-            state=SelfCheckState(cache_dir=options.cache_dir),
-            current_time=datetime.datetime.now(datetime.timezone.utc),
-            local_version=installed_dist.version,
-            get_remote_version=functools.partial(
-                _get_current_remote_pip_version, session, options
-            ),
-        )
-        if upgrade_prompt is not None:
-            logger.warning("%s", upgrade_prompt, extra={"rich": True})
-    except Exception:
-        logger.warning("There was an error checking the latest version of pip.")
-        logger.debug("See below for error", exc_info=True)
+    upgrade_prompt = _self_version_check_logic(
+        state=SelfCheckState(cache_dir=options.cache_dir),
+        current_time=datetime.datetime.now(datetime.timezone.utc),
+        local_version=installed_dist.version,
+        get_remote_version=functools.partial(
+            _get_current_remote_pip_version, session, options
+        ),
+    )
+    if upgrade_prompt is not None:
+        logger.warning("%s", upgrade_prompt, extra={"rich": True})

tests/unit/test_self_check_outdated.py

@@ -41,6 +41,7 @@ def test_pip_self_version_check_calls_underlying_implementation(
     # GIVEN
     mock_session = Mock()
     fake_options = Values({"cache_dir": str(tmpdir)})
+    mocked_function.return_value = None
 
     # WHEN
     self_outdated_check.pip_self_version_check(mock_session, fake_options)