Support pip_args for shared_libs-enabled virtual environments (#1256)

* Implement fix for #964

* fix tests + deprecation notice

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* test debug windows

* propagate un-explained windows fix to test_run

* doc/typing: shared-libs methods require explict pip_args argument

* rename changelog entry

---------

Co-authored-by: Grégoire Roussel <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

changelog.d/1256.removal.md

@@ -0,0 +1 @@
+Deprecate `--skip-maintenance` flag of `pipx list`; maintenance is now never executed there

changelog.d/964.bugfix.md

@@ -0,0 +1 @@
+Pass through `pip` arguments when upgrading shared libraries.

src/pipx/commands/inject.py

@@ -34,6 +34,7 @@ def inject_dep(
         )
 
     venv = Venv(venv_dir, verbose=verbose)
+    venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
 
     if not venv.package_metadata:
         raise PipxError(

src/pipx/commands/install.py

@@ -55,6 +55,7 @@ def install(
             exists = False
 
         venv = Venv(venv_dir, python=python, verbose=verbose)
+        venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
         if exists:
             if not reinstall and force and python_flag_was_passed:
                 print(

src/pipx/commands/list_packages.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from typing import Any, Collection, Dict, Tuple
 
-from pipx import paths, shared_libs
+from pipx import paths
 from pipx.colors import bold
 from pipx.commands.common import VenvProblems, get_venv_summary, venv_health_check
 from pipx.constants import EXIT_CODE_LIST_PROBLEM, EXIT_CODE_OK, ExitCode
@@ -89,19 +89,12 @@ def list_packages(
     include_injected: bool,
     json_format: bool,
     short_format: bool,
-    skip_maintenance: bool,
 ) -> ExitCode:
     """Returns pipx exit code."""
     venv_dirs: Collection[Path] = sorted(venv_container.iter_venv_dirs())
     if not venv_dirs:
         print(f"nothing has been installed with pipx {sleep}", file=sys.stderr)
 
-    if skip_maintenance:
-        shared_libs.shared_libs.skip_upgrade = True
-        logger.info("Skipping shared libs maintenance tasks")
-
-    venv_container.verify_shared_libs()
-
     if json_format:
         all_venv_problems = list_json(venv_dirs)
     elif short_format:

src/pipx/commands/reinstall.py

@@ -4,7 +4,6 @@
 
 from packaging.utils import canonicalize_name
 
-import pipx.shared_libs  # import instead of from so mockable in tests
 from pipx.commands.inject import inject_dep
 from pipx.commands.install import install
 from pipx.commands.uninstall import uninstall
@@ -26,6 +25,7 @@ def reinstall(
     local_man_dir: Path,
     python: str,
     verbose: bool,
+    force_reinstall_shared_libs: bool = False,
 ) -> ExitCode:
     """Returns pipx exit code."""
     if not venv_dir.exists():
@@ -44,6 +44,9 @@ def reinstall(
         return EXIT_CODE_REINSTALL_INVALID_PYTHON
 
     venv = Venv(venv_dir, verbose=verbose)
+    venv.check_upgrade_shared_libs(
+        pip_args=venv.pipx_metadata.main_package.pip_args, verbose=verbose, force_upgrade=force_reinstall_shared_libs
+    )
 
     if venv.pipx_metadata.main_package.package_or_url is not None:
         package_or_url = venv.pipx_metadata.main_package.package_or_url
@@ -104,9 +107,12 @@ def reinstall_all(
     skip: Sequence[str],
 ) -> ExitCode:
     """Returns pipx exit code."""
-    pipx.shared_libs.shared_libs.upgrade(verbose=verbose)
-
     failed: List[str] = []
+
+    # iterate on all packages and reinstall them
+    # for the first one, we also trigger
+    # a reinstall of shared libs beforehand
+    first_reinstall = True
     for venv_dir in venv_container.iter_venv_dirs():
         if venv_dir.name in skip:
             continue
@@ -117,11 +123,13 @@ def reinstall_all(
                 local_man_dir=local_man_dir,
                 python=python,
                 verbose=verbose,
+                force_reinstall_shared_libs=first_reinstall,
             )
         except PipxError as e:
             print(e, file=sys.stderr)
             failed.append(venv_dir.name)
         else:
+            first_reinstall = False
             if package_exit != 0:
                 failed.append(venv_dir.name)
     if len(failed) > 0:

src/pipx/commands/run.py

@@ -97,6 +97,7 @@ def run_script(
             logger.info(f"Reusing cached venv {venv_dir}")
         else:
             venv = Venv(venv_dir, python=python, verbose=verbose)
+            venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
             venv.create_venv(venv_args, pip_args)
             venv.install_unmanaged_packages(requirements, pip_args)
         python_path = venv.python_path
@@ -228,6 +229,7 @@ def _download_and_run(
     verbose: bool,
 ) -> NoReturn:
     venv = Venv(venv_dir, python=python, verbose=verbose)
+    venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
 
     if venv.pipx_metadata.main_package.package is not None:
         package_name = venv.pipx_metadata.main_package.package

src/pipx/commands/upgrade.py

@@ -136,6 +136,7 @@ def _upgrade_venv(
         logger.info("Ignoring --python as not combined with --install")
 
     venv = Venv(venv_dir, verbose=verbose)
+    venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
 
     if not venv.package_metadata:
         raise PipxError(
@@ -207,6 +208,7 @@ def upgrade_all(
     venv_container: VenvContainer,
     verbose: bool,
     *,
+    pip_args: List[str],
     include_injected: bool,
     skip: Sequence[str],
     force: bool,
@@ -216,6 +218,7 @@ def upgrade_all(
     venvs_upgraded = 0
     for venv_dir in venv_container.iter_venv_dirs():
         venv = Venv(venv_dir, verbose=verbose)
+        venv.check_upgrade_shared_libs(pip_args=pip_args, verbose=verbose)
         if venv_dir.name in skip or "--editable" in venv.pipx_metadata.main_package.pip_args:
             continue
         try:

src/pipx/main.py

@@ -283,14 +283,14 @@ def run_pipx_command(args: argparse.Namespace, subparsers: Dict[str, argparse.Ar
             include_injected=args.include_injected,
             skip=skip_list,
             force=args.force,
+            pip_args=pip_args,
         )
     elif args.command == "list":
         return commands.list_packages(
             venv_container,
             args.include_injected,
             args.json,
             args.short,
-            args.skip_maintenance,
         )
     elif args.command == "interpreter":
         if args.interpreter_command == "list":
@@ -610,7 +610,7 @@ def _add_list(subparsers: argparse._SubParsersAction, shared_parser: argparse.Ar
     g = p.add_mutually_exclusive_group()
     g.add_argument("--json", action="store_true", help="Output rich data in json format.")
     g.add_argument("--short", action="store_true", help="List packages only.")
-    g.add_argument("--skip-maintenance", action="store_true", help="Skip maintenance tasks.")
+    g.add_argument("--skip-maintenance", action="store_true", help="(deprecated) No-op")
 
 
 def _add_interpreter(

src/pipx/shared_libs.py

@@ -31,7 +31,6 @@ def __init__(self) -> None:
         self._site_packages: Optional[Path] = None
         self.has_been_updated_this_run = False
         self.has_been_logged_this_run = False
-        self.skip_upgrade = False
 
     @property
     def site_packages(self) -> Path:
@@ -40,7 +39,7 @@ def site_packages(self) -> Path:
 
         return self._site_packages
 
-    def create(self, verbose: bool = False) -> None:
+    def create(self, pip_args: List[str], verbose: bool = False) -> None:
         if not self.is_valid:
             with animate("creating shared libraries", not verbose):
                 create_process = run_subprocess(
@@ -52,7 +51,9 @@ def create(self, verbose: bool = False) -> None:
 
             # ignore installed packages to ensure no unexpected patches from the OS vendor
             # are used
-            self.upgrade(pip_args=["--force-reinstall"], verbose=verbose)
+            pip_args = pip_args or []
+            pip_args.append("--force-reinstall")
+            self.upgrade(pip_args=pip_args, verbose=verbose)
 
     @property
     def is_valid(self) -> bool:
@@ -70,7 +71,7 @@ def is_valid(self) -> bool:
 
     @property
     def needs_upgrade(self) -> bool:
-        if self.has_been_updated_this_run or self.skip_upgrade:
+        if self.has_been_updated_this_run:
             return False
 
         if not self.pip_path.is_file():
@@ -86,9 +87,9 @@ def needs_upgrade(self) -> bool:
             self.has_been_logged_this_run = True
         return time_since_last_update_sec > SHARED_LIBS_MAX_AGE_SEC
 
-    def upgrade(self, *, pip_args: Optional[List[str]] = None, verbose: bool = False) -> None:
+    def upgrade(self, *, pip_args: List[str], verbose: bool = False) -> None:
         if not self.is_valid:
-            self.create(verbose=verbose)
+            self.create(verbose=verbose, pip_args=pip_args)
             return
 
         # Don't try to upgrade multiple times per run

src/pipx/venv.py

@@ -77,10 +77,6 @@ def get_venv_dir(self, package_name: str) -> Path:
         """Return the expected venv path for given `package_name`."""
         return self._root.joinpath(canonicalize_name(package_name))
 
-    def verify_shared_libs(self) -> None:
-        for p in self.iter_venv_dirs():
-            Venv(p)
-
 
 class Venv:
     """Abstraction for a virtual environment with various useful methods for pipx"""
@@ -97,12 +93,21 @@ def __init__(self, path: Path, *, verbose: bool = False, python: str = DEFAULT_P
         except StopIteration:
             self._existing = False
 
+    def check_upgrade_shared_libs(self, verbose: bool, pip_args: List[str], force_upgrade: bool = False):
+        """
+        If necessary, run maintenance tasks to keep the shared libs up-to-date.
+
+        This can trigger `pip install`/`pip install --upgrade` operations,
+        so we expect the caller to provide sensible `pip_args`
+        ( provided by the user in the current CLI call
+        or retrieved from the metadata of a previous installation)
+        """
         if self._existing and self.uses_shared_libs:
             if shared_libs.is_valid:
-                if shared_libs.needs_upgrade:
-                    shared_libs.upgrade(verbose=verbose)
+                if force_upgrade or shared_libs.needs_upgrade:
+                    shared_libs.upgrade(verbose=verbose, pip_args=pip_args)
             else:
-                shared_libs.create(verbose)
+                shared_libs.create(verbose=verbose, pip_args=pip_args)
 
             if not shared_libs.is_valid:
                 raise PipxError(
@@ -161,7 +166,7 @@ def create_venv(self, venv_args: List[str], pip_args: List[str], override_shared
             venv_process = run_subprocess(cmd + venv_args + [str(self.root)], run_dir=str(self.root))
         subprocess_post_check(venv_process)
 
-        shared_libs.create(self.verbose)
+        shared_libs.create(verbose=self.verbose, pip_args=pip_args)
         if not override_shared:
             pipx_pth = get_site_packages(self.python_path) / PIPX_SHARED_PTH
             # write path pointing to the shared libs site-packages directory

tests/test_install.py

@@ -9,7 +9,7 @@
 
 from helpers import app_name, run_pipx_cli, skip_if_windows, unwrap_log_text
 from package_info import PKG
-from pipx import paths
+from pipx import paths, shared_libs
 
 TEST_DATA_PATH = "./testdata/test_package_specifier"
 
@@ -219,6 +219,21 @@ def test_existing_man_page_symlink_points_to_nothing(pipx_temp_env, capsys):
     assert "symlink missing or pointing to unexpected location" not in captured.out
 
 
+def test_pip_args_forwarded_to_shared_libs(pipx_ultra_temp_env, capsys, caplog):
+    # strategy:
+    # 1. start from an empty env to ensure the next command would trigger a shared lib update
+    assert shared_libs.shared_libs.needs_upgrade
+    # 2. install any package with --no-index
+    # and check that the shared library update phase fails
+    return_code = run_pipx_cli(["install", "--verbose", "--pip-args=--no-index", "pycowsay"])
+    assert "Upgrading shared libraries in" in caplog.text
+
+    captured = capsys.readouterr()
+    assert return_code != 0
+    assert "ERROR: Could not find a version that satisfies the requirement pip" in captured.err
+    assert "Failed to upgrade shared libraries" in caplog.text
+
+
 def test_pip_args_forwarded_to_package_name_determination(pipx_temp_env, capsys):
     assert run_pipx_cli(
         [

tests/test_list.py

@@ -175,12 +175,12 @@ def which(name):
     assert "standalone" in captured.out
 
 
-def test_skip_maintenance(pipx_temp_env):
+def test_list_does_not_trigger_maintenance(pipx_temp_env, caplog):
     assert not run_pipx_cli(["install", PKG["pycowsay"]["spec"]])
     assert not run_pipx_cli(["install", PKG["pylint"]["spec"]])
 
     now = time.time()
-    shared_libs.shared_libs.create(verbose=True)
+    shared_libs.shared_libs.create(verbose=True, pip_args=[])
     shared_libs.shared_libs.has_been_updated_this_run = False
 
     access_time = now  # this can be anything
@@ -190,16 +190,17 @@ def test_skip_maintenance(pipx_temp_env):
     )
     assert shared_libs.shared_libs.needs_upgrade
     run_pipx_cli(["list"])
-    assert shared_libs.shared_libs.has_been_updated_this_run
-    assert not shared_libs.shared_libs.needs_upgrade
+    assert not shared_libs.shared_libs.has_been_updated_this_run
+    assert shared_libs.shared_libs.needs_upgrade
 
+    # same test with --skip-maintenance, which is a no-op
+    # we expect the same result, along with a warning
     os.utime(
         shared_libs.shared_libs.pip_path,
         (access_time, -shared_libs.SHARED_LIBS_MAX_AGE_SEC - 5 * 60 + now),
     )
     shared_libs.shared_libs.has_been_updated_this_run = False
     assert shared_libs.shared_libs.needs_upgrade
     run_pipx_cli(["list", "--skip-maintenance"])
-    shared_libs.shared_libs.skip_upgrade = False
     assert not shared_libs.shared_libs.has_been_updated_this_run
     assert shared_libs.shared_libs.needs_upgrade

tests/test_reinstall_all.py

@@ -3,6 +3,7 @@
 import pytest  # type: ignore
 
 from helpers import PIPX_METADATA_LEGACY_VERSIONS, mock_legacy_venv, run_pipx_cli
+from pipx import shared_libs
 
 
 def test_reinstall_all(pipx_temp_env, capsys):
@@ -32,3 +33,13 @@ def test_reinstall_all_suffix_legacy_venv(pipx_temp_env, capsys, metadata_versio
     mock_legacy_venv(f"pycowsay{suffix}", metadata_version=metadata_version)
 
     assert not run_pipx_cli(["reinstall-all", "--python", sys.executable])
+
+
+def test_reinstall_all_triggers_shared_libs_upgrade(pipx_temp_env, caplog, capsys):
+    assert not run_pipx_cli(["install", "pycowsay"])
+
+    shared_libs.shared_libs.has_been_updated_this_run = False
+    caplog.clear()
+
+    assert not run_pipx_cli(["reinstall-all"])
+    assert "Upgrading shared libraries in" in caplog.text

tests/test_run.py

@@ -11,7 +11,7 @@
 import pipx.util
 from helpers import run_pipx_cli
 from package_info import PKG
-from pipx import paths
+from pipx import paths, shared_libs
 
 
 def test_help_text(pipx_temp_env, monkeypatch, capsys):
@@ -299,6 +299,21 @@ def test_run_with_requirements_and_args(caplog, pipx_temp_env, tmp_path):
     assert out.read_text() == "2"
 
 
+def test_pip_args_forwarded_to_shared_libs(pipx_ultra_temp_env, capsys, caplog):
+    # strategy:
+    # 1. start from an empty env to ensure the next command would trigger a shared lib update
+    assert shared_libs.shared_libs.needs_upgrade
+    # 2. install any package with --no-index
+    # and check that the shared library update phase fails
+    return_code = run_pipx_cli(["run", "--verbose", "--pip-args=--no-index", "pycowsay", "hello"])
+    assert "Upgrading shared libraries in" in caplog.text
+
+    captured = capsys.readouterr()
+    assert return_code != 0
+    assert "ERROR: Could not find a version that satisfies the requirement pip" in captured.err
+    assert "Failed to upgrade shared libraries" in caplog.text
+
+
 @mock.patch("os.execvpe", new=execvpe_mock)
 def test_run_with_invalid_requirement(capsys, pipx_temp_env, tmp_path):
     script = tmp_path / "test.py"

tests/test_shared_libs.py

@@ -15,7 +15,7 @@
 )
 def test_auto_update_shared_libs(capsys, pipx_ultra_temp_env, mtime_minus_now, needs_upgrade):
     now = time.time()
-    shared_libs.shared_libs.create(verbose=True)
+    shared_libs.shared_libs.create(verbose=True, pip_args=[])
     shared_libs.shared_libs.has_been_updated_this_run = False
 
     access_time = now  # this can be anything