homeassistant-stelloauth-addon/tests/test_supervisor.py

from __future__ import annotations

import importlib.machinery
import importlib.util
import json
import logging
import signal
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest


ROOT = Path(__file__).parents[1]
SUPERVISOR_PATH = ROOT / "stelloauth/rootfs/usr/local/bin/addon-supervisor"


def load_supervisor():
    loader = importlib.machinery.SourceFileLoader("addon_supervisor", str(SUPERVISOR_PATH))
    spec = importlib.util.spec_from_loader(loader.name, loader)
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[loader.name] = module
    loader.exec_module(module)
    return module


@pytest.fixture
def supervisor():
    return load_supervisor()


def test_options_load_valid_defaults(supervisor, tmp_path: Path) -> None:
    path = tmp_path / "options.json"
    path.write_text(
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": 5,
                "rate_limit_duration": "1h",
            }
        ),
        encoding="utf-8",
    )

    assert supervisor.load_options(path) == supervisor.Options("60s", 5, "1h")


def test_environment_maps_options_and_preserves_parent(supervisor, monkeypatch) -> None:
    monkeypatch.setenv("PARENT_SENTINEL", "preserved")

    environment = supervisor.build_environment(supervisor.Options("60s", 5, "1h"))

    assert environment["PARENT_SENTINEL"] == "preserved"
    assert {key: environment[key] for key in (
        "CLOAK_CDP_URL",
        "CLOAK_MAX_SESSIONS",
        "CLOAK_QUEUE_TIMEOUT",
        "RATE_LIMIT_COUNT",
        "RATE_LIMIT_DURATION",
        "HTTP_ADDRESS",
        "PORT",
        "METRICS_ADDRESS",
        "METRICS_PORT",
    )} == {
        "CLOAK_CDP_URL": "http://127.0.0.1:9222",
        "CLOAK_MAX_SESSIONS": "1",
        "CLOAK_QUEUE_TIMEOUT": "60s",
        "RATE_LIMIT_COUNT": "5",
        "RATE_LIMIT_DURATION": "1h",
        "HTTP_ADDRESS": "0.0.0.0",
        "PORT": "8080",
        "METRICS_ADDRESS": "127.0.0.1",
        "METRICS_PORT": "9090",
    }


@pytest.mark.parametrize(
    "content",
    [
        "{}",
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": 5,
                "rate_limit_duration": "1h",
                "unknown-SENTINEL": "secret-SENTINEL",
            }
        ),
        json.dumps({"queue_timeout": "60s", "rate_limit_count": 5}),
        json.dumps(
            {
                "queue_timeout": "0s-SENTINEL",
                "rate_limit_count": 5,
                "rate_limit_duration": "1h",
            }
        ),
        json.dumps(
            {
                "queue_timeout": "60-SENTINEL",
                "rate_limit_count": 5,
                "rate_limit_duration": "1h",
            }
        ),
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": True,
                "rate_limit_duration": "1h",
            }
        ),
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": 0,
                "rate_limit_duration": "1h",
            }
        ),
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": 21,
                "rate_limit_duration": "1h",
            }
        ),
        json.dumps(
            {
                "queue_timeout": "60s",
                "rate_limit_count": 5,
                "rate_limit_duration": "1-SENTINEL",
            }
        ),
        '{"queue_timeout":"malformed-SENTINEL"',
    ],
)
def test_invalid_options_raise_fixed_non_secret_error(
    supervisor, tmp_path: Path, caplog, content: str
) -> None:
    path = tmp_path / "options.json"
    path.write_text(content, encoding="utf-8")

    with caplog.at_level(logging.INFO), pytest.raises(
        supervisor.ConfigError, match="^Invalid add-on configuration$"
    ):
        supervisor.load_options(path)

    assert "SENTINEL" not in caplog.text
    assert "SENTINEL" not in str(sys.exc_info())


def test_unreadable_options_raise_fixed_non_secret_error(
    supervisor, tmp_path: Path, caplog
) -> None:
    missing = tmp_path / "missing-SENTINEL.json"

    with caplog.at_level(logging.INFO), pytest.raises(
        supervisor.ConfigError, match="^Invalid add-on configuration$"
    ):
        supervisor.load_options(missing)

    assert "SENTINEL" not in caplog.text


def test_invalid_options_main_logs_only_fixed_error(
    supervisor, tmp_path: Path, caplog, monkeypatch
) -> None:
    path = tmp_path / "options.json"
    path.write_text('{"credential":"secret-SENTINEL"}', encoding="utf-8")
    monkeypatch.setattr(supervisor, "OPTIONS_PATH", path)

    with caplog.at_level(logging.INFO):
        assert supervisor.main() == 2

    assert caplog.messages == ["Invalid add-on configuration"]
    assert "SENTINEL" not in caplog.text


def test_probe_cloak_uses_real_cdp_websocket_and_closes_profile(supervisor) -> None:
    calls: list[tuple[str, str, float]] = []

    def request(method: str, url: str, timeout: float):
        calls.append((method, url, timeout))
        if url == supervisor.CLOAK_VERSION:
            return supervisor.HttpResponse(
                200,
                json.dumps(
                    {
                        "webSocketDebuggerUrl": (
                            "ws://127.0.0.1:9222/devtools/browser/readiness"
                        )
                    }
                ).encode(),
            )
        return supervisor.HttpResponse(200, b"ok")

    supervisor.probe_cloak(request)

    assert [(method, url) for method, url, _ in calls] == [
        ("GET", supervisor.CLOAK_ROOT),
        ("GET", supervisor.CLOAK_VERSION),
        ("POST", supervisor.CLOAK_CLOSE),
    ]
    assert all(timeout == 2.0 for _, _, timeout in calls)


@pytest.mark.parametrize(
    "root_status,version_status,version_body,close_status",
    [
        (200, 200, b"{}", 200),
        (200, 200, b'{"webSocketDebuggerUrl":""}', 200),
        (200, 200, b"not-json-SENTINEL", 200),
        (
            200,
            200,
            b'{"webSocketDebuggerUrl":"ws://192.0.2.1:9222/devtools/browser/x"}',
            200,
        ),
        (
            200,
            200,
            b'{"webSocketDebuggerUrl":"WS://127.0.0.1:9222/devtools/browser/x"}',
            200,
        ),
        (
            503,
            200,
            b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}',
            200,
        ),
        (
            200,
            503,
            b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}',
            200,
        ),
        (
            200,
            200,
            b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}',
            503,
        ),
    ],
)
def test_probe_cloak_rejects_invalid_readiness(
    supervisor, root_status, version_status, version_body, close_status
) -> None:
    responses = {
        supervisor.CLOAK_ROOT: supervisor.HttpResponse(root_status, b"root"),
        supervisor.CLOAK_VERSION: supervisor.HttpResponse(version_status, version_body),
        supervisor.CLOAK_CLOSE: supervisor.HttpResponse(close_status, b"close"),
    }

    with pytest.raises(supervisor.ReadinessError):
        supervisor.probe_cloak(lambda _method, url, _timeout: responses[url])


def test_probe_stelloauth_requires_http_200(supervisor) -> None:
    calls = []

    def request(method: str, url: str, timeout: float):
        calls.append((method, url, timeout))
        return supervisor.HttpResponse(200, b"SENTINEL-body-is-ignored")

    supervisor.probe_stelloauth(request)
    assert calls == [("GET", supervisor.STELLOAUTH_ROOT, 2.0)]

    with pytest.raises(supervisor.ReadinessError):
        supervisor.probe_stelloauth(
            lambda _method, _url, _timeout: supervisor.HttpResponse(503, b"")
        )


class FakeClock:
    def __init__(self) -> None:
        self.now = 0.0
        self.sleeps: list[float] = []

    def monotonic(self) -> float:
        return self.now

    def sleep(self, delay: float) -> None:
        self.sleeps.append(delay)
        self.now += delay


def test_readiness_backoff_starts_at_quarter_second_and_caps_at_two(supervisor) -> None:
    clock = FakeClock()
    attempts = 0

    def probe() -> None:
        nonlocal attempts
        attempts += 1
        if attempts <= 5:
            raise OSError("transient-SENTINEL")

    supervisor.wait_until_ready(
        "Service", probe, 60.0, lambda: False, clock.monotonic, clock.sleep
    )

    assert clock.sleeps == [0.25, 0.5, 1.0, 2.0, 2.0]


@pytest.mark.parametrize("name,timeout", [("CloakBrowser", 60.0), ("Stelloauth", 30.0)])
def test_readiness_expires_at_service_timeout(supervisor, name: str, timeout: float) -> None:
    clock = FakeClock()

    with pytest.raises(
        supervisor.ReadinessError, match=f"^{name} did not become ready$"
    ):
        supervisor.wait_until_ready(
            name,
            lambda: (_ for _ in ()).throw(ValueError("transient-SENTINEL")),
            timeout,
            lambda: False,
            clock.monotonic,
            clock.sleep,
        )

    assert timeout <= clock.now <= timeout + 2.0
    assert max(clock.sleeps) == 2.0


def test_readiness_stop_flag_aborts_immediately(supervisor) -> None:
    clock = FakeClock()
    called = False

    def probe() -> None:
        nonlocal called
        called = True

    with pytest.raises(
        supervisor.ReadinessError, match="^CloakBrowser did not become ready$"
    ):
        supervisor.wait_until_ready(
            "CloakBrowser", probe, 60.0, lambda: True, clock.monotonic, clock.sleep
        )

    assert called is False
    assert clock.sleeps == []


def test_probe_http_request_disables_proxies(supervisor, monkeypatch) -> None:
    observed = {}

    class Response:
        status = 200

        def read(self) -> bytes:
            return b"response"

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

    class Opener:
        def open(self, request, timeout):
            observed["request"] = request
            observed["timeout"] = timeout
            return Response()

    def build_opener(handler):
        observed["handler"] = handler
        return Opener()

    monkeypatch.setattr(supervisor.urllib.request, "build_opener", build_opener)

    assert supervisor.http_request("POST", "http://127.0.0.1/", 3.0) == (
        200,
        b"response",
    )
    assert observed["handler"].proxies == {}
    assert observed["request"].get_method() == "POST"
    assert observed["timeout"] == 3.0


class FakeProcess:
    def __init__(self, pid: int, *, ignores_term: bool = False) -> None:
        self.pid = pid
        self.returncode: int | None = None
        self.ignores_term = ignores_term
        self.wait_calls: list[float | None] = []
        self.on_wait = None

    def poll(self) -> int | None:
        return self.returncode

    def wait(self, timeout: float | None) -> int:
        self.wait_calls.append(timeout)
        if self.returncode is None:
            self.returncode = -9
        if self.on_wait is not None:
            self.on_wait()
        return self.returncode


def manager_harness(
    supervisor,
    tmp_path: Path,
    *,
    cloak_probe=None,
    stelloauth_probe=None,
    ignores_term: tuple[bool, bool] = (False, False),
    live_descendants: tuple[bool, bool] = (False, False),
):
    clock = FakeClock()
    events: list[object] = []
    processes = [
        FakeProcess(1001, ignores_term=ignores_term[0]),
        FakeProcess(1002, ignores_term=ignores_term[1]),
    ]
    groups = {process.pid: True for process in processes}
    leaders_reaped = {process.pid: False for process in processes}
    term_delivered = {process.pid: False for process in processes}
    for index, process in enumerate(processes):
        def on_wait(index=index, process=process) -> None:
            leaders_reaped[process.pid] = True
            if not live_descendants[index] or (
                term_delivered[process.pid] and not process.ignores_term
            ):
                groups[process.pid] = False

        process.on_wait = on_wait
    group_checks: list[int] = []
    spawned: list[FakeProcess] = []

    def popen(command, *, env, start_new_session):
        process = processes[len(spawned)]
        spawned.append(process)
        events.append(("start", list(command), env, start_new_session))
        return process

    def killpg(pid: int, sent_signal: int) -> None:
        events.append(("signal", pid, sent_signal))
        process = next(item for item in processes if item.pid == pid)
        if sent_signal == signal.SIGKILL:
            groups[pid] = False
            if process.returncode is None:
                process.returncode = -sent_signal
        else:
            term_delivered[pid] = True
            if not process.ignores_term:
                if process.returncode is None:
                    process.returncode = -sent_signal
                if leaders_reaped[pid]:
                    groups[pid] = False

    def group_alive(pid: int) -> bool:
        group_checks.append(pid)
        return groups[pid]

    def default_cloak_probe() -> None:
        events.append("probe cloak")

    def default_stelloauth_probe() -> None:
        events.append("probe stelloauth")

    profile_path = tmp_path / "cloakserve"
    environment = {"SENSITIVE_SENTINEL": "credential-code-cookie-token-SENTINEL"}
    manager = supervisor.ProcessManager(
        environment,
        popen=popen,
        killpg=killpg,
        group_alive=group_alive,
        cloak_probe=cloak_probe or default_cloak_probe,
        stelloauth_probe=stelloauth_probe or default_stelloauth_probe,
        monotonic=clock.monotonic,
        sleep=clock.sleep,
        profile_path=profile_path,
    )
    return SimpleNamespace(
        manager=manager,
        clock=clock,
        events=events,
        processes=processes,
        groups=groups,
        group_checks=group_checks,
        spawned=spawned,
        profile_path=profile_path,
        environment=environment,
    )


@pytest.mark.parametrize(
    "exception,expected",
    [(None, True), (ProcessLookupError(), False), (PermissionError(), True)],
)
def test_process_group_liveness_uses_signal_zero(
    supervisor, exception: OSError | None, expected: bool
) -> None:
    calls = []

    def killpg(pgid: int, sent_signal: int) -> None:
        calls.append((pgid, sent_signal))
        if exception is not None:
            raise exception

    assert supervisor.process_group_alive(4242, killpg) is expected
    assert calls == [(4242, 0)]


def test_lifecycle_startup_order_and_profiles(supervisor, tmp_path: Path, monkeypatch) -> None:
    harness = manager_harness(supervisor, tmp_path)
    harness.profile_path.mkdir()
    stale = harness.profile_path / "stale-profile"
    stale.write_text("stale", encoding="utf-8")

    original_cloak = harness.manager._cloak_probe
    original_stelloauth = harness.manager._stelloauth_probe

    def cloak_probe() -> None:
        assert not stale.exists()
        assert harness.profile_path.stat().st_mode & 0o777 == 0o700
        original_cloak()

    def stelloauth_probe() -> None:
        original_stelloauth()
        harness.manager._handle_signal(signal.SIGTERM, None)

    harness.manager._cloak_probe = cloak_probe
    harness.manager._stelloauth_probe = stelloauth_probe
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0

    starts_and_probes = [event for event in harness.events if event == "probe cloak" or event == "probe stelloauth" or (isinstance(event, tuple) and event[0] == "start")]
    assert [(event[0], event[1]) if isinstance(event, tuple) else event for event in starts_and_probes] == [
        ("start", supervisor.CLOAK_COMMAND),
        "probe cloak",
        ("start", supervisor.STELLOAUTH_COMMAND),
        "probe stelloauth",
    ]
    for event in starts_and_probes:
        if isinstance(event, tuple):
            assert event[2] is harness.environment
            assert event[3] is True


def test_cloak_readiness_failure_never_starts_stelloauth(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    def failing_probe() -> None:
        raise supervisor.ReadinessError("secret-SENTINEL")

    harness = manager_harness(supervisor, tmp_path, cloak_probe=failing_probe)
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 1
    assert len(harness.spawned) == 1
    assert harness.clock.now >= 60.0
    assert harness.processes[0].wait_calls == [None]


def test_stelloauth_readiness_failure_stops_both_children(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    def failing_probe() -> None:
        raise OSError("credential-SENTINEL")

    harness = manager_harness(supervisor, tmp_path, stelloauth_probe=failing_probe)
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 1
    assert len(harness.spawned) == 2
    assert {(event[1], event[2]) for event in harness.events if event[0] == "signal"} == {
        (1001, signal.SIGTERM),
        (1002, signal.SIGTERM),
    }
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


@pytest.mark.parametrize(
    "child_index,child_status,expected_status",
    [(0, 0, 1), (0, 7, 7), (1, 0, 1), (1, 9, 9)],
)
def test_child_exit_stops_sibling_and_returns_failure(
    supervisor,
    tmp_path: Path,
    monkeypatch,
    child_index: int,
    child_status: int,
    expected_status: int,
) -> None:
    harness = manager_harness(supervisor, tmp_path)
    slept = False

    def sleep(delay: float) -> None:
        nonlocal slept
        harness.clock.sleep(delay)
        if not slept:
            slept = True
            harness.processes[child_index].returncode = child_status

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == expected_status
    sibling = harness.processes[1 - child_index]
    assert ("signal", sibling.pid, signal.SIGTERM) in harness.events
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_exited_leader_with_live_descendants_still_gets_group_sigterm(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(
        supervisor, tmp_path, live_descendants=(True, False)
    )
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        harness.clock.sleep(delay)
        if not triggered:
            triggered = True
            harness.processes[0].returncode = 7
            assert harness.groups[1001] is True

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 7
    assert ("signal", 1001, signal.SIGTERM) in harness.events
    assert ("signal", 1002, signal.SIGTERM) in harness.events
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_leader_exit_on_sigterm_without_descendants_finishes_without_waiting(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path)

    def stelloauth_probe() -> None:
        harness.manager._handle_signal(signal.SIGTERM, None)

    harness.manager._stelloauth_probe = stelloauth_probe
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert harness.clock.now == 0.0
    assert not any(
        event[0] == "signal" and event[2] == signal.SIGKILL
        for event in harness.events
    )
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_exited_leader_with_live_descendants_still_uses_deadline_and_sigkill(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(
        supervisor,
        tmp_path,
        ignores_term=(True, False),
        live_descendants=(True, False),
    )
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        harness.clock.sleep(delay)
        if not triggered:
            triggered = True
            harness.processes[0].returncode = 7

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 7
    assert harness.clock.now == pytest.approx(10.25)
    assert ("signal", 1001, signal.SIGTERM) in harness.events
    assert ("signal", 1001, signal.SIGKILL) in harness.events
    assert ("signal", 1002, signal.SIGTERM) in harness.events
    assert ("signal", 1002, signal.SIGKILL) not in harness.events
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_group_disappearance_ends_shared_wait_without_sigkill(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path, ignores_term=(True, True))
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        if not triggered:
            triggered = True
            harness.manager._handle_signal(signal.SIGTERM, None)
        harness.clock.sleep(delay)
        if harness.clock.now >= 0.5:
            harness.groups[1001] = False
            harness.groups[1002] = False

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert harness.clock.now == pytest.approx(0.5)
    assert all(event[2] != signal.SIGKILL for event in harness.events if event[0] == "signal")
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_dead_group_is_not_rechecked_or_signalled_after_possible_pgid_reuse(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path)
    calls = []

    def group_alive(pgid: int) -> bool:
        calls.append(pgid)
        if pgid == 1001:
            return len([item for item in calls if item == 1001]) > 1
        return harness.groups[pgid]

    harness.manager._group_alive = group_alive
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        harness.clock.sleep(delay)
        if not triggered:
            triggered = True
            harness.processes[0].returncode = 7

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 7
    assert calls.count(1001) == 1
    assert not any(
        event[0] == "signal" and event[1] == 1001 for event in harness.events
    )
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_group_disappearing_at_deadline_is_rechecked_before_sigkill(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path, ignores_term=(True, True))
    deadline_checks = {1001: 0, 1002: 0}

    def group_alive(pgid: int) -> bool:
        if harness.clock.now < 10.0:
            return True
        deadline_checks[pgid] += 1
        return deadline_checks[pgid] == 1

    harness.manager._group_alive = group_alive
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        if not triggered:
            triggered = True
            harness.manager._handle_signal(signal.SIGTERM, None)
        harness.clock.sleep(delay)

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert deadline_checks == {1001: 2, 1002: 2}
    assert not any(
        event[0] == "signal" and event[2] == signal.SIGKILL
        for event in harness.events
    )
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


@pytest.mark.parametrize("incoming_signal", [signal.SIGTERM, signal.SIGINT])
def test_signal_shutdown_forwards_sigterm_and_returns_zero(
    supervisor, tmp_path: Path, monkeypatch, incoming_signal: int
) -> None:
    harness = manager_harness(supervisor, tmp_path)
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        if not triggered:
            triggered = True
            harness.manager._handle_signal(incoming_signal, None)
        harness.clock.sleep(delay)

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert {(event[1], event[2]) for event in harness.events if event[0] == "signal"} == {
        (1001, signal.SIGTERM),
        (1002, signal.SIGTERM),
    }
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_signal_during_cloak_readiness_never_starts_stelloauth(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path)

    def cloak_probe() -> None:
        harness.manager._handle_signal(signal.SIGTERM, None)

    harness.manager._cloak_probe = cloak_probe
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert len(harness.spawned) == 1
    assert ("signal", 1001, signal.SIGTERM) in harness.events
    assert harness.processes[0].wait_calls == [None]


def test_child_exit_during_cloak_readiness_never_starts_stelloauth(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path)

    def cloak_probe() -> None:
        harness.processes[0].returncode = 7

    harness.manager._cloak_probe = cloak_probe
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 7
    assert len(harness.spawned) == 1
    assert harness.processes[0].wait_calls == [None]


def test_signal_while_starting_child_still_terminates_new_process_group(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path)
    original_popen = harness.manager._popen
    starts = 0

    def popen(command, *, env, start_new_session):
        nonlocal starts
        starts += 1
        if starts == 2:
            harness.manager._handle_signal(signal.SIGTERM, None)
        return original_popen(command, env=env, start_new_session=start_new_session)

    harness.manager._popen = popen
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert ("signal", 1002, signal.SIGTERM) in harness.events
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_shutdown_uses_one_ten_second_deadline_then_sigkills_remaining_groups(
    supervisor, tmp_path: Path, monkeypatch
) -> None:
    harness = manager_harness(supervisor, tmp_path, ignores_term=(True, True))
    triggered = False

    def sleep(delay: float) -> None:
        nonlocal triggered
        if not triggered:
            triggered = True
            harness.manager._handle_signal(signal.SIGTERM, None)
        harness.clock.sleep(delay)

    harness.manager._sleep = sleep
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    assert harness.manager.run() == 0
    assert harness.clock.now == pytest.approx(10.0)
    assert [
        (event[1], event[2]) for event in harness.events if event[0] == "signal"
    ] == [
        (1001, signal.SIGTERM),
        (1002, signal.SIGTERM),
        (1001, signal.SIGKILL),
        (1002, signal.SIGKILL),
    ]
    assert [process.wait_calls for process in harness.processes] == [[None], [None]]


def test_lifecycle_logs_are_fixed_and_contain_no_sensitive_values(
    supervisor, tmp_path: Path, monkeypatch, caplog
) -> None:
    harness = manager_harness(supervisor, tmp_path)

    def stelloauth_probe() -> None:
        harness.manager._handle_signal(signal.SIGTERM, None)

    harness.manager._stelloauth_probe = stelloauth_probe
    monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None)

    with caplog.at_level(logging.INFO):
        assert harness.manager.run() == 0

    assert caplog.messages == [
        "Cleaning CloakBrowser profiles",
        "Starting CloakBrowser",
        "CloakBrowser ready",
        "Starting Stelloauth",
        "Shutdown requested",
        "Stopping child processes",
        "Child processes stopped",
    ]
    lowered = caplog.text.lower()
    for sensitive in ("sentinel", "credential", "code", "cookie", "token"):
        assert sensitive not in lowered