test_testing_gates.py

"""Production must never honour a test-only hook.

Every variable exercised here replaces evidence the kit would otherwise gather
from the machine: which platform this is, whether a provider process really
started, which sessions the provider reports, whether an install should abort.
Setting an environment variable is not a privilege, so each hook is armed only
when ``SESSION_KIT_TESTING=1`` is also set. These tests assert both directions
-- ignored without the gate, honoured with it -- because a gate that is only
tested in its armed state is a gate nobody has proven closed.
"""

from __future__ import annotations

import os
import platform
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

from tests.support import REPO, run

sys.path.insert(0, os.fspath(REPO / "lib"))

from sessionkit_inventory import common as inventory_common  # noqa: E402


COMMON = REPO / "bin" / "session_kit_common"
LIFECYCLE = REPO / "lib" / "sh" / "session_kit_lifecycle.sh"
COMMANDS = REPO / "lib" / "sh" / "sp_commands.sh"

# A PID that exists (this process) paired with start ticks that cannot be its
# own, so the real evidence path always answers "not present". Only the
# override can turn this into a success, which is exactly what makes it a
# usable probe for whether the override was honoured.
UNPROVABLE_START_TICKS = "999999999999"


def sandbox_env(**extra: str) -> dict[str, str]:
    environment = {
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        "HOME": os.environ.get("HOME", "/tmp"),
        "PYTHONDONTWRITEBYTECODE": "1",
    }
    environment.update(extra)
    return environment


class ProviderPresenceOverrideGateTests(unittest.TestCase):
    """`sk_provider_process_present` is the entire proof that a launch worked."""

    def presence(self, **extra: str) -> subprocess.CompletedProcess[str]:
        return run(
            [
                "bash",
                "-c",
                'source "$1"; '
                f'sk_provider_process_present "$$" {UNPROVABLE_START_TICKS} codex',
                "presence-gate-test",
                COMMON,
            ],
            env=sandbox_env(**extra),
            check=False,
        )

    def test_override_is_ignored_without_the_testing_gate(self) -> None:
        result = self.presence(SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE="present")

        self.assertNotEqual(0, result.returncode)
        self.assertIn(
            "ignoring SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE", result.stderr
        )
        self.assertIn("reserved for isolated tests", result.stderr)

    def test_override_is_honoured_under_the_testing_gate(self) -> None:
        result = self.presence(
            SESSION_KIT_TESTING="1",
            SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE="present",
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertNotIn("ignoring", result.stderr)

    def test_an_absent_override_still_refuses_under_the_gate(self) -> None:
        """The gate decides whether the hook is read, not what it may say."""
        result = self.presence(
            SESSION_KIT_TESTING="1",
            SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE="absent",
        )

        self.assertNotEqual(0, result.returncode)

    def test_an_unset_override_never_notes_anything(self) -> None:
        result = self.presence()

        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stderr)


class PlatformOverrideGateTests(unittest.TestCase):
    def platform_of(self, **extra: str) -> subprocess.CompletedProcess[str]:
        return run(
            ["bash", "-c", 'source "$1"; sk_platform', "platform-gate-test", COMMON],
            env=sandbox_env(**extra),
            check=False,
        )

    def test_platform_override_is_ignored_without_the_testing_gate(self) -> None:
        result = self.platform_of(SESSION_KIT_TEST_PLATFORM="Darwin")

        self.assertEqual(platform.system(), result.stdout.strip())
        self.assertIn("ignoring SESSION_KIT_TEST_PLATFORM", result.stderr)

    def test_platform_override_is_honoured_under_the_testing_gate(self) -> None:
        result = self.platform_of(
            SESSION_KIT_TESTING="1", SESSION_KIT_TEST_PLATFORM="Darwin"
        )

        self.assertEqual("Darwin", result.stdout.strip())
        self.assertEqual("", result.stderr)


class LifecycleFailpointGateTests(unittest.TestCase):
    """An ungated failpoint lets any process abort every install."""

    def failpoint(self, name: str, **extra: str) -> subprocess.CompletedProcess[str]:
        return run(
            [
                "bash",
                "-c",
                'die() { printf "die: %s\\n" "$*" >&2; exit 9; }; '
                'source "$1"; lifecycle_failpoint "$2"; echo SURVIVED',
                "failpoint-gate-test",
                LIFECYCLE,
                name,
            ],
            env=sandbox_env(**extra),
            check=False,
        )

    def test_failpoint_is_ignored_without_the_testing_gate(self) -> None:
        result = self.failpoint("themes", SESSION_KIT_TEST_FAILPOINT="themes")

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("SURVIVED", result.stdout)

    def test_kill_mode_is_ignored_without_the_testing_gate(self) -> None:
        result = self.failpoint(
            "themes",
            SESSION_KIT_TEST_FAILPOINT="themes",
            SESSION_KIT_TEST_FAILPOINT_MODE="kill",
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("SURVIVED", result.stdout)

    def test_failpoint_is_honoured_under_the_testing_gate(self) -> None:
        result = self.failpoint(
            "themes", SESSION_KIT_TESTING="1", SESSION_KIT_TEST_FAILPOINT="themes"
        )

        self.assertEqual(9, result.returncode)
        self.assertIn("isolated test failpoint after themes", result.stderr)
        self.assertNotIn("SURVIVED", result.stdout)

    def test_a_different_failpoint_name_never_fires(self) -> None:
        result = self.failpoint(
            "themes",
            SESSION_KIT_TESTING="1",
            SESSION_KIT_TEST_FAILPOINT="theme-copy",
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("SURVIVED", result.stdout)

    def test_the_theme_copy_failpoint_shares_the_one_gate(self) -> None:
        """`session_kit_install.sh` asks the same helper, so it cannot drift."""
        source = (REPO / "lib/sh/session_kit_install.sh").read_text(encoding="utf-8")
        theme_copy = source.split("install_codex_themes() {", 1)[1].split(
            "install_codex_terminal_title() {", 1
        )[0]

        self.assertIn("lifecycle_failpoint_armed theme-copy", theme_copy)
        self.assertNotIn("SESSION_KIT_TEST_FAILPOINT", theme_copy)


class ProviderSnapshotFixtureGateTests(unittest.TestCase):
    """The fixture hooks replace the provider's own answer about what exists."""

    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.fixture = Path(self.temporary.name) / "agents.json"
        self.fixture.write_text(
            '[{"sessionId": "00000000-0000-4000-8000-000000000001"}]',
            encoding="utf-8",
        )

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def command_json(self, environ: dict[str, str]) -> object:
        called: list[list[str]] = []

        def runner(argv, timeout):  # noqa: ANN001 - mirrors the injected Runner
            called.append(list(argv))
            return "[]"

        payload = inventory_common._command_json(
            fixture_env="SESSION_KIT_CLAUDE_JSON_FILE",
            command_env="SESSION_KIT_CLAUDE_CMD",
            default_command=("claude", "agents", "--json"),
            runner=runner,
            timeout=1.0,
            environ=environ,
            load_json_file=lambda path: {"fixture": str(path)},
            command_from_env=lambda name, default: [
                *environ.get(name, default).split()
            ],
        )
        return payload, called

    def test_snapshot_fixture_is_ignored_without_the_testing_gate(self) -> None:
        payload, called = self.command_json(
            {"SESSION_KIT_CLAUDE_JSON_FILE": str(self.fixture)}
        )

        self.assertEqual([], payload)
        self.assertEqual([["claude", "agents", "--json"]], called)

    def test_snapshot_fixture_is_honoured_under_the_testing_gate(self) -> None:
        payload, called = self.command_json(
            {
                "SESSION_KIT_TESTING": "1",
                "SESSION_KIT_CLAUDE_JSON_FILE": str(self.fixture),
            }
        )

        self.assertEqual({"fixture": str(self.fixture)}, payload)
        self.assertEqual([], called)


class ProcRootGateTests(unittest.TestCase):
    """`/proc` is where every identity proof in the kit gets its answer."""

    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.fake_proc = Path(self.temporary.name) / "proc"
        self.fake_proc.mkdir()

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def test_fixture_root_is_ignored_without_the_testing_gate(self) -> None:
        chosen = inventory_common.proc_root(
            {"SESSION_KIT_PROC_ROOT": os.fspath(self.fake_proc)}
        )

        self.assertEqual(Path("/proc"), chosen)

    def test_fixture_root_is_honoured_under_the_testing_gate(self) -> None:
        chosen = inventory_common.proc_root(
            {
                "SESSION_KIT_TESTING": "1",
                "SESSION_KIT_PROC_ROOT": os.fspath(self.fake_proc),
            }
        )

        self.assertEqual(self.fake_proc, chosen)

    def test_an_unset_fixture_root_is_the_real_one(self) -> None:
        self.assertEqual(Path("/proc"), inventory_common.proc_root({}))
        self.assertEqual(
            Path("/proc"), inventory_common.proc_root({"SESSION_KIT_TESTING": "1"})
        )

    def test_the_reaper_reads_the_real_proc_without_the_gate(self) -> None:
        result = run(
            [
                "bash",
                "-c",
                'set -a; source /dev/stdin <<<"$(sed -n "1,/^SENTINEL=/p" "$1")"; '
                'printf "%s\\n" "$PROC_ROOT"',
                "reaper-proc-root-test",
                REPO / "bin" / "shpool_reaper",
            ],
            env=sandbox_env(SESSION_KIT_PROC_ROOT=os.fspath(self.fake_proc)),
            check=False,
        )

        self.assertEqual("/proc", result.stdout.strip(), result.stderr)

    def test_the_reaper_reads_the_fixture_proc_under_the_gate(self) -> None:
        result = run(
            [
                "bash",
                "-c",
                'set -a; source /dev/stdin <<<"$(sed -n "1,/^SENTINEL=/p" "$1")"; '
                'printf "%s\\n" "$PROC_ROOT"',
                "reaper-proc-root-test",
                REPO / "bin" / "shpool_reaper",
            ],
            env=sandbox_env(
                SESSION_KIT_TESTING="1",
                SESSION_KIT_PROC_ROOT=os.fspath(self.fake_proc),
            ),
            check=False,
        )

        self.assertEqual(os.fspath(self.fake_proc), result.stdout.strip(), result.stderr)


class HistoryExecutableOverrideGateTests(unittest.TestCase):
    """History tool substitutions are isolated-test hooks, never production input."""

    OVERRIDES = (
        "SESSION_KIT_JOURNAL_RENDER_TOOL",
        "SESSION_KIT_TRANSCRIPT_TEXT_TOOL",
        "SESSION_KIT_HISTORY_SEARCH_TOOL",
    )

    def selected_path(
        self, name: str, **extra: str
    ) -> subprocess.CompletedProcess[str]:
        return run(
            [
                "bash",
                "-c",
                'source "$1"; source "$2"; sk_history_tool_path "$3" /trusted/tool',
                "history-tool-gate-test",
                COMMON,
                COMMANDS,
                name,
            ],
            env=sandbox_env(**{name: "/untrusted/tool"}, **extra),
            check=False,
        )

    def test_overrides_are_ignored_without_the_testing_gate(self) -> None:
        for name in self.OVERRIDES:
            with self.subTest(name=name):
                result = self.selected_path(name)
                self.assertEqual("/trusted/tool", result.stdout.strip())
                self.assertIn(f"ignoring {name}", result.stderr)
                self.assertIn("reserved for isolated tests", result.stderr)

    def test_overrides_are_honoured_under_the_testing_gate(self) -> None:
        for name in self.OVERRIDES:
            with self.subTest(name=name):
                result = self.selected_path(name, SESSION_KIT_TESTING="1")
                self.assertEqual("/untrusted/tool", result.stdout.strip())
                self.assertEqual("", result.stderr)


class NoUngatedTestHookRemainsTests(unittest.TestCase):
    """A regression net: a new ungated hook fails here, not in production."""

    HOOKS = (
        "SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE",
        "SESSION_KIT_TEST_FAILPOINT",
        "SESSION_KIT_TEST_FAILPOINT_MODE",
        "SESSION_KIT_TEST_MONOTONIC_NS",
        "SESSION_KIT_TEST_PLATFORM",
        "SESSION_KIT_PROC_ROOT",
        "SESSION_KIT_SHPOOL_JSON_FILE",
        "SESSION_KIT_CLAUDE_JSON_FILE",
        "SESSION_KIT_JOURNAL_RENDER_TOOL",
        "SESSION_KIT_TRANSCRIPT_TEXT_TOOL",
        "SESSION_KIT_HISTORY_SEARCH_TOOL",
    )
    # The files that own a gate. Every other reader has to sit beside one.
    GATEKEEPERS = {
        "bin/session_kit_common",
        "bin/shpool_reaper",
        "lib/sh/session_kit_lifecycle.sh",
        "lib/sessionkit_inventory/common.py",
    }

    def production_files(self) -> list[Path]:
        listed = subprocess.run(
            ["git", "ls-files"],
            cwd=REPO,
            text=True,
            stdout=subprocess.PIPE,
            check=True,
        ).stdout.split()
        keep = []
        for name in listed:
            if name.startswith(("tests/", "docs/", ".github/")):
                continue
            path = REPO / name
            if path.suffix in {".md", ".json", ".txt", ".patch", ".tmTheme"}:
                continue
            if path.is_file():
                keep.append(path)
        return keep

    def test_every_test_hook_read_sits_next_to_the_testing_gate(self) -> None:
        offenders = []
        for path in self.production_files():
            relative = path.relative_to(REPO).as_posix()
            if relative in self.GATEKEEPERS:
                continue
            try:
                lines = path.read_text(encoding="utf-8").splitlines()
            except (OSError, UnicodeDecodeError):
                continue
            for number, line in enumerate(lines, 1):
                if not any(hook in line for hook in self.HOOKS):
                    continue
                # The gate may sit on this line or within six lines either
                # side, which covers the longest multi-line condition in the
                # tree while still requiring the hook and gate to be adjacent.
                window = "\n".join(lines[max(0, number - 7) : number + 6])
                if "SESSION_KIT_TESTING" in window:
                    continue
                if any(
                    marker in line
                    for marker in (
                        "sk_test_hook",
                        "sk_history_tool_path",
                        "failpoint_armed",
                        "proc_root(",
                        # Names the variable for `_command_json`, which is
                        # where that hook's gate lives.
                        "fixture_env=",
                    )
                ):
                    continue
                if line.lstrip().startswith(("#", "``", '"')):
                    continue
                offenders.append(f"{relative}:{number}: {line.strip()}")
        self.assertEqual([], offenders)


if __name__ == "__main__":
    unittest.main()