feat: devenv exec

devenv/constants.py

@@ -18,7 +18,7 @@
 shell_path = os.getenv("SHELL", struct_passwd.pw_shell)
 shell = shell_path.rsplit("/", 1)[-1]
 user = struct_passwd.pw_name
-home = struct_passwd.pw_dir
+home = os.getenv("HOME") if CI else struct_passwd.pw_dir
 
 # the *original* user's environment, readonly
 user_environ: typing.Mapping[str, str] = os.environ.copy()

devenv/exec.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import os
+from collections.abc import Sequence
+from typing import Dict
+
+from devenv.lib import proc
+from devenv.lib import venv
+
+help = """Executes a command, using devenv's repo-specific environment.
+Useful if your local environment's broken."""
+
+
+def main(context: Dict[str, str], argv: Sequence[str] | None = None) -> int:
+    if argv is None:
+        print(help)
+        return 1
+
+    reporoot = context["reporoot"]
+
+    env = {**os.environ}
+
+    venv_status = venv.check(reporoot)
+    if (
+        venv_status == venv.VENV_NOT_PRESENT
+        or venv_status == venv.VENV_VERSION_MISMATCH
+    ):
+        print(
+            "WARN: venv doesn't exist or isn't up to date. You should create it with devenv sync.",
+            # unflushed stdout is likely to dissappear due to the imminent exec
+            flush=True,
+        )
+    elif venv_status == venv.VENV_OK:
+        # if there is a good venv, we should use it for the exec.
+        env["VIRTUAL_ENV"] = f"{reporoot}/.venv"
+        env["PATH"] = f"{reporoot}/.venv/bin:{proc.base_path}"
+
+    # note: p variant of exec does in fact take env["PATH"] into
+    #       account for searches (you would think we may need to
+    #       also modify os.environ but it works without it)
+    cmd = argv[0]
+    args = tuple(argv[1:])
+    os.execvpe(cmd, args, env)

devenv/lib/venv.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import configparser
+import os
+
+VENV_OK = 1
+VENV_VERSION_MISMATCH = 2
+VENV_NOT_PRESENT = 3
+VENV_NOT_CONFIGURED = 4
+
+
+def check(reporoot: str) -> int:
+    repo_config = configparser.ConfigParser()
+    repo_config.read(f"{reporoot}/devenv/config.ini")
+
+    if not repo_config.has_section("python"):
+        # the repo doesn't configure venv support
+        # this is mainly here for `devenv exec` which
+        # may or may not be run in a python project
+        return VENV_NOT_CONFIGURED
+
+    if not os.path.exists(f"{reporoot}/.venv/pyvenv.cfg"):
+        return VENV_NOT_PRESENT
+
+    python_version = repo_config["python"]["version"]
+
+    with open(f"{reporoot}/.venv/pyvenv.cfg", "r") as f:
+        for line in f:
+            if line.startswith("version"):
+                venv_version = line.split("=")[1].strip()
+                if venv_version != python_version:
+                    return VENV_VERSION_MISMATCH
+
+    return VENV_OK

devenv/main.py

@@ -11,6 +11,7 @@
 
 from devenv import bootstrap
 from devenv import doctor
+from devenv import exec
 from devenv import pin_gha
 from devenv import sync
 from devenv.constants import CI
@@ -82,12 +83,16 @@ def parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(formatter_class=CustomHelpFormat)
     parser.add_argument(
         "command",
-        choices=("bootstrap", "doctor", "sync", "pin-gha"),
+        choices=("exec", "bootstrap", "doctor", "sync", "pin-gha"),
         metavar="COMMAND",
         help=f"""\
+core commands:
 bootstrap - {bootstrap.help}
 doctor    - {doctor.help}
+exec      - {exec.help}
 sync      - {sync.help}
+
+utilities:
 pin-gha   - {pin_gha.help}
 """,
     )
@@ -102,7 +107,7 @@ def parser() -> argparse.ArgumentParser:
 def devenv(argv: Sequence[str]) -> ExitCode:
     args, remainder = parser().parse_known_args(argv[1:])
 
-    # generic/standalone tools that do not care about devenv configuration
+    # generic/standalone commands that do not care about devenv configuration
     if args.command == "pin-gha":
         return pin_gha.main(remainder)
 
@@ -125,12 +130,14 @@ def devenv(argv: Sequence[str]) -> ExitCode:
         )
         return 1
 
-    # the remaining tools are repo-specific
+    # these commands are run inside a repo
     reporoot = gitroot()
     repo = reporoot.split("/")[-1]
 
     context = {"repo": repo, "reporoot": reporoot}
 
+    if args.command == "exec":
+        return exec.main(context, remainder)
     if args.command == "doctor":
         return doctor.main(context, remainder)
     if args.command == "sync":

tests/test_exec.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import os
+import subprocess
+
+config_template = """
+[devenv]
+coderoot = {coderoot}
+"""
+
+
+def test_python_project(tmp_path: str) -> None:
+    home = tmp_path
+    coderoot = f"{home}/coderoot"
+
+    config_dir = f"{home}/.config/sentry-devenv"
+    os.makedirs(config_dir)
+    with open(f"{config_dir}/config.ini", "w") as f:
+        f.write(config_template.format(coderoot=coderoot))
+
+    reporoot = f"{coderoot}/sentry"
+
+    os.makedirs(f"{reporoot}/.venv/bin")
+    os.symlink("/bin/sh", f"{reporoot}/.venv/bin/venv-executable")
+
+    os.makedirs(f"{reporoot}/devenv")
+    with open(f"{reporoot}/devenv/config.ini", "w") as f:
+        f.write(
+            """
+[python]
+version = 3.10.3
+"""
+        )
+    subprocess.run(("git", "init", "--quiet", "--bare", reporoot))
+
+    # first, let's test an outdated venv
+    with open(f"{reporoot}/.venv/pyvenv.cfg", "w") as f:
+        f.write("version = 3.8.16")
+
+    env = {**os.environ, "HOME": home, "CI": "1"}
+
+    p = subprocess.run(
+        ("devenv", "exec", "command", "-v", "venv-executable"),
+        cwd=reporoot,
+        env=env,
+        capture_output=True,
+    )
+    assert (
+        b"WARN: venv doesn't exist or isn't up to date. You should create it with devenv sync."
+        in p.stdout
+    )
+    # since the venv wasn't in good standing it shouldn't have been
+    # used for the exec
+    assert b"venv-executable: command not found" in p.stderr
+    assert p.returncode != 0
+
+    with open(f"{reporoot}/.venv/pyvenv.cfg", "w") as f:
+        f.write("version = 3.10.3")
+
+    p = subprocess.run(
+        ("devenv", "exec", "venv-executable", "-c", "echo great success"),
+        cwd=reporoot,
+        env=env,
+        capture_output=True,
+    )
+    assert p.stdout == b"great success\n"
+
+    p = subprocess.run(
+        # -- should also work
+        ("devenv", "exec", "--", "venv-executable", "-c", "echo great success"),
+        cwd=reporoot,
+        env=env,
+        capture_output=True,
+    )
+    assert p.stdout == b"great success\n"