ServerKit/backend/tests/subprocess_stub.py

"""Shared subprocess stub kit (plan 75 §G7).

The suite stubs subprocess by hand, everywhere, and the shape of those stubs
is what entrenches the very duplication plan 75 exists to remove:

* ``_FakeProc(returncode, stdout, stderr)`` is pasted **byte-identical** in
  four test modules (plus variants elsewhere);
* ~25 ``def fake_run`` closures re-implement argv dispatch;
* **zero** of them patch ``run_privileged``/``run_unprivileged``. Every stub
  targets raw ``subprocess`` *inside one service module*, so each new service
  test recreates the raw seam instead of the wrapper seam — the 250-raw-call
  surface grows a test that pins it in place.

This module is the single seam. Tests script argv → result; when G1 migrates
services onto ``run_checked()`` the retarget happens **here**, once, instead of
in twelve test files.

Two properties are deliberate, and both come straight from plan 74's outage:

1. **An unscripted command is an error, never a success.** A stub that invents
   ``returncode=0`` for a command the test never described is the same defect
   as a blanket ``except`` reporting "not installed" — a fact nobody
   established. :class:`UnscriptedCommand` is a ``BaseException`` for the same
   reason :class:`popen_guard.SbinPathError` is: services wrap subprocess in
   ``except Exception`` and convert it to ``{'success': False}``, so an
   ``AssertionError`` here would be swallowed by exactly the handlers this kit
   is meant to test, and the test would see a plausible failure dict instead
   of "you forgot to script this".

2. **Stubbed execs are still sbin-guarded.** Patching ``subprocess`` removes
   the §B2 runtime guard for the duration of the test. The kit calls
   :func:`popen_guard.guard_argv` itself, so scripting a command does not buy
   an exemption from the PATH rule.

Usage::

    def test_reload(fake_subprocess):
        fake_subprocess.script(['systemctl', 'reload'], stdout='')
        fake_subprocess.script(['nginx', '-t'], returncode=1, stderr='bad')
        ...
        assert fake_subprocess.argv_for(['nginx']) == ['nginx', '-t']

Matching ignores the two things that legitimately vary at the door: a ``sudo``
prefix (``run_privileged`` adds one only when the process is not root) and
argv[0] absolutisation (``/usr/sbin/nginx`` vs ``nginx``). A test asserts on
the command, not on how the helper had to spell it today.
"""

import os
import subprocess

from popen_guard import guard_argv

_SUDO_FLAGS_WITH_VALUE = ('-u', '--user', '-g', '--group')


class UnscriptedCommand(BaseException):
    """A stubbed subprocess call the test never described.

    ``BaseException`` on purpose — see the module docstring. Do not "fix" this
    by making it an ``AssertionError``; a blanket ``except Exception`` in the
    service under test would eat it and the failure would surface as a
    confusing assertion about a result dict three layers away.
    """


class FakeProc:
    """``subprocess.CompletedProcess`` stand-in — the one every test shares.

    Carries the same attribute surface tests actually use (``returncode``,
    ``stdout``, ``stderr``, ``args``) plus ``check_returncode()`` so code under
    test that calls it behaves like the real thing.
    """

    def __init__(self, returncode=0, stdout='', stderr='', args=None):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr
        self.args = args

    def check_returncode(self):
        if self.returncode:
            raise subprocess.CalledProcessError(
                self.returncode, self.args, self.stdout, self.stderr)

    def __repr__(self):
        return (f'FakeProc(returncode={self.returncode!r}, '
                f'stdout={self.stdout!r}, stderr={self.stderr!r})')


class FakePopen:
    """Minimal ``Popen`` stand-in over a :class:`FakeProc`.

    Separate type on purpose: a ``Popen``'s ``.stdout`` is an iterable of
    *lines* while a ``CompletedProcess``'s is a string, and one object cannot
    honestly be both — iterating a string yields characters, which is how a
    line-streaming test silently passes over nonsense.
    """

    def __init__(self, proc):
        self._proc = proc
        self.args = proc.args
        self.returncode = None
        text = proc.stdout or ''
        self.stdout = iter(text.splitlines(keepends=True)) if text else iter(())
        self.stderr = iter(())

    def wait(self, timeout=None):
        self.returncode = self._proc.returncode
        return self.returncode

    def poll(self):
        return self.returncode

    def communicate(self, input=None, timeout=None):
        self.returncode = self._proc.returncode
        return self._proc.stdout, self._proc.stderr

    def kill(self):
        self.returncode = self._proc.returncode

    terminate = kill

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.wait()
        return False


def normalize_argv(argv):
    """argv as the *test* means it: no ``sudo`` prefix, argv[0] by basename.

    ``run_privileged`` prepends ``sudo -n`` only when the process is not root,
    and resolves argv[0] to an absolute path only when ``$PATH`` misses it.
    Both are correct, both vary by machine, and neither is what a test is
    asserting about. Normalising here is what lets one scripted rule match on
    a root box, a sudo box, and a dev box alike.
    """
    if isinstance(argv, (str, bytes)) or not argv:
        return []
    parts = [str(a) for a in argv]
    if os.path.basename(parts[0]) == 'sudo':
        parts = parts[1:]
        while parts and parts[0].startswith('-'):
            if parts[0] in _SUDO_FLAGS_WITH_VALUE:
                parts = parts[2:]
            elif parts[0] == '--':
                parts = parts[1:]
                break
            else:
                parts = parts[1:]
    if not parts:
        return []
    return [os.path.basename(parts[0])] + parts[1:]


class _Rule:
    def __init__(self, prefix, responder):
        self.prefix = list(prefix)
        self.responder = responder


class ScriptedSubprocess:
    """Scriptable ``subprocess`` replacement: argv prefix → result.

    Rules are matched **longest prefix first**, ties broken by most recently
    scripted. So a broad rule can be laid down first and narrowed later::

        fake.script(['docker'], stdout='')                  # default
        fake.script(['docker', 'inspect'], returncode=1)    # override
    """

    def __init__(self):
        self._rules = []
        self.calls = []  # [(argv, kwargs), ...] — every call, in order

    # ---- scripting -----------------------------------------------------
    def script(self, prefix, *, returncode=0, stdout='', stderr='',
               raises=None):
        """Answer any command starting with *prefix*.

        *raises* takes an exception instance or class and is raised instead of
        returning — this is how the §A probe-contract suites force
        ``FileNotFoundError`` / ``TimeoutExpired`` / ``PermissionError``
        without needing a machine that lacks the binary.
        """
        def responder(argv, kwargs):
            if raises is not None:
                raise raises() if isinstance(raises, type) else raises
            return FakeProc(returncode=returncode, stdout=stdout,
                            stderr=stderr, args=argv)

        return self.when(prefix, responder)

    def when(self, prefix, responder):
        """Answer *prefix* with ``responder(argv, kwargs) -> FakeProc``.

        The responder receives the **normalised** argv (no sudo prefix,
        argv[0] by basename) for the same reason matching uses it: a stub that
        emulates ``tee`` should not have to know whether the door decided to
        add ``sudo`` on this machine. The raw argv is still recorded, and
        :meth:`argv_for` hands it back.
        """
        self._rules.append(_Rule(normalize_argv(prefix), responder))
        return self

    # ---- the patched entry points --------------------------------------
    def run(self, cmd, *args, **kwargs):
        return self._dispatch(cmd, kwargs)

    def popen(self, cmd, *args, **kwargs):
        return FakePopen(self._dispatch(cmd, kwargs))

    def check_output(self, cmd, *args, **kwargs):
        proc = self._dispatch(cmd, kwargs)
        proc.check_returncode()
        return proc.stdout

    def call(self, cmd, *args, **kwargs):
        return self._dispatch(cmd, kwargs).returncode

    def check_call(self, cmd, *args, **kwargs):
        proc = self._dispatch(cmd, kwargs)
        proc.check_returncode()
        return 0

    def install(self, monkeypatch, module=subprocess):
        """Patch every exec entry point on *module* onto this script."""
        monkeypatch.setattr(module, 'run', self.run)
        monkeypatch.setattr(module, 'Popen', self.popen)
        monkeypatch.setattr(module, 'check_output', self.check_output)
        monkeypatch.setattr(module, 'call', self.call)
        monkeypatch.setattr(module, 'check_call', self.check_call)
        return self

    # ---- assertions ----------------------------------------------------
    def commands(self):
        """Every argv seen, normalised — the list to assert ordering on."""
        return [normalize_argv(argv) for argv, _ in self.calls]

    def argv_for(self, prefix):
        """The RAW argv of the first call matching *prefix*, or ``None``.

        Raw, not normalised: this is what a test asserting "the secret went to
        stdin and never onto the argv" needs to inspect.
        """
        wanted = normalize_argv(prefix)
        for argv, _ in self.calls:
            if self._matches(normalize_argv(argv), wanted):
                return argv
        return None

    def kwargs_for(self, prefix):
        """The kwargs of the first call matching *prefix*, or ``None``."""
        wanted = normalize_argv(prefix)
        for argv, kwargs in self.calls:
            if self._matches(normalize_argv(argv), wanted):
                return kwargs
        return None

    def writes(self):
        """``{path: content}`` for every privileged file write seen.

        ``write_privileged_file`` (plan 75 §G2) spells a write as
        ``tee [-a] <path>`` with the content on stdin, and three suites already
        rebuilt this dict by hand from their own stub. Reading it here means
        that when the door switches to a tmp-file + ``os.replace`` write, the
        assertions follow in one place instead of in each of them.
        """
        out = {}
        for raw, kwargs in self.calls:
            argv = normalize_argv(raw)
            if not argv or argv[0] != 'tee':
                continue
            path = argv[2] if len(argv) > 2 and argv[1] == '-a' else argv[1]
            content = kwargs.get('input', '')
            out[path] = out.get(path, '') + content if argv[1] == '-a' else content
        return out

    def ran(self, prefix):
        """True when at least one call matched *prefix*."""
        return self.argv_for(prefix) is not None

    # ---- internals ------------------------------------------------------
    @staticmethod
    def _matches(argv, prefix):
        return len(argv) >= len(prefix) and argv[:len(prefix)] == prefix

    def _dispatch(self, cmd, kwargs):
        guard_argv(cmd)
        self.calls.append((cmd, kwargs))
        argv = normalize_argv(cmd)
        best = None
        for rule in self._rules:
            if not self._matches(argv, rule.prefix):
                continue
            if best is None or len(rule.prefix) >= len(best.prefix):
                best = rule
        if best is None:
            raise UnscriptedCommand(
                f'no scripted result for {list(cmd)!r} (normalised {argv!r}).\n'
                f'  scripted: {[r.prefix for r in self._rules] or "nothing"}\n'
                'A stub that answered anyway would be inventing a fact the '
                'test never established — plan 74\'s bug class, in test form. '
                'Call fake_subprocess.script([...]) for this command.'
            )
        return best.responder(argv, kwargs)