Add missing repo automation

.envrc

@@ -0,0 +1,7 @@
+layout_saltext() {
+    VIRTUAL_ENV="$(python3 tools/initialize.py --print-venv)"
+    PATH_add "$VIRTUAL_ENV/bin"
+    export VIRTUAL_ENV
+}
+
+layout_saltext

tools/helpers/cmd.py

@@ -0,0 +1,286 @@
+"""
+Polyfill for very basic ``plumbum`` functionality, no external libs required.
+Makes scripts that call a lot of CLI commands much more pleasant to write.
+"""
+
+import os
+import platform
+import shlex
+import shutil
+import subprocess
+from contextlib import contextmanager
+from dataclasses import dataclass
+from dataclasses import field
+from pathlib import Path
+
+
+class CommandNotFound(RuntimeError):
+    """
+    Raised when a command cannot be found in $PATH
+    """
+
+
+@dataclass(frozen=True)
+class ProcessResult:
+    """
+    The full process result, returned by ``.run`` methods.
+    The ``__call__`` ones just return stdout.
+    """
+
+    retcode: int
+    stdout: str | bytes
+    stderr: str | bytes
+    argv: tuple
+
+    def check(self, retcode=None):
+        """
+        Check if the retcode is expected. retcode can be a list.
+        """
+        if retcode is None:
+            expected = [0]
+        elif not isinstance(retcode, (list, tuple)):
+            expected = [retcode]
+        if self.retcode not in expected:
+            raise ProcessExecutionError(self.argv, self.retcode, self.stdout, self.stderr)
+
+    def __str__(self):
+        msg = [
+            "Process execution result:",
+            f"Command: {shlex.join(self.argv)}",
+            f"Retcode: {self.retcode}",
+            "Stdout:   |",
+        ]
+        msg += [" " * 10 + "| " + line for line in str(self.stdout).splitlines()]
+        msg.append("Stderr:   |")
+        msg += [" " * 10 + "| " + line for line in str(self.stderr).splitlines()]
+        return "\n".join(msg)
+
+
+class ProcessExecutionError(OSError):
+    """
+    Raised by ProcessResult.check when an unexpected retcode was returned.
+    """
+
+    def __init__(self, argv, retcode, stdout, stderr):
+        self.argv = argv
+        self.retcode = retcode
+        if isinstance(stdout, bytes):
+            stdout = ascii(stdout)
+        if isinstance(stderr, bytes):
+            stderr = ascii(stderr)
+        self.stdout = stdout
+        self.stderr = stderr
+
+    def __str__(self):
+        msg = [
+            "Process finished with unexpected exit code",
+            f"Retcode: {self.retcode}",
+            f"Command: {shlex.join(self.argv)}",
+            "Stdout:   |",
+        ]
+        msg += [" " * 10 + "| " + line for line in str(self.stdout).splitlines()]
+        msg.append("Stderr:   |")
+        msg += [" " * 10 + "| " + line for line in str(self.stderr).splitlines()]
+        return "\n".join(msg)
+
+
+class Local:
+    """
+    Glue for command environment defaults.
+    Should be treated as a singleton.
+
+    Example:
+
+        local = Local()
+
+        some_cmd = local["some_cmd"]
+        with local.cwd(some_path), local.env(FOO="bar"):
+            some_cmd("baz")
+
+        # A changed $PATH requires to rediscover commands.
+        with local.prepend_path(important_path):
+            local["other_cmd"]()
+        with local.venv(venv_path):
+            local["python"]("-m", "pip", "install", "salt")
+
+    """
+
+    def __init__(self):
+        # Explicitly cast values to strings to avoid problems on Windows
+        self._env = {k: str(v) for k, v in os.environ.items()}
+
+    def __getitem__(self, exe):
+        """
+        Return a LocalCommand in this context.
+        """
+        return LocalCommand(exe, _local=self)
+
+    @property
+    def path(self):
+        """
+        List of paths in the context's $PATH.
+        """
+        return self._env.get("PATH", "").split(os.pathsep)
+
+    @contextmanager
+    def cwd(self, path):
+        """
+        Set the default current working directory for commands inside this context.
+        """
+        prev = Path(os.getcwd())
+        new = prev / path
+        os.cwd(new)
+        try:
+            yield
+        finally:
+            os.cwd(prev)
+
+    @contextmanager
+    def env(self, **kwargs):
+        """
+        Override default env vars (sourced from the current process' environment)
+        for commands inside this context.
+        """
+        prev = self._env.copy()
+        self._env.update((k, str(v)) for k, v in kwargs.items())
+        try:
+            yield
+        finally:
+            self._env = prev
+
+    @contextmanager
+    def path_prepend(self, *args):
+        """
+        Prepend paths to $PATH for commands inside this context.
+
+        Note: If you have saved a reference to an already requested command,
+        its $PATH will be updated, but it might not be the command
+        that would have been returned by a new request.
+        """
+        new_path = [str(arg) for arg in args] + self.path
+        with self.env(PATH=os.pathsep.join(new_path)):
+            yield
+
+    @contextmanager
+    def venv(self, venv_dir):
+        """
+        Enter a Python virtual environment. Effectively prepends its bin dir
+        to $PATH and sets ``VIRTUAL_ENV``.
+        """
+        venv_dir = Path(venv_dir)
+        if not venv_dir.is_dir() or not (venv_dir / "pyvenv.cfg").exists():
+            raise ValueError(f"Not a virtual environment: {venv_dir}")
+        venv_bin_dir = venv_dir / "bin"
+        if platform.system() == "Windows":
+            venv_bin_dir = venv_dir / "Scripts"
+        with self.path_prepend(venv_bin_dir), self.env(VIRTUAL_ENV=str(venv_dir)):
+            yield
+
+
+@dataclass(frozen=True)
+class Executable:
+    """
+    Utility class used to avoid repeated command lookups.
+    """
+
+    _exe: str
+
+    def __str__(self):
+        return self._exe
+
+    def __repr__(self):
+        return f"Executable <{self._exe}>"
+
+
+@dataclass(frozen=True)
+class Command:
+    """
+    A command object, can be instantiated directly. Does not follow ``Local``.
+    """
+
+    exe: Executable | str
+    args: tuple[str, ...] = ()
+
+    def __post_init__(self):
+        if not isinstance(self.exe, Executable):
+            if not (full_exe := self._which(self.exe)):
+                raise CommandNotFound(self.exe)
+            object.__setattr__(self, "exe", Executable(full_exe))
+
+    def _which(self, exe):
+        return shutil.which(exe)
+
+    def _get_env(self, overrides=None):
+        base = {k: str(v) for k, v in os.environ.items()}
+        base.update(overrides or {})
+        return base
+
+    def __getitem__(self, arg_or_args):
+        """
+        Returns a subcommand with bound parameters.
+
+        Example:
+
+            git = Command("git")["-c", "commit.gpgsign=0"]
+            # ...
+            git("add", ".")
+            git("commit", "-m", "testcommit")
+
+        """
+        if not isinstance(arg_or_args, tuple):
+            arg_or_args = (arg_or_args,)
+        return type(self)(self.exe, tuple(*self.args, *arg_or_args), _local=self._local)
+
+    def __call__(self, *args, **kwargs):
+        """
+        Run this command and return stdout.
+        """
+        return self.run(*args, **kwargs).stdout
+
+    def __str__(self):
+        return shlex.join([self.exe] + list(self.args))
+
+    def __repr__(self):
+        return f"Command<{self.exe}, {self.args!r}>"
+
+    def run(self, *args, check=True, env=None, **kwargs):
+        """
+        Run this command and return the full output.
+        """
+        kwargs.setdefault("stdout", subprocess.PIPE)
+        kwargs.setdefault("stderr", subprocess.PIPE)
+        kwargs.setdefault("text", True)
+        argv = [str(self.exe), *self.args, *args]
+        proc = subprocess.run(argv, check=False, env=self._get_env(env), **kwargs)
+        ret = ProcessResult(
+            retcode=proc.returncode,
+            stdout=proc.stdout,
+            stderr=proc.stderr,
+            argv=argv,
+        )
+        if check:
+            ret.check()
+        return ret
+
+
+@dataclass(frozen=True)
+class LocalCommand(Command):
+    """
+    Command returned by Local()["some_command"]. Follows local contexts.
+    """
+
+    _local: Local = field(kw_only=True, repr=False)
+
+    def _which(self, exe):
+        return shutil.which(exe, path=self._local._env.get("PATH", ""))
+
+    def _get_env(self, overrides=None):
+        base = self._local._env.copy()
+        base.update(overrides or {})
+        return base
+
+
+# Should be imported from here.
+local = Local()
+# We must assume git is installed
+git = local["git"]

tools/helpers/copier.py

@@ -0,0 +1,68 @@
+import sys
+from functools import wraps
+from pathlib import Path
+
+from . import prompt
+
+try:
+    # In case we have it, use it.
+    # It's always installed in the Copier environment, so if you ensure you
+    # call this via ``copier_python``, this will work.
+    import yaml
+except ImportError:
+    yaml = None
+
+
+COPIER_ANSWERS = Path(".copier-answers.yml").resolve()
+
+
+def _needs_answers(func):
+    @wraps(func)
+    def _wrapper(*args, **kwargs):
+        if not COPIER_ANSWERS.exists():
+            raise RuntimeError(f"Missing answers file at {COPIER_ANSWERS}")
+        return func(*args, **kwargs)
+
+    return _wrapper
+
+
+@_needs_answers
+def load_answers():
+    """
+    Load the complete answers file. Depends on PyYAML.
+    """
+    if not yaml:
+        raise RuntimeError("Missing pyyaml in environment")
+    with open(COPIER_ANSWERS) as f:
+        return yaml.safe_load(f)
+
+
+@_needs_answers
+def discover_project_name():
+    """
+    Specifically discover project name. No dependency.
+    """
+    for line in COPIER_ANSWERS.read_text().splitlines():
+        if line.startswith("project_name"):
+            return line.split(":", maxsplit=1)[1].strip()
+    raise RuntimeError("Failed discovering project name")
+
+
+def finish_task(msg, success, err_exit=False, extra=None):
+    """
+    Print final conclusion of task (migration) run in Copier.
+
+    We usually want to exit with 0, even when something fails,
+    because a failing task/migration should not crash Copier.
+    """
+    print("\n", file=sys.stderr)
+    if success:
+        prompt.pprint(f"\n    ✓ {msg}", bold=True, bg=prompt.DARKGREEN, stream=sys.stderr)
+    elif success is None:
+        prompt.pprint(
+            f"\n    ✓ {msg}", bold=True, fg=prompt.YELLOW, bg=prompt.DARKGREEN, stream=sys.stderr
+        )
+        success = True
+    else:
+        prompt.warn(f"    ✗ {msg}", extra)
+    raise SystemExit(int(not success and err_exit))

tools/helpers/git.py

@@ -0,0 +1,30 @@
+from pathlib import Path
+
+from .cmd import git
+
+
+def ensure_git():
+    """
+    Ensure the repository has been initialized.
+    """
+    if Path(".git").is_dir():
+        return
+    git("init", "--initial-branch", "main")
+
+
+def list_untracked():
+    """
+    List untracked files.
+    """
+    for path in git("ls-files", "-z", "-o", "--exclude-standard").split("\x00"):
+        if path:
+            yield path
+
+
+def list_conflicted():
+    """
+    List files with merge conflicts.
+    """
+    for path in git("diff", "-z", "--name-only", "--diff-filter=U", "--relative").split("\x00"):
+        if path:
+            yield path