Add --python-executable option for pip-sync

piptools/scripts/sync.py

@@ -1,6 +1,7 @@
 import itertools
 import os
 import shlex
+import shutil
 import sys
 from typing import List, Optional, Tuple, cast
 
@@ -15,7 +16,12 @@
 from ..exceptions import PipToolsError
 from ..logging import log
 from ..repositories import PyPIRepository
-from ..utils import flat_map
+from ..utils import (
+    flat_map,
+    get_pip_version_for_python_executable,
+    get_required_pip_specification,
+    get_sys_path_for_python_executable,
+)
 
 DEFAULT_REQUIREMENTS_FILE = "requirements.txt"
 
@@ -55,6 +61,10 @@
     is_flag=True,
     help="Ignore package index (only looking at --find-links URLs instead)",
 )
+@click.option(
+    "--python-executable",
+    help="Custom python executable path if targeting an environment other than current",
+)
 @click.option("-v", "--verbose", count=True, help="Show more output")
 @click.option("-q", "--quiet", count=True, help="Give less output")
 @click.option(
@@ -77,6 +87,7 @@ def cli(
     extra_index_url: Tuple[str, ...],
     trusted_host: Tuple[str, ...],
     no_index: bool,
+    python_executable: Optional[str],
     verbose: int,
     quiet: int,
     user_only: bool,
@@ -108,6 +119,26 @@ def cli(
             log.error("ERROR: " + msg)
             sys.exit(2)
 
+    if python_executable:
+        resolved_python_executable = shutil.which(python_executable)
+        if resolved_python_executable is None:
+            msg = "Could not resolve '{}' as valid executable path or alias"
+            log.error(msg.format(python_executable))
+            sys.exit(2)
+
+        # Ensure that target python executable has the right version of pip installed
+        pip_version = get_pip_version_for_python_executable(python_executable)
+        required_pip_specification = get_required_pip_specification()
+        if not required_pip_specification.contains(pip_version):
+            msg = (
+                "Target python executable '{}' has pip version {} installed. "
+                "Version {} is expected."
+            )
+            log.error(
+                msg.format(python_executable, pip_version, required_pip_specification)
+            )
+            sys.exit(2)
+
     install_command = cast(InstallCommand, create_command("install"))
     options, _ = install_command.parse_args([])
     session = install_command._build_session(options)
@@ -125,7 +156,15 @@ def cli(
         log.error(str(e))
         sys.exit(2)
 
-    installed_dists = get_installed_distributions(skip=[], user_only=user_only)
+    paths = (
+        None
+        if python_executable is None
+        else get_sys_path_for_python_executable(python_executable)
+    )
+
+    installed_dists = get_installed_distributions(
+        skip=[], user_only=user_only, paths=paths, local_only=python_executable is None
+    )
     to_install, to_uninstall = sync.diff(merged_requirements, installed_dists)
 
     install_flags = (
@@ -149,6 +188,7 @@ def cli(
             dry_run=dry_run,
             install_flags=install_flags,
             ask=ask,
+            python_executable=python_executable,
         )
     )
 

piptools/subprocess_utils.py

@@ -0,0 +1,18 @@
+# WARNING! BE CAREFUL UPDATING THIS FILE
+# Consider possible security implications associated with subprocess module.
+import subprocess  # nosec
+
+
+def run_python_snippet(python_executable: str, code_to_run: str) -> str:
+    """
+    Executes python code by calling python_executable with '-c' option.
+    """
+    py_exec_cmd = python_executable, "-c", code_to_run
+
+    # subprocess module should never be used with untrusted input
+    return subprocess.check_output(  # nosec
+        py_exec_cmd,
+        shell=False,
+        text=True,
+        universal_newlines=True,
+    )

piptools/sync.py

@@ -178,12 +178,15 @@ def sync(
     dry_run: bool = False,
     install_flags: Optional[List[str]] = None,
     ask: bool = False,
+    python_executable: Optional[str] = None,
 ) -> int:
     """
     Install and uninstalls the given sets of modules.
     """
     exit_code = 0
 
+    python_executable = python_executable or sys.executable
+
     if not to_uninstall and not to_install:
         log.info("Everything up-to-date", err=False)
         return exit_code
@@ -216,7 +219,7 @@ def sync(
         if to_uninstall:
             run(  # nosec
                 [
-                    sys.executable,
+                    python_executable,
                     "-m",
                     "pip",
                     "uninstall",
@@ -244,7 +247,7 @@ def sync(
             try:
                 run(  # nosec
                     [
-                        sys.executable,
+                        python_executable,
                         "-m",
                         "pip",
                         "install",

piptools/utils.py

@@ -1,5 +1,7 @@
 import collections
 import itertools
+import json
+import os
 import shlex
 from typing import (
     Callable,
@@ -23,6 +25,9 @@
 from pip._vendor.packaging.markers import Marker
 from pip._vendor.packaging.specifiers import SpecifierSet
 from pip._vendor.packaging.version import Version
+from pip._vendor.pkg_resources import get_distribution
+
+from piptools.subprocess_utils import run_python_snippet
 
 _KT = TypeVar("_KT")
 _VT = TypeVar("_VT")
@@ -358,3 +363,36 @@ def get_compile_command(click_ctx: click.Context) -> str:
                     left_args.append(f"{option_long_name}={shlex.quote(str(val))}")
 
     return " ".join(["pip-compile", *sorted(left_args), *sorted(right_args)])
+
+
+def get_required_pip_specification() -> SpecifierSet:
+    project_dist = get_distribution("pip-tools")
+    requirement = next((r for r in project_dist.requires() if r.name == "pip"), None)
+    assert (
+        requirement is not None
+    ), "'pip' is expected to be in the list of pip-tools requirements"
+    return requirement.specifier
+
+
+def get_pip_version_for_python_executable(python_executable: str) -> Version:
+    """
+    Returns pip version for the given python executable.
+    """
+    str_version = run_python_snippet(
+        python_executable, "import pip;print(pip.__version__)"
+    )
+    return Version(str_version)
+
+
+def get_sys_path_for_python_executable(python_executable: str) -> List[str]:
+    """
+    Returns sys.path list for the given python executable.
+    """
+    result = run_python_snippet(
+        python_executable, "import sys;import json;print(json.dumps(sys.path))"
+    )
+
+    paths = json.loads(result)
+    assert isinstance(paths, list)
+    assert all(isinstance(i, str) for i in paths)
+    return [os.path.abspath(path) for path in paths]

tests/test_cli_sync.py

@@ -1,8 +1,10 @@
+import os
 import subprocess
 import sys
 from unittest import mock
 
 import pytest
+from pip._vendor.packaging.version import Version
 
 from piptools.scripts.sync import DEFAULT_REQUIREMENTS_FILE, cli
 
@@ -242,3 +244,116 @@ def test_sync_dry_run_returns_non_zero_exit_code(runner):
     out = runner.invoke(cli, ["--dry-run"])
 
     assert out.exit_code == 1
+
+
+@mock.patch("piptools.scripts.sync.get_pip_version_for_python_executable")
+@mock.patch("piptools.scripts.sync.get_sys_path_for_python_executable")
+@mock.patch("piptools.scripts.sync.get_installed_distributions")
+@mock.patch("piptools.sync.run")
+def test_python_executable_option(
+    run,
+    get_installed_distributions,
+    get_sys_path_for_python_executable,
+    get_pip_version_for_python_executable,
+    runner,
+    fake_dist,
+):
+    """
+    Make sure sync command can run with `--python-executable` option
+    """
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    custom_executable = os.path.abspath("custom_executable")
+    with open(custom_executable, "w") as exec_file:
+        exec_file.write("")
+
+    os.chmod(custom_executable, 0o700)
+
+    sys_paths = ["", "./"]
+    get_sys_path_for_python_executable.return_value = sys_paths
+    get_installed_distributions.return_value = [fake_dist("django==1.8")]
+    get_pip_version_for_python_executable.return_value = Version("20.3")
+
+    runner.invoke(cli, ["--python-executable", custom_executable])
+
+    get_installed_distributions.assert_called_once_with(
+        skip=[], user_only=False, paths=sys_paths, local_only=False
+    )
+
+    assert run.call_count == 2
+
+    call_args = [call[0][0] for call in run.call_args_list]
+    called_uninstall_options = [args for args in call_args if args[3] == "uninstall"]
+    called_install_options = [args[:-1] for args in call_args if args[3] == "install"]
+
+    assert called_uninstall_options == [
+        [custom_executable, "-m", "pip", "uninstall", "-y", "django"]
+    ]
+    assert called_install_options == [[custom_executable, "-m", "pip", "install", "-r"]]
+
+
+@pytest.mark.parametrize(
+    "python_executable",
+    (
+        "/tmp/invalid_executable",
+        "invalid_python",
+    ),
+)
+def test_invalid_python_executable(runner, python_executable):
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    out = runner.invoke(cli, ["--python-executable", python_executable])
+    assert out.exit_code == 2, out
+    message = "Could not resolve '{}' as valid executable path or alias\n"
+    assert out.stderr == message.format(python_executable)
+
+
+@mock.patch("piptools.scripts.sync.get_pip_version_for_python_executable")
+def test_invalid_pip_version_in_python_executable(
+    get_pip_version_for_python_executable, runner
+):
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    custom_executable = os.path.abspath("custom_executable")
+    with open(custom_executable, "w") as exec_file:
+        exec_file.write("")
+
+    os.chmod(custom_executable, 0o700)
+
+    get_pip_version_for_python_executable.return_value = Version("19.1")
+
+    out = runner.invoke(cli, ["--python-executable", custom_executable])
+    assert out.exit_code == 2, out
+    message = (
+        "Target python executable '{}' has pip version 19.1 installed. "
+        "Version"  # ">=20.3 is expected.\n" part is omitted
+    )
+    assert out.stderr.startswith(message.format(custom_executable))
+
+
+@mock.patch("piptools.sync.run")
+def test_default_python_executable_option(run, runner):
+    """
+    Make sure sys.executable is used when --python-executable is not provided
+    """
+    with open("requirements.txt", "w") as req_in:
+        req_in.write("small-fake-a==1.10.0")
+
+    runner.invoke(cli)
+
+    assert run.call_count == 2
+
+    call_args = [call[0][0] for call in run.call_args_list]
+    called_install_options = [args[:-1] for args in call_args if args[3] == "install"]
+    assert called_install_options == [
+        [
+            sys.executable,
+            "-m",
+            "pip",
+            "install",
+            "-r",
+        ]
+    ]

tests/test_subprocess_utils.py

@@ -0,0 +1,8 @@
+import sys
+
+from piptools.subprocess_utils import run_python_snippet
+
+
+def test_run_python_snippet_returns_multilne():
+    result = run_python_snippet(sys.executable, r'print("MULTILINE\nOUTPUT", end="")')
+    assert result == "MULTILINE\nOUTPUT"

tests/test_utils.py

@@ -2,8 +2,11 @@
 import operator
 import os
 import shlex
+import sys
 
+import pip
 import pytest
+from pip._vendor.packaging.version import Version
 
 from piptools.scripts.compile import cli as compile_cli
 from piptools.utils import (
@@ -15,6 +18,8 @@
     format_specifier,
     get_compile_command,
     get_hashes_from_ireq,
+    get_pip_version_for_python_executable,
+    get_sys_path_for_python_executable,
     is_pinned_requirement,
     is_url_requirement,
     lookup_table,
@@ -423,3 +428,15 @@ def test_drop_extras(from_line, given, expected):
         assert ireq.markers is None
     else:
         assert str(ireq.markers).replace("'", '"') == expected.replace("'", '"')
+
+
+def test_get_pip_version_for_python_executable():
+    result = get_pip_version_for_python_executable(sys.executable)
+    assert Version(pip.__version__) == result
+
+
+def test_get_sys_path_for_python_executable():
+    result = get_sys_path_for_python_executable(sys.executable)
+    # not testing for equality, because pytest adds extra paths into current sys.path
+    for path in result:
+        assert path in sys.path