diff --git a/src/pip/_internal/cli/base_command.py b/src/pip/_internal/cli/base_command.py index b4f37394dc1..bc1ab65949d 100644 --- a/src/pip/_internal/cli/base_command.py +++ b/src/pip/_internal/cli/base_command.py @@ -1,6 +1,5 @@ """Base Command class, and related routines""" -import functools import logging import logging.config import optparse @@ -8,7 +7,7 @@ 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) diff --git a/src/pip/_internal/cli/index_command.py b/src/pip/_internal/cli/index_command.py index 9991326f36b..226f8da1e94 100644 --- a/src/pip/_internal/cli/index_command.py +++ b/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) diff --git a/src/pip/_internal/self_outdated_check.py b/src/pip/_internal/self_outdated_check.py index 2185f2fb108..f9a91af9d84 100644 --- a/src/pip/_internal/self_outdated_check.py +++ b/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}) diff --git a/tests/unit/test_self_check_outdated.py b/tests/unit/test_self_check_outdated.py index 6b6fcea55a9..99a556e8ac7 100644 --- a/tests/unit/test_self_check_outdated.py +++ b/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)