ersatztv/scripts/tests/hook_fire_isolation.py

"""Resolve where a subprocess would write its hook-fire records, and require an isolated one.

Split out of `conftest.py` so it can be imported: a test that drives these functions, and a suite
that has to build a subprocess environment from scratch, both need them, and importing `conftest`
under its own name would construct a SECOND `ProductionLogGuard` — whose `inspect.signature` call
would then read the already-wrapped `Popen` and fail at import. A plain module is imported once and
cached, so there is one guard and one production path per session.

See `conftest.py`'s docstring for what the two isolation layers are and why neither observes the
production directory (ersatztv#776, #809, #822).
"""

from __future__ import annotations

import inspect
import os
import subprocess
from collections.abc import Mapping
from pathlib import Path

ENV_VAR = "ETV_HOOK_FIRE_LOG_DIR"

# `scripts/hook-fire-log.sh`'s `etv_hook_fire_log_dir()` is one parameter expansion:
#     ${ETV_HOOK_FIRE_LOG_DIR:-${HOME:-/tmp}/.cache/ersatztv/hook-fire}
# Both `:-` defaults are reproduced below rather than restated as a literal path, because a guard
# that hardcodes where the sink writes stops guarding the moment the sink moves — and does so
# silently, by comparing against a directory nothing uses any more. The agreement is not left as an
# argument: `test_hook_fire_log.py::test_the_resolver_agrees_with_the_SINKS_OWN_expansion` runs both
# over the same matrix of environments.
_DEFAULT_SUFFIX = (".cache", "ersatztv", "hook-fire")
_HOME_FALLBACK = "/tmp"  # noqa: S108 — mirrors the sink's own `${HOME:-/tmp}`


def resolved_log_dir(env: Mapping[str, str]) -> Path:
    """The directory `etv_hook_fire_log_dir` yields for a child launched with `env`.

    Shell's `:-` treats an EMPTY value as absent, so `""` must fall through to the default exactly
    as it does there; `env.get(k) or default` is that semantic and `env.get(k, default)` is not.

    SCOPE, because the obvious wider claim would be false: this models the log DIRECTORY, not the
    final file. `etv_hook_fire_record` writes to `${ETV_HOOK_FIRE_LOG_FILE:-...}`, so an inherited
    `ETV_HOOK_FIRE_LOG_FILE` would in principle steer a record anywhere. It cannot in practice —
    `etv_hook_fire_begin` recomputes that variable from this directory before any write, on both of
    its paths, and nothing exports it — so it is deliberately not modelled here rather than
    modelled wrongly. The one path reaching a write WITHOUT `begin` is the `record` CLI subcommand,
    which does honour an inherited value; it has no call site in this repo. If a future change stops
    recomputing the variable, or a caller starts using `record`, this becomes a real gap. (Cited by
    SYMBOL rather than line number: the commit that added this note also shifted every line number
    it first carried, by editing the header of the very file it points at.)
    """
    return Path(env.get(ENV_VAR) or Path(env.get("HOME") or _HOME_FALLBACK, *_DEFAULT_SUFFIX))


# Resolved from a mapping holding only HOME — what the sink sees when the variable is unset. The
# mapping carries no `ETV_HOOK_FIRE_LOG_DIR`, so this reads the sink's DEFAULT regardless of what
# this session later exports; it does not depend on running before `pytest_configure`.
PRODUCTION_LOG_DIR = resolved_log_dir({"HOME": os.environ.get("HOME", "")})

# BOTH of the sink's default branches, because there are two shared logs and an explicit value can
# name either. `${HOME:-/tmp}` means an empty mapping yields `/tmp/.cache/ersatztv/hook-fire`, which
# is machine-wide, survives every run and is readable with `hook-fire-log.sh report --dir` exactly
# like the `$HOME` one. Requiring the variable (below) closes the OMISSION route for both branches;
# this tuple is what stops the variable being set TO one of them. Where `HOME` is unset the two
# collapse to the same path, which is harmless.
SHARED_LOG_DIRS = (PRODUCTION_LOG_DIR, resolved_log_dir({}))


def _anchored(path: Path, cwd) -> str:
    """Absolute realpath, resolving a RELATIVE value against the CHILD's working directory.

    `Popen(cwd=...)` changes the directory the child resolves a relative `ETV_HOOK_FIRE_LOG_DIR`
    against, so anchoring at the parent's cwd models the wrong process. That is a LIVE BYPASS:
    `ETV_HOOK_FIRE_LOG_DIR=.cache/ersatztv/hook-fire` with `cwd=$HOME` lands
    exactly on the shared log while a parent-anchored check sees an unrelated path under the repo
    and clears it.

    THREE cwd shapes, not four. `Popen` accepts str, bytes and `PathLike`; it calls `os.fsencode`,
    which rejects an int, so a file DESCRIPTOR is not a launch shape and needs no handling here —
    measured on CPython 3.9 and 3.13, and pinned by
    `test_the_guard_JUDGES_every_cwd_SHAPE_that_Popen_ACCEPTS`. A branch for the int case is DEAD
    code, because `os.fspath` already raises `TypeError` on
    an int, and deleting it left that branch's own test green. `os.fsdecode(os.fspath(...))`
    normalises the three shapes that do exist; `os.fspath` alone leaves
    `os.path.join` raising on mixing str with bytes.

    realpath, not `==`: `/tmp` is `/private/tmp` on macOS and `$HOME` may itself be a symlink.
    """
    value = os.fsdecode(os.fspath(path))
    if os.path.isabs(value):
        return os.path.realpath(value)  # `cwd` cannot change where an absolute path points
    base = os.getcwd() if cwd is None else os.fsdecode(os.fspath(cwd))
    return os.path.realpath(os.path.join(base, value))


def _decoded(env: Mapping) -> dict[str, str]:
    """Normalise a subprocess environment to str keys and values, ONCE, at the boundary.

    `Popen` accepts a bytes-keyed and/or bytes-valued `env` on POSIX, and both reach here from a
    legal launch. Unnormalised, a bytes KEY makes the `ETV_HOOK_FIRE_LOG_DIR` lookup miss and the
    launch is rejected as carrying no isolated dir — a false positive on a correctly isolated
    child — while a bytes VALUE makes `Path(...)` raise `TypeError`. Both are the "red on a
    legitimate launch" failure ersatztv#809 explicitly forbids, and they are the same shape as the
    bytes-`cwd` regression `_anchored` records.

    Done here rather than at each lookup so there is one place that knows about the two encodings,
    and no later reader has to remember which call sites handle which.
    """
    return {os.fsdecode(k): os.fsdecode(v) for k, v in env.items()}


def isolation_violation(env: Mapping | None, cwd=None) -> str | None:
    """The message for a child that would log outside this session's isolation, else `None`.

    THE RULE IS "CARRIES AN ISOLATED DIRECTORY", NOT "IS NOT THE PRODUCTION ONE", and the difference
    is the whole point. `etv_hook_fire_log_dir` has TWO fallback branches — `$HOME/.cache/...` and,
    when `HOME` is unset or empty, `/tmp/.cache/...` — so a check that compares against a single
    resolved production path models one of them and is blind to the other. Measured on this suite:
    of the 83 launches whose environment is built from scratch, 81 carry no `HOME` either, so they
    resolve to `/tmp/.cache/ersatztv/hook-fire` — a machine-wide, run-to-run persistent hook-fire
    log that a production-path comparison waves straight through. Requiring the variable
    covers both branches and every branch a future sink might add, and it does not depend on what
    `HOME` happens to be on the machine running the suite — which is also what stops the check
    firing on legitimate launches wherever `HOME` is unset or is itself `/tmp`.

    `env=None` means the child inherits `os.environ`, which `pytest_configure` has already isolated
    — but it is RESOLVED rather than assumed safe, so a test that deletes the variable is caught by
    this clause instead of by nothing.
    """
    env = _decoded(os.environ if env is None else env)
    if not env.get(ENV_VAR):
        return (
            f"this test launched a subprocess carrying no {ENV_VAR}, so `scripts/hook-fire-log.sh` "
            f"falls back to its own default — {resolved_log_dir(env)} — which is a hook-fire log "
            "shared between runs and between everyone using this machine. Any hook the child drives "
            "appends there, and the fire-log library is fail-open by design, so the leak is "
            "invisible from the hook side and every assertion stays green. Build the child's "
            f"environment from `os.environ` (conftest.py has already isolated it), or copy {ENV_VAR} "
            "into it explicitly if it must be built from scratch."
        )
    # An explicit value still must not BE a shared log — the variable can be set wrongly as easily
    # as it can be omitted, and the likeliest wrong value is the sink's own default transliterated
    # into a `dict.get(k, "<default>")`, which is how BOTH branches become reachable again.
    target = _anchored(resolved_log_dir(env), cwd)
    for shared in SHARED_LOG_DIRS:
        # Absolute by construction, so `cwd` cannot change where they point.
        root = os.path.realpath(shared)
        # CONTAINMENT, not equality: `etv_hook_fire_report` reads its directory with
        # `find "$dir" -name '*.jsonl' -type f`, which RECURSES, so a subdirectory of a shared log
        # is picked up by `hook-fire-log.sh report --dir <shared>` exactly like a file at its root.
        # An equality test calls `$HOME/.cache/ersatztv/hook-fire/sub` clean while records written
        # there show up in the report — the same "looks identical, answers a different question"
        # failure the whole isolation exists to remove.
        if target == root or target.startswith(root + os.sep):
            return (
                f"this test launched a subprocess whose {ENV_VAR} is inside a shared hook-fire log "
                f"({shared}) — one of the two directories `etv_hook_fire_log_dir` falls back to, "
                "both of which outlive the run, are shared with everyone using this machine, and "
                "are read RECURSIVELY by `hook-fire-log.sh report --dir`. Point it at a directory "
                "this test session owns and deletes."
            )
    return None


class ProductionLogGuard:
    """Resolve every subprocess launch against the sink's rule, at the launch itself.

    Only `Popen` is wrapped, because `subprocess.run`, `check_output` and `call` all reach the
    module-global `Popen` by name at call time — so wrapping the one covers the four, and a second
    wrapper would be a second place to keep in step with the first.

    Its population is every launch that actually happens. That is the point: a list of hook-driving
    suites, or a scan of their source, would have to be kept in step with the suite, and
    `testing.guard-derives-population-from-source` is the record of what that costs.

    WHAT IT DOES NOT SEE, stated so the next reader does not have to rediscover it. A child started
    outside `Popen` — `os.execve`, `os.posix_spawn`, `os.system` — is not routed through this at
    all. Neither is a child that is launched correctly and then re-execs, nor a shell script the
    suite writes that unsets `ETV_HOOK_FIRE_LOG_DIR` itself before sourcing the sink: the guard
    inspects the environment handed OVER, not what the child does with it afterwards. Nor is a
    caller that bound `Popen` BY NAME (`from subprocess import Popen`) before `pytest_configure`
    installed this — only a pytest plugin imports that early, and none here does. None of these
    occur in this suite today, and each would be a deliberate act rather than the accident this
    exists to catch; the accident is an environment assembled without the variable, and that is
    visible at the launch.

    A test that DELETES the variable from `os.environ` and then launches with `env=None` IS caught,
    because `isolation_violation` resolves `os.environ` rather than assuming it safe.

    Nor does it check that the directory is a GOOD one, only that it is not a shared hook-fire log:
    `ETV_HOOK_FIRE_LOG_DIR=/tmp` or `=$HOME/.cache/ersatztv` passes while writing a session file into
    a machine-wide, run-to-run persistent directory. The rule's name — "carries an isolated dir" —
    is wider than what is implemented, which is "carries a dir that is not one of the two shared
    logs". Nothing does that today. A bytes-keyed or bytes-valued `env` IS handled, at the boundary.

    One more residual, measured rather than assumed: containment is a STRING comparison, and macOS
    volumes are case-insensitive, so `$HOME/.cache/ErsatzTV/hook-fire` is the same directory as
    `$HOME/.cache/ersatztv/hook-fire` and is cleared. Left as a note rather than fixed with an
    inode comparison, deliberately: nothing spells it that way, the likeliest wrong value — the
    sink's own default transliterated — matches case exactly, and two consecutive fixes here have
    each created the next finding, which is this corpus's stated trigger for subtracting a layer
    rather than adding one.
    """

    def __init__(self) -> None:
        self._real_popen = subprocess.Popen
        # Derived from the real signature, never a literal index: `env` is a positional parameter of
        # `Popen.__init__`, and a hand-counted position would silently start reading `cwd` — or
        # nothing — if CPython ever reordered it, leaving the guard passing on every launch.
        params = list(inspect.signature(self._real_popen.__init__).parameters)
        self._env_pos = params.index("env") - 1  # `self` is not passed at the call site
        # `cwd` too, because a RELATIVE log dir resolves against the CHILD's directory, not ours.
        self._cwd_pos = params.index("cwd") - 1
        self.checked = 0
        self.installed = False

    def install(self) -> None:
        guard = self
        real = self._real_popen

        class GuardedPopen(real):  # type: ignore[misc,valid-type]
            # Declared in the class body, not attached afterwards, so it is a real attribute a type
            # checker can see — and so `test_a_launch_NOT_CARRYING_...` can assert the guard is
            # INSTALLED before concluding anything from its own launch going through.
            etv_production_log_guard = guard

            def __init__(self, *args, **kwargs):
                guard.check(args, kwargs)
                super().__init__(*args, **kwargs)

        GuardedPopen.__name__ = real.__name__
        GuardedPopen.__qualname__ = real.__qualname__
        subprocess.Popen = GuardedPopen
        self.installed = True

    def uninstall(self) -> None:
        subprocess.Popen = self._real_popen
        self.installed = False

    def _argument(self, name: str, pos: int, args: tuple, kwargs: dict):
        """What a `Popen` call passed for `name`, whether by keyword or in its positional slot."""
        if name in kwargs:
            return kwargs[name]
        return args[pos] if len(args) > pos else None

    def env_of(self, args: tuple, kwargs: dict):
        return self._argument("env", self._env_pos, args, kwargs)

    def cwd_of(self, args: tuple, kwargs: dict):
        return self._argument("cwd", self._cwd_pos, args, kwargs)

    def check(self, args: tuple, kwargs: dict) -> None:
        self.checked += 1
        violation = isolation_violation(self.env_of(args, kwargs), self.cwd_of(args, kwargs))
        if violation is not None:
            raise AssertionError(violation)