Support custom python-executable (#1333)

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"
 
@@ -58,6 +64,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(
@@ -80,6 +90,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,
@@ -111,6 +122,9 @@ def cli(
  log.error("ERROR: " + msg)
  sys.exit(2)
 
+ if python_executable:
+ _validate_python_executable(python_executable)
+
  install_command = cast(InstallCommand, create_command("install"))
  options, _ = install_command.parse_args([])
  session = install_command._build_session(options)
@@ -128,7 +142,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 = (
@@ -152,10 +174,35 @@ def cli(
  dry_run=dry_run,
  install_flags=install_flags,
  ask=ask,
+ python_executable=python_executable,
  )
  )
 
 
+def _validate_python_executable(python_executable: str) -> None:
+ """
+ Validates incoming python_executable argument passed to CLI.
+ """
+ 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, prereleases=True):
+ 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)
+
+
 def _compose_install_flags(
  finder: PackageFinder,
  no_index: bool,

piptools/subprocess_utils.py

@@ -0,0 +1,17 @@
+# 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,
+ 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,41 @@ 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:
+ """
+ Returns pip version specifier requested by current pip-tools installation.
+ """
+ project_dist = get_distribution("pip-tools")
+ requirement = next( # pragma: no branch
+ (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,99 @@ 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.sync.run")
+def test_python_executable_option(
+ run,
+ 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(sys.executable)
+
+ runner.invoke(cli, ["--python-executable", custom_executable])
+
+ assert run.call_count == 2
+
+ call_args = [call[0][0] for call in run.call_args_list]
+ called_uninstall_options = [
+ args[:5] 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"]
+ ]
+ 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,16 @@ 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)
+ assert result, "get_sys_path_for_python_executable should not return empty result"
+ # not testing for equality, because pytest adds extra paths into current sys.path
+ for path in result:
+ assert path in sys.path