Switch tests to use custom subprocess module. (#2694)

This works around lack of Windows shebang support since running PEX
files directly is so entrenched in tests.

More work towards #2658.

testing/__init__.py

@@ -10,7 +10,6 @@
 import platform
 import random
 import re
-import subprocess
 import sys
 from collections import Counter
 from contextlib import contextmanager
@@ -34,6 +33,10 @@
 from pex.util import named_temporary_file
 from pex.venv.virtualenv import InstallationChoice, Virtualenv
 
+# Explicitly re-export subprocess to enable a transparent substitution in tests that supports
+# executing PEX files directly on Windows.
+from testing import subprocess as subprocess
+
 try:
     from unittest import mock
 except ImportError:

testing/subprocess.py

@@ -0,0 +1,78 @@
+# Copyright 2025 Pex project contributors.
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import absolute_import
+
+import os
+import subprocess
+import sys
+
+from pex.executables import is_python_script
+from pex.os import WINDOWS
+from pex.pex_info import PexInfo
+from pex.typing import TYPE_CHECKING, cast
+from pex.venv.virtualenv import InvalidVirtualenvError, Virtualenv
+
+if TYPE_CHECKING:
+    from typing import Any, List, Optional, Sequence
+
+
+PIPE = subprocess.PIPE
+STDOUT = subprocess.STDOUT
+CalledProcessError = subprocess.CalledProcessError
+
+
+def _maybe_load_pex_info(path):
+    # type: (str) -> Optional[PexInfo]
+    try:
+        return PexInfo.from_pex(path)
+    except (KeyError, IOError, OSError):
+        return None
+
+
+def _safe_args(args):
+    # type: (Sequence[str]) -> List[str]
+    if WINDOWS:
+        argv0 = args[0]
+        pex_info = _maybe_load_pex_info(argv0)
+        if pex_info and is_python_script(argv0, check_executable=False):
+            try:
+                return [Virtualenv(os.path.dirname(argv0)).interpreter.binary] + list(args)
+            except InvalidVirtualenvError:
+                pass
+        if pex_info or argv0.endswith(".py"):
+            return [sys.executable] + list(args)
+    return args if isinstance(args, list) else list(args)
+
+
+def call(
+    args,  # type: Sequence[str]
+    **kwargs  # type: Any
+):
+    # type: (...) -> int
+    return subprocess.call(args=_safe_args(args), **kwargs)
+
+
+def check_call(
+    args,  # type: Sequence[str]
+    **kwargs  # type: Any
+):
+    # type: (...) -> None
+    subprocess.check_call(args=_safe_args(args), **kwargs)
+
+
+def check_output(
+    args,  # type: Sequence[str]
+    **kwargs  # type: Any
+):
+    # type: (...) -> bytes
+    return cast(bytes, subprocess.check_output(args=_safe_args(args), **kwargs))
+
+
+class Popen(subprocess.Popen):
+    def __init__(
+        self,
+        args,  # type: Sequence[str]
+        **kwargs  # type: Any
+    ):
+        super(Popen, self).__init__(_safe_args(args), **kwargs)  # type: ignore[call-arg]

tests/build_system/test_pep_518.py

@@ -2,7 +2,6 @@
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 import os.path
-import subprocess
 from textwrap import dedent
 
 from pex.build_system import pep_518
@@ -15,6 +14,7 @@
 from pex.typing import TYPE_CHECKING
 from pex.variables import ENV
 from pex.venv.virtualenv import Virtualenv
+from testing import subprocess
 from testing.build_system import hatchling_only_supports_37_and_greater
 
 if TYPE_CHECKING:

tests/integration/build_system/test_pep_518.py

@@ -2,15 +2,14 @@
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 import os.path
-import subprocess
 
 from pex.build_system.pep_518 import BuildSystem, load_build_system
 from pex.pip.version import PipVersion
 from pex.resolve.configured_resolver import ConfiguredResolver
 from pex.resolve.resolver_configuration import PipConfiguration, ReposConfiguration
 from pex.targets import LocalInterpreter
 from pex.typing import TYPE_CHECKING
-from testing import make_env, run_pex_command
+from testing import make_env, run_pex_command, subprocess
 
 if TYPE_CHECKING:
     from typing import Any

tests/integration/cli/commands/test_cache_prune.py

@@ -5,7 +5,6 @@
 
 import os.path
 import shutil
-import subprocess
 import time
 from datetime import datetime, timedelta
 from textwrap import dedent
@@ -33,7 +32,7 @@
 from pex.pip.version import PipVersion, PipVersionValue
 from pex.typing import TYPE_CHECKING
 from pex.variables import ENV
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 from testing.pytest.tmp import Tempdir
 

tests/integration/cli/commands/test_issue_1413.py

@@ -3,7 +3,6 @@
 
 import os
 import shutil
-import subprocess
 
 import pytest
 
@@ -15,7 +14,7 @@
 from pex.resolve.path_mappings import PathMapping, PathMappings
 from pex.resolve.resolved_requirement import ArtifactURL, Pin
 from pex.typing import TYPE_CHECKING
-from testing import make_env, run_pex_command
+from testing import make_env, run_pex_command, subprocess
 from testing.cli import run_pex3
 from testing.pytest.tmp import Tempdir, TempdirFactory
 

tests/integration/cli/commands/test_issue_1667.py

@@ -2,14 +2,13 @@
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 import os.path
-import subprocess
 import sys
 
 import pytest
 
 from pex.interpreter import PythonInterpreter
 from pex.typing import TYPE_CHECKING
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_issue_1688.py

@@ -2,13 +2,12 @@
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 import os
-import subprocess
 
 from pex.interpreter import PythonInterpreter
 from pex.pex_info import PexInfo
 from pex.resolve.lockfile import json_codec
 from pex.typing import TYPE_CHECKING
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_issue_1734.py

@@ -3,12 +3,11 @@
 
 import os
 import re
-import subprocess
 
 from pex.interpreter import PythonInterpreter
 from pex.interpreter_constraints import InterpreterConstraint
 from pex.typing import TYPE_CHECKING
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_issue_2050.py

@@ -3,7 +3,6 @@
 import json
 import os
 import re
-import subprocess
 import sys
 import tempfile
 from textwrap import dedent
@@ -23,7 +22,15 @@
 from pex.targets import LocalInterpreter
 from pex.third_party.packaging.specifiers import SpecifierSet
 from pex.typing import TYPE_CHECKING
-from testing import IS_LINUX, PY310, PY_VER, ensure_python_interpreter, make_env, run_pex_command
+from testing import (
+    IS_LINUX,
+    PY310,
+    PY_VER,
+    ensure_python_interpreter,
+    make_env,
+    run_pex_command,
+    subprocess,
+)
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_issue_2057.py

@@ -3,7 +3,6 @@
 
 import os.path
 import shutil
-import subprocess
 import tempfile
 from textwrap import dedent
 
@@ -12,7 +11,7 @@
 
 from pex.resolve.lockfile import json_codec
 from pex.typing import TYPE_CHECKING
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_local_project_lock.py

@@ -4,15 +4,14 @@
 import os
 import re
 import shutil
-import subprocess
 
 import pytest
 
 from pex.common import touch
 from pex.interpreter import PythonInterpreter
 from pex.typing import TYPE_CHECKING
 from pex.venv.virtualenv import InstallationChoice, Virtualenv
-from testing import PY27, ensure_python_interpreter, run_pex_command
+from testing import PY27, ensure_python_interpreter, run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_lock.py

@@ -5,7 +5,6 @@
 import os
 import re
 import shutil
-import subprocess
 import sys
 from textwrap import dedent
 
@@ -43,6 +42,7 @@
     ensure_python_interpreter,
     make_env,
     run_pex_command,
+    subprocess,
 )
 from testing.build_system import hatchling_only_supports_37_and_greater
 from testing.cli import run_pex3

tests/integration/cli/commands/test_lock_foreign_platform_sdist.py

@@ -4,7 +4,6 @@
 from __future__ import absolute_import
 
 import os.path
-import subprocess
 from textwrap import dedent
 
 import pytest
@@ -15,7 +14,7 @@
 from pex.resolve.configured_resolver import ConfiguredResolver
 from pex.result import try_
 from pex.typing import TYPE_CHECKING
-from testing import PY_VER, data, run_pex_command
+from testing import PY_VER, data, run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_lock_reproducibility_hash_seed.py

@@ -5,14 +5,14 @@
 
 import filecmp
 import os.path
-import subprocess
 from textwrap import dedent
 
 import pytest
 from tools.commands.test_venv import make_env
 
 from pex.common import safe_open
 from pex.typing import TYPE_CHECKING
+from testing import subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_lock_sync.py

@@ -7,7 +7,6 @@
 import os.path
 import re
 import shutil
-import subprocess
 import sys
 from textwrap import dedent
 
@@ -46,6 +45,7 @@
     ensure_python_interpreter,
     make_env,
     re_exact,
+    subprocess,
 )
 from testing.cli import run_pex3
 from testing.find_links import FindLinksRepo

tests/integration/cli/commands/test_lock_update.py

@@ -5,7 +5,6 @@
 import os
 import re
 import shutil
-import subprocess
 
 import pytest
 
@@ -17,7 +16,7 @@
 from pex.resolve.lockfile.model import Lockfile
 from pex.resolve.resolved_requirement import ArtifactURL
 from pex.typing import TYPE_CHECKING
-from testing import make_env, run_pex_command
+from testing import make_env, run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_vcs_lock.py

@@ -4,7 +4,6 @@
 import filecmp
 import os.path
 import shutil
-import subprocess
 import sys
 import tempfile
 from textwrap import dedent
@@ -17,7 +16,7 @@
 from pex.resolve.locked_resolve import VCSArtifact
 from pex.resolve.lockfile import json_codec
 from pex.typing import TYPE_CHECKING
-from testing import run_pex_command
+from testing import run_pex_command, subprocess
 from testing.cli import run_pex3
 
 if TYPE_CHECKING:

tests/integration/cli/commands/test_venv_create.py

@@ -6,7 +6,6 @@
 import glob
 import os.path
 import shutil
-import subprocess
 import sys
 from textwrap import dedent
 
@@ -25,7 +24,15 @@
 from pex.resolve import abbreviated_platforms
 from pex.typing import TYPE_CHECKING
 from pex.venv.virtualenv import Virtualenv
-from testing import IS_MAC, PY39, PY310, ensure_python_interpreter, make_env, run_pex_command
+from testing import (
+    IS_MAC,
+    PY39,
+    PY310,
+    ensure_python_interpreter,
+    make_env,
+    run_pex_command,
+    subprocess,
+)
 from testing.cli import run_pex3
 from testing.pytest.tmp import Tempdir, TempdirFactory