fix(tenv): reinstall when a shim is missing

devenv/lib/tenv.py

@@ -8,6 +8,21 @@
 from devenv.lib import fs
 from devenv.lib import proc
 
+# binaries shipped inside the tenv archive, extracted into TENV_ROOT/bin
+BINARIES = ("tenv", "tofu", "terraform", "terragrunt", "tf")
+
+# the subset we put on PATH as shims. `tf` is left out because repos ship
+# their own `tf` wrapper and our shim would shadow it.
+SHIMS = tuple(binary for binary in BINARIES if binary != "tf")
+
+# This makes sure we're executing with our custom TENV_ROOT, otherwise
+# there's potential for collision with ~/.tenv. This isolation also makes
+# uninstallation safe.
+SHIM = """#!/bin/sh
+export TENV_ROOT={TENV_ROOT}
+exec {TENV_ROOT}/bin/{binary} "$@"
+"""
+
 
 def _install(url: str, sha256: str, into: str) -> None:
     TENV_ROOT = f"{into}/tenv-root"
@@ -19,73 +34,53 @@ def _install(url: str, sha256: str, into: str) -> None:
 
         # the archive was atomically placed into tmpd so
         # these are on the same fs and can be atomically moved too
-        os.replace(f"{tmpd}/terraform", f"{TENV_ROOT}/bin/terraform")
-        os.replace(f"{tmpd}/tf", f"{TENV_ROOT}/bin/tf")
-        os.replace(f"{tmpd}/tofu", f"{TENV_ROOT}/bin/tofu")
-        os.replace(f"{tmpd}/terragrunt", f"{TENV_ROOT}/bin/terragrunt")
-        os.replace(f"{tmpd}/tenv", f"{TENV_ROOT}/bin/tenv")
-
-        # those all need to go inside a bin instead of like, TENV_ROOT/terraform
-        # because tenv wants to mkdir that
-
-    # These shims make sure we're executing with our custom TENV_ROOT,
-    # otherwise there's potential for collision with ~/.tenv.
-    # This isolation also makes uninstallation safe.
-    fs.write_script(
-        f"{into}/tenv",
-        """#!/bin/sh
-export TENV_ROOT={TENV_ROOT}
-exec {TENV_ROOT}/bin/tenv "$@"
-""",
-        shell_escape={"TENV_ROOT": TENV_ROOT},
-    )
-    fs.write_script(
-        f"{into}/terraform",
-        """#!/bin/sh
-export TENV_ROOT={TENV_ROOT}
-exec {TENV_ROOT}/bin/terraform "$@"
-""",
-        shell_escape={"TENV_ROOT": TENV_ROOT},
-    )
-    fs.write_script(
-        f"{into}/tofu",
-        """#!/bin/sh
-export TENV_ROOT={TENV_ROOT}
-exec {TENV_ROOT}/bin/tofu "$@"
-""",
-        shell_escape={"TENV_ROOT": TENV_ROOT},
-    )
-    fs.write_script(
-        f"{into}/terragrunt",
-        """#!/bin/sh
-export TENV_ROOT={TENV_ROOT}
-exec {TENV_ROOT}/bin/terragrunt "$@"
-""",
-        shell_escape={"TENV_ROOT": TENV_ROOT},
-    )
+        #
+        # those all need to go inside a bin instead of like,
+        # TENV_ROOT/terraform because tenv wants to mkdir that
+        for binary in BINARIES:
+            os.replace(f"{tmpd}/{binary}", f"{TENV_ROOT}/bin/{binary}")
+
+    for binary in SHIMS:
+        fs.write_script(
+            f"{into}/{binary}",
+            SHIM,
+            shell_escape={"TENV_ROOT": TENV_ROOT, "binary": binary},
+        )
 
 
 def uninstall(binroot: str) -> None:
-    for d in (f"{binroot}/tenv-root",):
-        shutil.rmtree(d, ignore_errors=True)
-
-    for fp in (
-        f"{binroot}/tenv",
-        f"{binroot}/terraform",
-        f"{binroot}/tofu",
-        f"{binroot}/terragrunt",
-    ):
+    # the shims go first: interrupted here, the next sync sees no tenv on
+    # PATH and reinstalls cleanly. the other order strands shims pointing
+    # at a tenv-root that's already gone.
+    for binary in SHIMS:
         try:
-            os.remove(fp)
+            os.remove(f"{binroot}/{binary}")
         except FileNotFoundError:
             # it's better to do this than to guard with
             # os.path.exists(fp) because if it's an invalid or circular
             # symlink the result'll be False!
             pass
 
+    shutil.rmtree(f"{binroot}/tenv-root", ignore_errors=True)
+
+
+def _missing_shims(binroot: str) -> tuple[str, ...]:
+    return tuple(
+        binary
+        for binary in SHIMS
+        if not os.path.exists(f"{binroot}/{binary}")
+        or not os.path.exists(f"{binroot}/tenv-root/bin/{binary}")
+    )
+
 
 def _version(binpath: str) -> str:
-    stdout = proc.run((binpath, "version"), stdout=True)
+    try:
+        stdout = proc.run((binpath, "version"), stdout=True)
+    except RuntimeError:
+        # a shim whose TENV_ROOT is gone exits 126, and a truncated or
+        # wrong-arch binary won't run either. either way it's reinstallable,
+        # so don't take the whole sync down with us.
+        return ""
     # tenv version v1.3.0
     return stdout.split()[-1]
 
@@ -95,15 +90,19 @@ def install(version: str, url: str, sha256: str, reporoot: str) -> None:
     binpath = f"{binroot}/tenv"
 
     if shutil.which("tenv", path=binroot) == binpath:
-        installed_version = _version(binpath)
-        if version == installed_version:
-            return
-        print(f"installed tenv {installed_version} is unexpected!")
+        missing = _missing_shims(binroot)
+        if missing:
+            print(f"tenv {version} is missing {', '.join(missing)}...")
+        else:
+            installed_version = _version(binpath)
+            if version == installed_version:
+                return
+            print(f"installed tenv {installed_version} is unexpected!")
 
     print(f"installing tenv {version}...")
     uninstall(binroot)
     _install(url, sha256, binroot)
 
     installed_version = _version(binpath)
     if version != installed_version:
-        raise SystemExit("Failed to install tenv {version}!")
+        raise SystemExit(f"Failed to install tenv {version}!")

tests/lib/test_tenv.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import os
+import pathlib
+import shutil
+from unittest.mock import patch
+
+from devenv.lib import tenv
+
+
+def _unpack(_archive_file: str, tmpd: str) -> None:
+    for binary in tenv.BINARIES:
+        open(f"{tmpd}/{binary}", "w").close()
+
+
+def _install(reporoot: str, version: str = "v0.0.0") -> None:
+    with (
+        patch("devenv.lib.archive.download"),
+        patch("devenv.lib.archive.unpack", side_effect=_unpack),
+        patch(
+            "devenv.lib.tenv.proc.run",
+            # once to check what's installed, once to verify
+            side_effect=[f"tenv version {version}"] * 2,
+        ),
+    ):
+        tenv.install(version, "url", "sha256", reporoot)
+
+
+def test_install(tmp_path: pathlib.Path) -> None:
+    reporoot = f"{tmp_path}/test"
+    binroot = f"{reporoot}/.devenv/bin"
+
+    _install(reporoot)
+
+    # terragrunt shells out to `tofu`, so it has to be on PATH too
+    for binary in tenv.SHIMS:
+        assert os.path.exists(f"{binroot}/tenv-root/bin/{binary}"), binary
+        with open(f"{binroot}/{binary}", "r") as f:
+            assert (
+                f.read()
+                == f"""#!/bin/sh
+export TENV_ROOT={binroot}/tenv-root
+exec {binroot}/tenv-root/bin/{binary} "$@"
+"""
+            )
+
+    # tf is extracted but deliberately not shimmed onto PATH
+    assert os.path.exists(f"{binroot}/tenv-root/bin/tf")
+    assert not os.path.exists(f"{binroot}/tf")
+
+
+def test_install_is_a_noop_when_complete(tmp_path: pathlib.Path) -> None:
+    reporoot = f"{tmp_path}/test"
+
+    _install(reporoot)
+
+    with (
+        patch("devenv.lib.archive.download") as mock_download,
+        patch(
+            "devenv.lib.tenv.proc.run",
+            side_effect=["tenv version v0.0.0"],  # tenv version
+        ),
+    ):
+        tenv.install("v0.0.0", "url", "sha256", reporoot)
+
+    assert mock_download.mock_calls == []
+
+
+def test_install_replaces_a_missing_shim(tmp_path: pathlib.Path) -> None:
+    reporoot = f"{tmp_path}/test"
+    binroot = f"{reporoot}/.devenv/bin"
+
+    _install(reporoot)
+
+    # a checkout installed by a devenv that predates the tofu shim: the
+    # version matches, so this has to reinstall anyway or it stays broken
+    os.remove(f"{binroot}/tofu")
+    os.remove(f"{binroot}/tenv-root/bin/tofu")
+
+    _install(reporoot)
+
+    assert os.path.exists(f"{binroot}/tofu")
+    assert os.path.exists(f"{binroot}/tenv-root/bin/tofu")
+
+
+def test_install_recovers_from_a_missing_tenv_root(
+    tmp_path: pathlib.Path,
+) -> None:
+    reporoot = f"{tmp_path}/test"
+    binroot = f"{reporoot}/.devenv/bin"
+
+    _install(reporoot)
+
+    # an interrupted sync or a moved checkout leaves the shims on PATH
+    # pointing at a tenv-root that isn't there, so probing the version
+    # exits 126 - that has to reinstall, not take the whole sync down
+    shutil.rmtree(f"{binroot}/tenv-root")
+
+    def run(cmd: tuple[str, ...], stdout: bool = False) -> str:
+        # the shim only runs if its TENV_ROOT survived
+        if not os.path.exists(f"{binroot}/tenv-root/bin/tenv"):
+            raise RuntimeError(f"Command `{cmd[0]} version` failed! (code 126)")
+        return "tenv version v0.0.0"
+
+    with (
+        patch("devenv.lib.archive.download"),
+        patch("devenv.lib.archive.unpack", side_effect=_unpack),
+        patch("devenv.lib.tenv.proc.run", side_effect=run),
+    ):
+        tenv.install("v0.0.0", "url", "sha256", reporoot)
+
+    assert tenv._missing_shims(binroot) == ()
+
+
+def test_version_of_an_unrunnable_tenv_is_empty() -> None:
+    with patch(
+        "devenv.lib.tenv.proc.run",
+        side_effect=RuntimeError("failed! (code 126)"),
+    ):
+        assert tenv._version(f"{os.sep}nonexistent{os.sep}tenv") == ""
+
+
+def test_uninstall_removes_shims_before_tenv_root(
+    tmp_path: pathlib.Path,
+) -> None:
+    binroot = f"{tmp_path}/bin"
+    os.makedirs(f"{binroot}/tenv-root/bin")
+    for binary in tenv.SHIMS:
+        open(f"{binroot}/{binary}", "w").close()
+
+    surviving_shims: list[list[str]] = []
+    real_rmtree = shutil.rmtree
+
+    def spy(path: str, ignore_errors: bool = False) -> None:
+        surviving_shims.append(
+            [b for b in tenv.SHIMS if os.path.exists(f"{binroot}/{b}")]
+        )
+        real_rmtree(path, ignore_errors=ignore_errors)
+
+    with patch("devenv.lib.tenv.shutil.rmtree", side_effect=spy):
+        tenv.uninstall(binroot)
+
+    # interrupted between the two steps, we'd rather have no tenv on PATH
+    # than shims pointing at a tenv-root that's already gone
+    assert surviving_shims == [[]]
+
+
+def test_uninstall_removes_every_shim(tmp_path: pathlib.Path) -> None:
+    binroot = f"{tmp_path}/bin"
+    os.makedirs(f"{binroot}/tenv-root/bin")
+    for binary in tenv.SHIMS:
+        open(f"{binroot}/{binary}", "w").close()
+
+    tenv.uninstall(binroot)
+
+    assert not os.path.exists(f"{binroot}/tenv-root")
+    assert os.listdir(binroot) == []