Fix loose --venv PEXes to be robust to moves.

Both packed and zipped PEXes undergo an unzip into the PEX_ROOT prior to
any other action, establishing a known private cache of their contents.
Loose PEXes skip this step and run directly from their loose contents.
This is fine and speedy for the "zipapp" execution mode, but leads to an
incorrect venv cache in symlink mode where venv entries symlink back out
to the loose PEX and can thus be invalidated by moves outside Pex
control. Force loose PEXes in `--venv` mode to use copies so that their
cached venv is robust to external moves.

Fixes pex-tool#2023

pex/bin/pex.py

@@ -876,7 +876,11 @@ def do_main(
             compress=options.compress,
         )
         if options.seed != Seed.NONE:
-            seed_info = seed_cache(options, pex, verbose=options.seed == Seed.VERBOSE)
+            seed_info = seed_cache(
+                options,
+                PEX(pex_file, interpreter=interpreter),
+                verbose=options.seed == Seed.VERBOSE,
+            )
             print(seed_info)
     else:
         if not _compatible_with_current_platform(interpreter, targets.platforms):

pex/pex.py

@@ -18,6 +18,7 @@
 from pex.finders import get_entry_point_from_console_script, get_script_from_distributions
 from pex.inherit_path import InheritPath
 from pex.interpreter import PythonInterpreter
+from pex.layout import Layout
 from pex.orderedset import OrderedSet
 from pex.pex_info import PexInfo
 from pex.targets import LocalInterpreter
@@ -162,9 +163,17 @@ def __init__(
         self._vars = env
         self._envs = None  # type: Optional[Iterable[PEXEnvironment]]
         self._activated_dists = None  # type: Optional[Iterable[Distribution]]
+        self._layout = None  # type: Optional[Layout.Value]
         if verify_entry_point:
             self._do_entry_point_verification()
 
+    @property
+    def layout(self):
+        # type: () -> Layout.Value
+        if self._layout is None:
+            self._layout = Layout.identify(self._pex)
+        return self._layout
+
     def pex_info(self, include_env_overrides=True):
         # type: (bool) -> PexInfo
         pex_info = self._pex_info.copy()

pex/pex_bootstrapper.py

@@ -17,6 +17,7 @@
     InterpreterConstraints,
     UnsatisfiableInterpreterConstraintsError,
 )
+from pex.layout import Layout
 from pex.orderedset import OrderedSet
 from pex.pex_info import PexInfo
 from pex.targets import LocalInterpreter
@@ -528,6 +529,13 @@ def ensure_venv(
                         continue
 
                     os.symlink(venv_dir, os.path.join(short_venv.work_dir, "venv"))
+
+                    # Loose PEXes don't need to unpack themselves to the PEX_ROOT before running;
+                    # so we'll not have a stable base there to symlink from. As such, always copy
+                    # for loose PEXes to ensure the PEX_ROOT venv is stable in the face of
+                    # modification of the source loose PEX.
+                    symlink = pex.layout != Layout.LOOSE and not pex_info.venv_site_packages_copies
+
                     shebang = populate_venv(
                         virtualenv,
                         pex,
@@ -536,7 +544,7 @@ def ensure_venv(
                             short_venv_dir, "venv", "bin", os.path.basename(pex.interpreter.binary)
                         ),
                         collisions_ok=collisions_ok,
-                        symlink=not pex_info.venv_site_packages_copies,
+                        symlink=symlink,
                     )
 
                     # There are popular Linux distributions with shebang length limits

tests/integration/test_issue_2023.py

@@ -0,0 +1,83 @@
+# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+import os.path
+import shutil
+import subprocess
+import sys
+from textwrap import dedent
+
+import pytest
+from colors import colors
+
+from pex.layout import Layout
+from pex.testing import run_pex_command
+from pex.typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any, List
+
+
+@pytest.mark.parametrize(
+    "layout", [pytest.param(layout, id=layout.value) for layout in Layout.values()]
+)
+@pytest.mark.parametrize(
+    "execution_mode_args",
+    [
+        pytest.param([], id="UNZIP"),
+        pytest.param(["--venv", "--venv-site-packages-copies"], id="VENV (copies)"),
+        pytest.param(["--venv", "--no-venv-site-packages-copies"], id="VENV (symlinks)"),
+    ],
+)
+def test_unpack_robustness(
+    tmpdir,  # type: Any
+    layout,  # type: Layout.Value
+    execution_mode_args,  # type: List[str]
+):
+    # type: (...) -> None
+    exe = os.path.join(str(tmpdir), "exe.py")
+    with open(exe, "w") as fp:
+        fp.write(
+            dedent(
+                """\
+                import colors
+
+                print(colors.cyan("Wowbagger hasn't gotten to me yet."))
+                """
+            )
+        )
+
+    pex = os.path.join(str(tmpdir), "pex")
+    pex_root = os.path.join(str(tmpdir), "pex_root")
+    run_pex_command(
+        args=[
+            "--runtime-pex-root",
+            pex_root,
+            "ansicolors==1.1.8",
+            "--exe",
+            exe,
+            "--layout",
+            layout.value,
+            "-o",
+            pex,
+        ]
+        + execution_mode_args
+    ).assert_success()
+
+    def assert_pex_works(pex_path):
+        # type: (str) -> None
+        assert (
+            colors.cyan("Wowbagger hasn't gotten to me yet.")
+            == subprocess.check_output(args=[sys.executable, pex_path]).decode("utf-8").strip()
+        )
+
+    assert_pex_works(pex)
+
+    elsewhere = os.path.join(str(tmpdir), "elsewhere")
+    os.mkdir(elsewhere)
+    dest = os.path.join(elsewhere, "other")
+    shutil.move(pex, dest)
+    assert_pex_works(dest)
+
+    shutil.rmtree(pex_root)
+    assert_pex_works(dest)