diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index 5b44c17..98ecbad 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -21,6 +21,7 @@ from gatelock import ( from wake_alarm._alarm_display import _restore_display from wake_alarm._alarm_flash import AlarmFlashMixin +from wake_alarm._alarm_session import finish_session, start_session from wake_alarm._alarm_state import _AlarmHardware, _AlarmProgress from wake_alarm._alarm_timers import AlarmTimersMixin from wake_alarm._alarm_view import _build_alarm_view, _ViewGroup @@ -29,8 +30,13 @@ from wake_alarm._challenges import ( _make_challenge, ) from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED +from wake_alarm._correlation import correlation_id, correlation_id_for_today from wake_alarm._escape_flow import EscapeHatch, build_tracker from wake_alarm._gating import parse_args, should_run_alarm +from wake_alarm._inhibit import ( + release_power_key_inhibitor, + take_power_key_inhibitor, +) from wake_alarm._lock_setup import build_config, claim_arbiter from wake_alarm._logging_setup import configure_logging from wake_alarm._reactivate import start_alarm_reactivation @@ -58,11 +64,16 @@ _logger = logging.getLogger(__name__) class WakeAlarm(AlarmTimersMixin, AlarmFlashMixin): """Fullscreen wake alarm with escalating beep and dismiss challenge.""" - def __init__(self, *, demo_mode: bool = False) -> None: + def __init__( + self, *, demo_mode: bool = False, correlation: str | None = None + ) -> None: """Initialize the wake alarm. Args: demo_mode: If True, use a smaller window and shorter timers. + correlation: The morning's correlation id, derived from the + *resolved alarm instant* by the caller. Defaults to one + derived from now, which is only right for a demo run. """ self.demo_mode = demo_mode self.dismissed = False @@ -91,7 +102,15 @@ class WakeAlarm(AlarmTimersMixin, AlarmFlashMixin): self._schedule_code_refresh() self._schedule_skip_window_close() start_beep_thread(self._stop_beep, self._progress.started_at) - self._hardware = _AlarmHardware(audio_restore=_activate_all_outputs()) + # Taken while the alarm is audible, released in on_close(). A short + # power-button press would otherwise end the alarm in one tap. + inhibitor = take_power_key_inhibitor() + # Written before anything can go wrong: a power cut from here on + # leaves a "ringing" record for _trigger to re-fire from on boot. + start_session(correlation or correlation_id(datetime.now().astimezone())) + self._hardware = _AlarmHardware( + audio_restore=_activate_all_outputs(), inhibitor=inhibitor + ) # The first activation above runs before the monitor has finished its # HDMI handshake, so the sink that matters most may not exist yet. # Keep retrying in the background until the alarm is dismissed -- only @@ -145,6 +164,11 @@ class WakeAlarm(AlarmTimersMixin, AlarmFlashMixin): def on_close(self) -> None: """Restore audio/display/plug; runs on every exit path, even SIGTERM.""" self._stop_beep.set() + # Closing the session here, not in _dismiss_alarm, so the escape hatch + # and a SIGTERM close it too. The one path that cannot reach this is + # the power cut it exists to catch. + finish_session() + release_power_key_inhibitor(self._hardware.inhibitor) _restore_audio_state(self._hardware.audio_restore) _restore_display() turn_off_plug() @@ -217,7 +241,7 @@ def main() -> None: args.trigger_now, ) prepare_hardware() - alarm = WakeAlarm(demo_mode=args.demo) + alarm = WakeAlarm(demo_mode=args.demo, correlation=correlation_id_for_today()) alarm.run() diff --git a/wake_alarm/_alarm_session.py b/wake_alarm/_alarm_session.py new file mode 100644 index 0000000..7f3948d --- /dev/null +++ b/wake_alarm/_alarm_session.py @@ -0,0 +1,227 @@ +"""The fire-time record that makes a hard power-off survivable. + +The real gap behind item 2 is not the power button itself -- it is that +nothing on disk ever says *"an alarm is ringing right now and was never +dismissed"*. ``wake_state.json`` is written only on **dismissal**, so cutting +power mid-alarm leaves no trace at all: the next boot sees an undismissed day +that looks exactly like a day the alarm had not reached yet, and the catch-up +window has long since closed. Holding the power button therefore *works*, and +that is what this module removes. + +So the alarm writes a record the moment it starts ringing, and clears it when +the ring genuinely ends. ``_trigger`` re-fires on a record still marked +``ringing``, ignoring the catch-up window. + +**Deliberately no date field.** ``_state.py`` keys its record by a UTC date, +and ``screen_locker/_wake_state.py`` mirrors that helper to read the very same +file from a different repo -- so the UTC basis there is load-bearing and is not +touched. Rather than add a second date key on a different basis, this record +carries only ``fired_at`` as an aware ISO timestamp, and freshness is an +elapsed-time comparison. Timezone-agnostic by construction, so the local +midnight question never arises. + +**Bounded, because an unfinished session is not self-clearing.** If the alarm +process dies before dismissal -- Tk cannot open a display, a crash at startup +-- nothing ever runs ``on_close()`` to mark the session finished, and an +unbounded re-fire rule would restart the alarm every single minute forever. +The record is therefore honoured only within ``REFIRE_MAX_AGE_SECONDS`` of +``fired_at``, and the ordinary day / dismissed / workout gates still apply on +top of it. + +Signed with :mod:`gatelock.log_integrity`, exactly like ``_state.py``, so the +record cannot simply be edited to say ``dismissed``. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json +import logging + +from gatelock.log_integrity import compute_entry_hmac, verify_entry_hmac + +from wake_alarm._constants import ( + ALARM_SESSION_FILE, + CLOCK_SKEW_TOLERANCE_SECONDS, + REFIRE_MAX_AGE_SECONDS, +) + +_logger = logging.getLogger(__name__) + +STATE_RINGING = "ringing" +STATE_FINISHED = "finished" + + +def _now() -> datetime: + """Return the current instant, aware and local. The seam tests patch.""" + return datetime.now(tz=timezone.utc).astimezone() + + +def _write(entry: dict[str, object]) -> bool: + """Sign and write *entry*, reporting whether it landed on disk.""" + signature = compute_entry_hmac(entry) + if signature is not None: + entry["hmac"] = signature + else: + _logger.warning("HMAC key unavailable — saving unsigned alarm session") + try: + ALARM_SESSION_FILE.parent.mkdir(parents=True, exist_ok=True) + with ALARM_SESSION_FILE.open("w") as handle: + json.dump(entry, handle, indent=2) + except OSError as exc: + _logger.warning("Failed to write the alarm session record: %s", exc) + return False + return True + + +def start_session(correlation: str) -> bool: + """Record that an alarm has started ringing. + + Args: + correlation: The derived correlation id for this morning, so the + record joins up with the JSONL log without a handshake. + + Returns: + True when the record was written. + """ + written = _write( + { + "state": STATE_RINGING, + "fired_at": _now().isoformat(), + "correlation": correlation, + } + ) + if written: + _logger.warning( + "Alarm session opened (%s) — a power-off now re-fires on boot", + correlation, + extra={"event": "session_opened", "correlation": correlation}, + ) + return written + + +def finish_session() -> bool: + """Mark the current session finished, however the alarm ended. + + Called from ``on_close()``, so it covers dismissal, the escape hatch and + SIGTERM alike -- every path where the ring genuinely stopped with the + machine still up. The one path it cannot cover is the one it exists for: + power cut mid-alarm, which leaves the record saying ``ringing``. + + Returns: + True when the record was written. + """ + entry = load_session() + if entry is None: + return False + return _write( + { + "state": STATE_FINISHED, + "fired_at": entry.get("fired_at"), + "correlation": entry.get("correlation"), + } + ) + + +def load_session() -> dict[str, object] | None: + """Load and verify the session record. + + Returns: + The record, or None when it is absent, unreadable, or fails its HMAC + check. A tampered record is treated as no record: it must not be + possible to *create* a ringing session by hand either. + """ + if not ALARM_SESSION_FILE.exists(): + return None + try: + with ALARM_SESSION_FILE.open() as handle: + entry = json.load(handle) + except (OSError, json.JSONDecodeError): + _logger.warning("Cannot read the alarm session record") + return None + if not isinstance(entry, dict): + return None + if not verify_entry_hmac(entry): + _logger.warning("Alarm session HMAC verification failed") + return None + return entry + + +def _fired_at(entry: dict[str, object]) -> datetime | None: + """Parse the record's ``fired_at``, or None when it is unusable.""" + raw = entry.get("fired_at") + if not isinstance(raw, str): + return None + try: + parsed = datetime.fromisoformat(raw) + except ValueError: + return None + if parsed.tzinfo is None: + return None + return parsed + + +def unfinished_session_age(*, now: datetime | None = None) -> float | None: + """Return how long an unfinished alarm has been ringing, in seconds. + + Args: + now: Instant to measure against; defaults to the current local time. + + Returns: + The elapsed seconds when a ringing session is on disk and its + timestamp is usable, else None. A slightly future-stamped record is + clamped to zero rather than discarded -- see below. + """ + entry = load_session() + if entry is None or entry.get("state") != STATE_RINGING: + return None + fired_at = _fired_at(entry) + if fired_at is None: + _logger.warning("Alarm session has no usable fired_at; ignoring it") + return None + age = ((now if now is not None else _now()) - fired_at).total_seconds() + if age >= 0: + return age + # A record stamped in the future means the clock moved *backwards* since + # it was written -- NTP correcting after a resume, or RTC-vs-system skew + # at boot. This machine wakes from hibernate, so that is a routine event, + # not a broken one. Treating it as "not ringing" would fail open on + # exactly the scenario this record exists to close: alarm fires on a fast + # clock, the clock is corrected, and the power-off escape works again. + # So clamp to zero and let it re-fire. + if age < -CLOCK_SKEW_TOLERANCE_SECONDS: + _logger.warning( + "Alarm session is stamped %.0fs in the future — beyond the " + "clock-skew tolerance, ignoring it", + -age, + extra={"event": "session_clock_skew", "skew_seconds": -age}, + ) + return None + _logger.info("Alarm session is %.0fs in the future; treating it as fresh", -age) + return 0.0 + + +def should_refire(*, now: datetime | None = None) -> bool: + """Report whether an alarm was cut short and must ring again. + + Bounded on purpose: an alarm process that died before dismissal leaves a + record nothing will ever clear, so without the age limit this would + restart the alarm every minute indefinitely. + + Args: + now: Instant to measure against; defaults to the current local time. + + Returns: + True when a ringing session exists and is younger than the limit. + """ + age = unfinished_session_age(now=now) + if age is None: + return False + if age > REFIRE_MAX_AGE_SECONDS: + _logger.info( + "Ignoring an unfinished alarm session %.0fs old (limit %.0fs)", + age, + REFIRE_MAX_AGE_SECONDS, + ) + return False + return True diff --git a/wake_alarm/_alarm_state.py b/wake_alarm/_alarm_state.py index 4ce0b84..a696d2b 100644 --- a/wake_alarm/_alarm_state.py +++ b/wake_alarm/_alarm_state.py @@ -11,6 +11,8 @@ import time from typing import TYPE_CHECKING if TYPE_CHECKING: + import subprocess + from wake_alarm._challenges import _Challenge from wake_alarm._sinks import _AudioState @@ -29,6 +31,12 @@ class _AlarmProgress: @dataclass class _AlarmHardware: - """Hardware state captured at alarm start, restored when it closes.""" + """Machine state the alarm takes at start and gives back when it closes. + + Both fields are released together in ``on_close()``, so they live together: + the audio routing the alarm rewrote, and the logind inhibitor holding the + power button down while it rings. + """ audio_restore: _AudioState | None + inhibitor: subprocess.Popen[bytes] | None = None diff --git a/wake_alarm/_constants.py b/wake_alarm/_constants.py index c109b02..cbad70b 100644 --- a/wake_alarm/_constants.py +++ b/wake_alarm/_constants.py @@ -95,6 +95,43 @@ ESCAPE_OFFER_AFTER_SECONDS: int = 900 # of the alarm. Costs nothing when no plug is configured. PLUG_PREROLL_SECONDS: float = 4.0 +# The fire-time session record. Beside the sync config rather than in the +# package dir, so a reinstall cannot erase evidence that an alarm was cut +# short. Distinct from WAKE_STATE_FILE, which records only *dismissal* and is +# also read cross-repo by screen_locker/_wake_state.py. +ALARM_SESSION_FILE: Path = Path.home() / ".config" / "wake_alarm" / "alarm_session.json" + +# How long after fire time an unfinished session still re-fires the alarm. +# Bounded because nothing clears the record when the alarm process dies before +# dismissal: unbounded, a crash at startup would restart the alarm every minute +# forever. Long enough to survive a power-off plus a boot, short enough that a +# session can never leak into the following night. +REFIRE_MAX_AGE_SECONDS: float = 5400.0 + +# How far a session record may be stamped in the *future* and still count as +# ringing. A backward clock jump (NTP correcting after a resume, RTC skew at +# boot) is routine on a machine that hibernates nightly, and discarding such a +# record would fail open on exactly the power-off escape the record closes. +# Bounded so a wildly wrong RTC still cannot create an endless re-fire. +CLOCK_SKEW_TOLERANCE_SECONDS: float = 900.0 + +# --- Power-button inhibitor (item 2) -------------------------------------- +# What logind is asked to block while the alarm rings. handle-power-key stops +# the short press; shutdown and idle stop a scripted poweroff and the idle +# timer from doing the same job by another route. +INHIBIT_WHAT: str = "handle-power-key:shutdown:idle" + +# The --who string. Also the pkill pattern used to clear a stale blocker from a +# crashed alarm, so it must stay distinctive. +INHIBIT_WHO: str = "wake-alarm" +INHIBIT_WHY: str = "Wake alarm is ringing and must be dismissed, not switched off" + +# The inhibitor child self-terminates after this long even if nobody releases +# it. An orphaned block-mode inhibitor holding handle-power-key would leave the +# power button dead until reboot -- a worse failure than the one being fixed -- +# so it is never `sleep infinity`. Comfortably longer than any real alarm. +INHIBIT_MAX_SECONDS: float = 7200.0 + # When the failsafe silences the alarm. Audio ONLY: the lock, the typed # challenge and the workout gate all persist past this point, so waiting it # out in bed is not a way to escape the alarm. It exists so a ring nobody is diff --git a/wake_alarm/_correlation.py b/wake_alarm/_correlation.py index 47f0bb3..b88b273 100644 --- a/wake_alarm/_correlation.py +++ b/wake_alarm/_correlation.py @@ -12,10 +12,9 @@ UTC, which is a separate known mismatch tracked in the hardening plan). from __future__ import annotations -from typing import TYPE_CHECKING +from datetime import datetime -if TYPE_CHECKING: - from datetime import datetime +from wake_alarm._sync import get_alarm # ``wake-YYYYMMDD-HHMM``. Stable across processes and across devices. _CORRELATION_FORMAT = "wake-%Y%m%d-%H%M" @@ -49,3 +48,20 @@ def correlation_id_for(hour: int, minute: int, *, now: datetime) -> str: The correlation id string. """ return correlation_id(now.replace(hour=hour, minute=minute)) + + +def correlation_id_for_today() -> str: + """Derive this morning's correlation id from the *resolved alarm time*. + + Not from "now": the alarm can start up to a full catch-up window late, and + a re-fire after a power-off can start hours later still. The phone derives + its id from the alarm time it was set to, with no handshake between the + two, so deriving the PC's from the start instant would silently stop the + two log streams joining -- ``wake-20260827-0735`` against + ``wake-20260827-0700``. + + Returns: + The correlation id both halves compute independently. + """ + alarm = get_alarm() + return correlation_id_for(alarm.hour, alarm.minute, now=datetime.now().astimezone()) diff --git a/wake_alarm/_inhibit.py b/wake_alarm/_inhibit.py new file mode 100644 index 0000000..ae702c9 --- /dev/null +++ b/wake_alarm/_inhibit.py @@ -0,0 +1,191 @@ +"""Hold a logind inhibitor so one power-button press cannot kill the alarm. + +Item 2 of the hardening plan. Today a short press on the power button runs +``HandlePowerKey=poweroff`` and the alarm is gone in one tap -- less effort +than typing a single challenge round. + +Verified on this host before writing any of this: + + $ grep PowerKey /etc/systemd/logind.conf + #HandlePowerKey=poweroff + #HandlePowerKeyLongPress=ignore + #PowerKeyIgnoreInhibited=no + +``PowerKeyIgnoreInhibited=no`` is the default and is in force, which is exactly +what makes a ``--mode=block`` inhibitor suppress the short press. So no +``logind.conf`` edit is needed, and none is made: a drop-in would change the +machine permanently, while an inhibitor is scoped to the alarm's lifetime. + +**What this cannot block, stated plainly:** a ~4 second firmware hold cuts +power below the operating system, and so does the PSU switch. Neither is +defeatable from userspace. The durable defence against those is the fire-time +session record in :mod:`wake_alarm._alarm_session` -- the alarm re-fires on the +next boot rather than being escaped -- not this inhibitor. + +**Why a bounded ``sleep`` and not ``sleep infinity``:** the inhibitor is a +child process. If the alarm dies without running ``on_close()`` -- a crash, a +SIGKILL -- an orphaned blocker holding ``handle-power-key`` would leave the +power button dead until the next reboot, which is a far worse failure than the +one being fixed. The child therefore exits on its own after +``INHIBIT_MAX_SECONDS`` even if nobody ever releases it, and a stale one from a +previous run is killed at startup -- by pid, via ``/proc``, never by matching a +command-line substring (see :func:`_is_our_inhibitor` for why). +""" + +from __future__ import annotations + +import logging +import os +import pathlib +import shutil +import signal +import subprocess + +from wake_alarm._constants import ( + INHIBIT_MAX_SECONDS, + INHIBIT_WHAT, + INHIBIT_WHO, + INHIBIT_WHY, +) + +_logger = logging.getLogger(__name__) + + +def _systemd_inhibit_path() -> str | None: + """Return the absolute path to systemd-inhibit, or None when absent.""" + return shutil.which("systemd-inhibit") + + +def _is_our_inhibitor(pid: str) -> bool: + """Report whether *pid* is one of this alarm's inhibitor children. + + Reads the process's own argv rather than trusting a pattern match. This + matters: an earlier version used ``pkill -f "--who=wake-alarm"``, which on + a real run matched the *shell that had just typed that pattern* and killed + it instead of the inhibitor. Matching a command line by substring is + inherently self-matching; asking a specific pid what it is, is not. + """ + try: + argv = pathlib.Path("/proc", pid, "cmdline").read_bytes() + except OSError: + # The process exited between listing and reading -- it is not stale, + # it is simply gone. + return False + return f"--who={INHIBIT_WHO}".encode() in argv + + +def kill_stale_inhibitors() -> int: + """Kill any inhibitor left behind by an earlier alarm process. + + A blocker that outlived its alarm holds the power button down for every + later session, so this runs before a new one is taken rather than trusting + that the previous process exited cleanly. + + Matches on the exact process name and then confirms via ``/proc``, so it + can only ever kill a real ``systemd-inhibit`` of ours. + + Returns: + The number of stale processes signalled. + """ + pgrep = shutil.which("pgrep") + if pgrep is None: + return 0 + try: + listed = subprocess.run( + [pgrep, "-x", "systemd-inhibit"], + capture_output=True, + timeout=5, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("Could not check for a stale power-key inhibitor") + return 0 + killed = 0 + for pid in listed.stdout.decode(errors="replace").split(): + if not _is_our_inhibitor(pid): + continue + try: + os.kill(int(pid), signal.SIGTERM) + except (OSError, ValueError): + continue + killed += 1 + if killed: + _logger.warning( + "Killed %d stale power-key inhibitor(s) left by an earlier alarm", + killed, + extra={"event": "inhibitor_stale_killed", "count": killed}, + ) + return killed + + +def take_power_key_inhibitor() -> subprocess.Popen[bytes] | None: + """Block the power key for the alarm's lifetime. + + Returns: + The inhibitor child process, or None when systemd-inhibit is missing + or could not be started. A missing inhibitor never stops the alarm: + being loud without the power key blocked beats not ringing at all. + """ + inhibit = _systemd_inhibit_path() + if inhibit is None: + _logger.warning( + "systemd-inhibit not on PATH; the power button can still kill " + "the alarm in one press", + extra={"event": "inhibitor_unavailable"}, + ) + return None + kill_stale_inhibitors() + cmd = [ + inhibit, + f"--what={INHIBIT_WHAT}", + "--mode=block", + f"--who={INHIBIT_WHO}", + f"--why={INHIBIT_WHY}", + "sleep", + str(int(INHIBIT_MAX_SECONDS)), + ] + try: + # Deliberately not a context manager: this child must outlive the + # call and is reaped by release_power_key_inhibitor(). + process = subprocess.Popen( + cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except OSError: + _logger.warning( + "Could not take the power-key inhibitor; the alarm rings unprotected", + exc_info=True, + extra={"event": "inhibitor_failed"}, + ) + return None + _logger.warning( + "Power key inhibited for up to %ds (pid %d) — a short press no longer " + "kills the alarm; a ~4s firmware hold still cuts power", + int(INHIBIT_MAX_SECONDS), + process.pid, + extra={"event": "inhibitor_taken", "pid": process.pid}, + ) + return process + + +def release_power_key_inhibitor(process: subprocess.Popen[bytes] | None) -> None: + """Release the inhibitor taken by :func:`take_power_key_inhibitor`. + + Safe to call more than once and safe to call with None, because it runs + from ``on_close()``, which itself runs on every exit path. + + Args: + process: The inhibitor child, or None when one was never taken. + """ + if process is None: + return + if process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + _logger.warning("Power-key inhibitor ignored SIGTERM; killed it") + _logger.info("Power key released", extra={"event": "inhibitor_released"}) diff --git a/wake_alarm/_trigger.py b/wake_alarm/_trigger.py index ec618bf..aef6cf6 100644 --- a/wake_alarm/_trigger.py +++ b/wake_alarm/_trigger.py @@ -26,6 +26,7 @@ import logging import subprocess import sys +from wake_alarm._alarm_session import should_refire from wake_alarm._constants import ( ALARM_DAYS, TRIGGER_CATCHUP_MINUTES, @@ -138,6 +139,15 @@ def _decide(now: datetime, alarm: Alarm) -> tuple[bool, str]: if skip is not None: return False, skip + # An alarm that was cut short -- almost always a power-button press or a + # hard power-off -- left a "ringing" session record that nothing cleared. + # It re-fires regardless of the catch-up window, because the whole point is + # that the machine may have been off for the rest of that window. The day / + # dismissed / workout gates above still apply, and the record itself is + # age-bounded, so this cannot loop forever on a crashed alarm. + if should_refire(now=now): + return True, "an earlier alarm was cut short and never dismissed" + fire_at, window_end = _fire_window(alarm, now) if now < fire_at: return False, f"alarm {alarm} is still ahead" diff --git a/wake_alarm/tests/conftest.py b/wake_alarm/tests/conftest.py index 3462e3b..c09d32f 100644 --- a/wake_alarm/tests/conftest.py +++ b/wake_alarm/tests/conftest.py @@ -12,8 +12,14 @@ output. Two consequences for tests, both of which bite silently: Sink reactivation adds a fourth: ``WakeAlarm.__init__`` starts a background thread that really shells out to ``pactl set-card-profile`` and would reshape -the *host's* audio routing mid-test. It is stubbed here and re-patched -explicitly by the tests that assert the wiring. +the *host's* audio routing mid-test. + +The power-key inhibitor and the session record add a fifth and a sixth: an +unstubbed run really spawns ``systemd-inhibit`` (blocking the machine's power +button for as long as the test process lives) and really writes the +*production* ``alarm_session.json``, which the next real trigger tick would +then act on. All three are stubbed here and re-patched explicitly by the tests +that assert the wiring. The escape hatch adds a third: ``WakeAlarm.__init__`` builds its tracker over the real ``ESCAPE_HISTORY_FILE``, so an unpatched run both reads a *production* @@ -93,6 +99,10 @@ def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None tmp_path / "escape_history.json", ), patch("wake_alarm._alarm.start_alarm_reactivation"), + patch("wake_alarm._alarm.take_power_key_inhibitor", return_value=None), + patch("wake_alarm._alarm.release_power_key_inhibitor"), + patch("wake_alarm._alarm.start_session"), + patch("wake_alarm._alarm.finish_session"), patch("gatelock._surfaces.tk.Toplevel", side_effect=make_toplevel), patch("gatelock._outputs.RandrBackend.create", return_value=None), patch("gatelock._outputs.scan_xrandr", return_value=FAKE_OUTPUTS), diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index 27fd232..2153e36 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -128,6 +128,32 @@ class TestWakeAlarmInit: alarm._stop_beep.set() assert start.call_args.args[0].is_set() + def test_init_takes_the_power_key_inhibitor( + self, + mock_tk_module: MagicMock, + ) -> None: + """A short power-button press must not end the alarm in one tap.""" + assert mock_tk_module is not None + with patch("wake_alarm._alarm.take_power_key_inhibitor") as take: + alarm = WakeAlarm(demo_mode=True) + take.assert_called_once_with() + assert alarm._hardware.inhibitor is take.return_value + alarm._stop_beep.set() + + def test_init_opens_a_session_record( + self, + mock_tk_module: MagicMock, + ) -> None: + """Written while ringing, so a power cut leaves a trace to re-fire on.""" + assert mock_tk_module is not None + with patch("wake_alarm._alarm.start_session") as start: + alarm = WakeAlarm(demo_mode=True) + start.assert_called_once() + # The correlation id joins this record to the JSONL log with no + # handshake -- both sides derive it from the same alarm instant. + assert start.call_args.args[0].startswith("wake-") + alarm._stop_beep.set() + def test_production_mode_fullscreen( self, mock_tk_module: MagicMock, @@ -142,99 +168,3 @@ class TestWakeAlarmInit: # it applies it snaps to one monitor or the whole bounding box. assert alarm._view._views, "a surface must exist per live output" alarm._stop_beep.set() - - -class TestCodeRefreshAndTimer: - """Tests for code refresh and timer update methods.""" - - def test_code_refresh_changes_challenge( - self, - mock_tk_module: MagicMock, - ) -> None: - """Code refresh generates a new challenge each call.""" - alarm = WakeAlarm(demo_mode=True) - displays = set() - for _ in range(50): - alarm._schedule_code_refresh() - displays.add(alarm._progress.current_challenge.display) - assert len(displays) > 1 - alarm._stop_beep.set() - - def test_code_refresh_noop_when_not_active( - self, - mock_tk_module: MagicMock, - ) -> None: - """Code refresh is a no-op when alarm is no longer active.""" - alarm = WakeAlarm(demo_mode=True) - alarm._active = False - old_challenge = alarm._progress.current_challenge - alarm._schedule_code_refresh() - assert alarm._progress.current_challenge is old_challenge - alarm._stop_beep.set() - - def test_update_timer_noop_when_not_active( - self, - mock_tk_module: MagicMock, - ) -> None: - """Timer update is a no-op when alarm is inactive.""" - alarm = WakeAlarm(demo_mode=True) - alarm._active = False - alarm._update_timer() # Should not raise - alarm._stop_beep.set() - - -class TestScreenFlash: - """Tests for _start_screen_flash and _flash_step.""" - - def test_flash_step_shows_dark_on_flash_off( - self, - mock_tk_module: MagicMock, - ) -> None: - """When _flash_on=False, the background is set to dark colour.""" - alarm = WakeAlarm(demo_mode=True) - mock_root = mock_tk_module.Tk.return_value - mock_root.configure.reset_mock() - mock_root.after.reset_mock() - - alarm._progress.flash_on = False - alarm._flash_step() - - mock_root.configure.assert_called_once_with(bg=alarm._view.config.bg) - assert alarm._progress.flash_on is True - mock_root.after.assert_called_with(750, alarm._flash_step) - alarm._stop_beep.set() - - def test_flash_step_shows_red_on_flash_on( - self, - mock_tk_module: MagicMock, - ) -> None: - """When _flash_on=True, the background is set to red.""" - alarm = WakeAlarm(demo_mode=True) - mock_root = mock_tk_module.Tk.return_value - mock_root.configure.reset_mock() - mock_root.after.reset_mock() - - alarm._progress.flash_on = True - alarm._flash_step() - - mock_root.configure.assert_called_once_with(bg=alarm._view.config.danger) - assert alarm._progress.flash_on is False - mock_root.after.assert_called_with(750, alarm._flash_step) - alarm._stop_beep.set() - - def test_flash_step_stops_when_inactive( - self, - mock_tk_module: MagicMock, - ) -> None: - """When alarm is no longer active, _flash_step returns without scheduling.""" - alarm = WakeAlarm(demo_mode=True) - mock_root = mock_tk_module.Tk.return_value - alarm._active = False - mock_root.configure.reset_mock() - mock_root.after.reset_mock() - - alarm._flash_step() - - mock_root.configure.assert_not_called() - mock_root.after.assert_not_called() - alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_part4.py b/wake_alarm/tests/test_alarm_part4.py index ca6238d..c9f60c7 100644 --- a/wake_alarm/tests/test_alarm_part4.py +++ b/wake_alarm/tests/test_alarm_part4.py @@ -153,6 +153,29 @@ class TestClose: mock_restore.assert_called_once_with(state) alarm._stop_beep.set() + def test_on_close_releases_the_inhibitor_and_closes_the_session( + self, + mock_tk_module: MagicMock, + ) -> None: + """Both are released here, not in _dismiss_alarm. + + on_close() is the one path that runs on *every* exit -- dismissal, the + escape hatch and SIGTERM alike -- so an alarm killed by systemd still + gives the power button back and still marks its session finished. + """ + del mock_tk_module + alarm = WakeAlarm(demo_mode=True) + sentinel = object() + alarm._hardware.inhibitor = sentinel + with ( + patch("wake_alarm._alarm.release_power_key_inhibitor") as release, + patch("wake_alarm._alarm.finish_session") as finish, + ): + alarm.on_close() + release.assert_called_once_with(sentinel) + finish.assert_called_once_with() + alarm._stop_beep.set() + class TestMultiMonitorSurfaces: """Per-output behaviour added in the gatelock v0.2.0 migration.""" diff --git a/wake_alarm/tests/test_alarm_session.py b/wake_alarm/tests/test_alarm_session.py new file mode 100644 index 0000000..c2340d0 --- /dev/null +++ b/wake_alarm/tests/test_alarm_session.py @@ -0,0 +1,230 @@ +"""Tests for the fire-time session record that survives a hard power-off.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import json +from typing import TYPE_CHECKING + +import pytest + +from wake_alarm import _alarm_session +from wake_alarm._alarm_session import ( + STATE_FINISHED, + STATE_RINGING, + finish_session, + load_session, + should_refire, + start_session, + unfinished_session_age, +) + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture(autouse=True) +def session_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Point the record at a temporary file, never the real one.""" + path = tmp_path / "alarm_session.json" + monkeypatch.setattr(_alarm_session, "ALARM_SESSION_FILE", path) + return path + + +class TestStartSession: + """Opening a record the moment the alarm starts ringing.""" + + def test_writes_a_signed_ringing_record(self, session_file: Path) -> None: + """The record exists before anything can go wrong.""" + assert start_session("wake-20260827-0700") is True + entry = json.loads(session_file.read_text()) + assert entry["state"] == STATE_RINGING + assert entry["correlation"] == "wake-20260827-0700" + assert "hmac" in entry + # An aware timestamp: freshness is measured as elapsed time, so the + # offset has to be carried rather than assumed. + assert datetime.fromisoformat(str(entry["fired_at"])).tzinfo is not None + + def test_creates_the_parent_directory( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A fresh install has no ~/.config/wake_alarm yet.""" + monkeypatch.setattr( + _alarm_session, "ALARM_SESSION_FILE", tmp_path / "new" / "s.json" + ) + assert start_session("wake-20260827-0700") is True + + def test_unwritable_path_is_survivable( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A failed write must never stop the alarm from ringing.""" + blocker = tmp_path / "blocker" + blocker.write_text("not a directory") + monkeypatch.setattr(_alarm_session, "ALARM_SESSION_FILE", blocker / "s.json") + with caplog.at_level("WARNING"): + assert start_session("wake-20260827-0700") is False + assert "Failed to write" in caplog.text + + def test_saves_unsigned_when_no_hmac_key( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """An unsigned record still beats no record at all.""" + monkeypatch.setattr(_alarm_session, "compute_entry_hmac", lambda _entry: None) + with caplog.at_level("WARNING"): + assert start_session("wake-20260827-0700") is True + assert "unsigned alarm session" in caplog.text + + +class TestFinishSession: + """Closing the record however the alarm actually ended.""" + + def test_marks_the_record_finished(self, session_file: Path) -> None: + """Dismissal, escape hatch and SIGTERM all land here.""" + start_session("wake-20260827-0700") + assert finish_session() is True + entry = json.loads(session_file.read_text()) + assert entry["state"] == STATE_FINISHED + # fired_at is preserved: it is the audit trail of when it rang. + assert entry["fired_at"] is not None + assert entry["correlation"] == "wake-20260827-0700" + + def test_no_record_to_finish(self) -> None: + """on_close() runs even when no session was ever opened.""" + assert finish_session() is False + + +class TestLoadSession: + """Reading the record back, rejecting anything untrustworthy.""" + + def test_absent_file(self) -> None: + """No record is not an error.""" + assert load_session() is None + + def test_unreadable_json( + self, session_file: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A truncated write is treated as no record.""" + session_file.write_text("{not json") + with caplog.at_level("WARNING"): + assert load_session() is None + assert "Cannot read" in caplog.text + + def test_non_dict_payload(self, session_file: Path) -> None: + """A JSON list is not a record.""" + session_file.write_text("[1, 2, 3]") + assert load_session() is None + + def test_tampered_record_is_rejected( + self, session_file: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """Hand-editing state must not be able to fake -- or force -- a ring.""" + start_session("wake-20260827-0700") + entry = json.loads(session_file.read_text()) + entry["state"] = STATE_FINISHED + session_file.write_text(json.dumps(entry)) + with caplog.at_level("WARNING"): + assert load_session() is None + assert "HMAC verification failed" in caplog.text + + +class TestShouldRefire: + """The decision _trigger acts on.""" + + def test_no_record_does_not_refire(self) -> None: + """The ordinary case: nothing was cut short.""" + assert should_refire() is False + + def test_ringing_record_refires(self) -> None: + """The whole point: a power-off mid-alarm rings again.""" + start_session("wake-20260827-0700") + assert should_refire() is True + + def test_finished_record_does_not_refire(self) -> None: + """A dismissed alarm stays dismissed.""" + start_session("wake-20260827-0700") + finish_session() + assert should_refire() is False + + def test_stale_record_does_not_refire( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Bounded on purpose. + + An alarm that died before dismissal leaves a record nothing will ever + clear; unbounded, this would restart the alarm every minute forever. + """ + start_session("wake-20260827-0700") + much_later = datetime.now(tz=timezone.utc).astimezone() + timedelta(days=1) + with caplog.at_level("INFO"): + assert should_refire(now=much_later) is False + assert "Ignoring an unfinished alarm session" in caplog.text + + def test_a_backward_clock_jump_still_refires( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Fails *closed*, which is the whole point of the record. + + NTP correcting after a resume moves the clock backwards, leaving the + record stamped in the future. Discarding it there would mean: alarm + fires on a fast clock, clock is corrected, and the power-off escape + quietly works again -- on a machine that hibernates nightly, i.e. the + exact scenario this record exists to close. + """ + start_session("wake-20260827-0700") + corrected = datetime.now(tz=timezone.utc).astimezone() - timedelta(minutes=5) + with caplog.at_level("INFO"): + assert should_refire(now=corrected) is True + assert "in the future" in caplog.text + + def test_a_wildly_future_record_is_ignored( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Bounded, so a broken RTC cannot create an endless re-fire.""" + start_session("wake-20260827-0700") + way_earlier = datetime.now(tz=timezone.utc).astimezone() - timedelta(hours=6) + with caplog.at_level("WARNING"): + assert should_refire(now=way_earlier) is False + assert "beyond the" in caplog.text + + def test_missing_timestamp_is_ignored( + self, + session_file: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A record with no usable fired_at cannot be aged, so it is dropped.""" + monkeypatch.setattr(_alarm_session, "verify_entry_hmac", lambda _e: True) + session_file.write_text(json.dumps({"state": STATE_RINGING})) + with caplog.at_level("WARNING"): + assert unfinished_session_age() is None + assert "no usable fired_at" in caplog.text + + @pytest.mark.parametrize( + "fired_at", + [ + "not-a-timestamp", + # Naive: without an offset the elapsed-time comparison is + # meaningless, which is exactly the ambiguity this record avoids. + "2026-08-27T07:00:00", + ], + ) + def test_unusable_timestamps_are_ignored( + self, session_file: Path, monkeypatch: pytest.MonkeyPatch, fired_at: str + ) -> None: + """Neither an unparsable nor a naive timestamp can drive a re-fire.""" + monkeypatch.setattr(_alarm_session, "verify_entry_hmac", lambda _e: True) + session_file.write_text( + json.dumps({"state": STATE_RINGING, "fired_at": fired_at}) + ) + assert should_refire() is False + + def test_age_is_reported_for_a_live_session(self) -> None: + """The age helper is what makes the bound observable.""" + start_session("wake-20260827-0700") + later = datetime.now(tz=timezone.utc).astimezone() + timedelta(seconds=90) + age = unfinished_session_age(now=later) + assert age is not None + assert 89 <= age <= 92 diff --git a/wake_alarm/tests/test_alarm_timers_flash.py b/wake_alarm/tests/test_alarm_timers_flash.py new file mode 100644 index 0000000..5a902ec --- /dev/null +++ b/wake_alarm/tests/test_alarm_timers_flash.py @@ -0,0 +1,192 @@ +"""Tests for _alarm.py — the timers and screen flash driven by root.after. + +Split from ``test_alarm_part2.py`` for the 250-line cap. The header -- +``_make_mock_tk``, ``_block_real_tk``, ``_block_extra_devices`` and +``mock_tk_module`` -- is copied whole rather than trimmed: those fixtures sit +*after* the imports, so a header-only copy silently drops them and every test +in the file errors. +""" + +from __future__ import annotations + +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +if TYPE_CHECKING: + from collections.abc import Generator + +from wake_alarm._alarm import ( + WakeAlarm, +) + +# --------------------------------------------------------------------------- +# Helpers (duplicated from part 1 so this file is self-contained) +# --------------------------------------------------------------------------- + + +def _make_mock_tk() -> MagicMock: + """Build a MagicMock that stands in for the tkinter module.""" + mock = MagicMock() + mock_root = MagicMock() + mock_root.winfo_screenwidth.return_value = 1920 + mock_root.winfo_screenheight.return_value = 1080 + mock.Tk.return_value = mock_root + mock.Frame.return_value = MagicMock() + mock.Label.return_value = MagicMock() + mock.Entry.return_value = MagicMock() + mock.TclError = tk.TclError + mock.END = tk.END + return mock + + +@pytest.fixture(autouse=True) +def _block_real_tk() -> Generator[MagicMock]: + """Prevent any real Tk windows in tests.""" + mock = _make_mock_tk() + with ( + patch("wake_alarm._alarm.tk", mock), + patch("wake_alarm._dismiss_view.tk", mock), + patch("wake_alarm._alarm_view.tk", mock), + patch( + "wake_alarm._alarm.GateRoot", + return_value=mock.Tk.return_value, + ), + ): + yield mock + + +@pytest.fixture(autouse=True) +def _block_extra_devices() -> Generator[MagicMock]: + """Prevent real subprocess.Popen calls for extra ALSA devices.""" + with ( + patch("wake_alarm._beep_loop._play_on_all_sinks") as mock, + patch("wake_alarm._startup._set_max_brightness"), + patch("wake_alarm._startup._wake_display"), + patch("wake_alarm._alarm._restore_display"), + patch("wake_alarm._startup._warn_if_no_real_sink"), + patch("wake_alarm._alarm._activate_all_outputs", return_value=None), + patch("wake_alarm._alarm._restore_audio_state"), + patch("wake_alarm._startup.turn_on_plug"), + patch("wake_alarm._alarm.turn_off_plug"), + ): + yield mock + + +@pytest.fixture +def mock_tk_module() -> Generator[MagicMock]: + """Provide explicit access to the mocked tk module.""" + mock = _make_mock_tk() + with ( + patch("wake_alarm._alarm.tk", mock), + patch("wake_alarm._dismiss_view.tk", mock), + patch("wake_alarm._alarm_view.tk", mock), + patch( + "wake_alarm._alarm.GateRoot", + return_value=mock.Tk.return_value, + ), + ): + yield mock + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestCodeRefreshAndTimer: + """Tests for code refresh and timer update methods.""" + + def test_code_refresh_changes_challenge( + self, + mock_tk_module: MagicMock, + ) -> None: + """Code refresh generates a new challenge each call.""" + alarm = WakeAlarm(demo_mode=True) + displays = set() + for _ in range(50): + alarm._schedule_code_refresh() + displays.add(alarm._progress.current_challenge.display) + assert len(displays) > 1 + alarm._stop_beep.set() + + def test_code_refresh_noop_when_not_active( + self, + mock_tk_module: MagicMock, + ) -> None: + """Code refresh is a no-op when alarm is no longer active.""" + alarm = WakeAlarm(demo_mode=True) + alarm._active = False + old_challenge = alarm._progress.current_challenge + alarm._schedule_code_refresh() + assert alarm._progress.current_challenge is old_challenge + alarm._stop_beep.set() + + def test_update_timer_noop_when_not_active( + self, + mock_tk_module: MagicMock, + ) -> None: + """Timer update is a no-op when alarm is inactive.""" + alarm = WakeAlarm(demo_mode=True) + alarm._active = False + alarm._update_timer() # Should not raise + alarm._stop_beep.set() + + +class TestScreenFlash: + """Tests for _start_screen_flash and _flash_step.""" + + def test_flash_step_shows_dark_on_flash_off( + self, + mock_tk_module: MagicMock, + ) -> None: + """When _flash_on=False, the background is set to dark colour.""" + alarm = WakeAlarm(demo_mode=True) + mock_root = mock_tk_module.Tk.return_value + mock_root.configure.reset_mock() + mock_root.after.reset_mock() + + alarm._progress.flash_on = False + alarm._flash_step() + + mock_root.configure.assert_called_once_with(bg=alarm._view.config.bg) + assert alarm._progress.flash_on is True + mock_root.after.assert_called_with(750, alarm._flash_step) + alarm._stop_beep.set() + + def test_flash_step_shows_red_on_flash_on( + self, + mock_tk_module: MagicMock, + ) -> None: + """When _flash_on=True, the background is set to red.""" + alarm = WakeAlarm(demo_mode=True) + mock_root = mock_tk_module.Tk.return_value + mock_root.configure.reset_mock() + mock_root.after.reset_mock() + + alarm._progress.flash_on = True + alarm._flash_step() + + mock_root.configure.assert_called_once_with(bg=alarm._view.config.danger) + assert alarm._progress.flash_on is False + mock_root.after.assert_called_with(750, alarm._flash_step) + alarm._stop_beep.set() + + def test_flash_step_stops_when_inactive( + self, + mock_tk_module: MagicMock, + ) -> None: + """When alarm is no longer active, _flash_step returns without scheduling.""" + alarm = WakeAlarm(demo_mode=True) + mock_root = mock_tk_module.Tk.return_value + alarm._active = False + mock_root.configure.reset_mock() + mock_root.after.reset_mock() + + alarm._flash_step() + + mock_root.configure.assert_not_called() + mock_root.after.assert_not_called() + alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_correlation.py b/wake_alarm/tests/test_correlation.py index 65ea7e5..f2b9fdd 100644 --- a/wake_alarm/tests/test_correlation.py +++ b/wake_alarm/tests/test_correlation.py @@ -3,8 +3,14 @@ from __future__ import annotations from datetime import datetime +from typing import TYPE_CHECKING +from wake_alarm import _correlation from wake_alarm._correlation import correlation_id, correlation_id_for +from wake_alarm._sync import Alarm + +if TYPE_CHECKING: + import pytest def _local(year: int, month: int, day: int, hour: int, minute: int) -> datetime: @@ -41,3 +47,23 @@ def test_correlation_id_for_ignores_the_time_of_now() -> None: early = correlation_id_for(6, 0, now=_local(2026, 8, 26, 1, 2)) late = correlation_id_for(6, 0, now=_local(2026, 8, 26, 23, 58)) assert early == late == "wake-20260826-0600" + + +def test_today_id_comes_from_the_alarm_not_the_start_instant( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A late start must not shift the id away from the phone's. + + The alarm can begin a full catch-up window late, or hours late on a + re-fire. Both halves derive the id independently with no handshake, so + deriving the PC's from *now* rather than from the resolved alarm time + would silently stop the two log streams joining. + """ + monkeypatch.setattr(_correlation, "get_alarm", lambda: Alarm(6, 45, "remote")) + today = datetime.now().astimezone() + + # Started at 07:35 -- 50 minutes into the catch-up window. + assert _correlation.correlation_id_for_today() == correlation_id_for( + 6, 45, now=today + ) + assert _correlation.correlation_id_for_today().endswith("-0645") diff --git a/wake_alarm/tests/test_inhibit.py b/wake_alarm/tests/test_inhibit.py new file mode 100644 index 0000000..ac0def3 --- /dev/null +++ b/wake_alarm/tests/test_inhibit.py @@ -0,0 +1,224 @@ +"""Tests for the power-key inhibitor held while the alarm rings.""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest + +from wake_alarm import _inhibit +from wake_alarm._constants import INHIBIT_WHO + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def _no_stale_scan(monkeypatch: pytest.MonkeyPatch) -> None: + """Stub the stale scan, so taking an inhibitor never scans the real host. + + Not autouse: TestStaleCleanup exercises the real function, and an autouse + stub would silently make every one of those tests assert nothing. + """ + monkeypatch.setattr(_inhibit, "kill_stale_inhibitors", lambda: 0) + + +def test_resolves_systemd_inhibit_from_path( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The binary is looked up on PATH, never hardcoded.""" + monkeypatch.setattr(_inhibit.shutil, "which", lambda name: f"/usr/bin/{name}") + assert _inhibit._systemd_inhibit_path() == "/usr/bin/systemd-inhibit" + + +@pytest.mark.usefixtures("_no_stale_scan") +class TestTakeInhibitor: + """Taking the inhibitor that blocks the power button.""" + + def test_spawns_a_blocking_inhibitor(self, monkeypatch: pytest.MonkeyPatch) -> None: + """The child blocks handle-power-key and self-terminates.""" + monkeypatch.setattr( + _inhibit, "_systemd_inhibit_path", lambda: "/usr/bin/systemd-inhibit" + ) + captured: list[list[str]] = [] + + def fake_popen(cmd: list[str], **_kwargs: object) -> MagicMock: + captured.append(cmd) + return MagicMock(pid=4242) + + monkeypatch.setattr(_inhibit.subprocess, "Popen", fake_popen) + + assert _inhibit.take_power_key_inhibitor() is not None + + cmd = captured[0] + assert "--mode=block" in cmd + assert "--what=handle-power-key:shutdown:idle" in cmd + # Never `sleep infinity`: an orphaned blocker would hold the power + # button down until the next reboot. + assert cmd[-2] == "sleep" + assert int(cmd[-1]) > 0 + + def test_missing_binary_never_stops_the_alarm( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """No systemd-inhibit means an unprotected alarm, not no alarm.""" + monkeypatch.setattr(_inhibit, "_systemd_inhibit_path", lambda: None) + with caplog.at_level("WARNING"): + assert _inhibit.take_power_key_inhibitor() is None + assert "systemd-inhibit not on PATH" in caplog.text + + def test_spawn_failure_is_survivable( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """A failed spawn is logged and the alarm rings on.""" + monkeypatch.setattr( + _inhibit, "_systemd_inhibit_path", lambda: "/usr/bin/systemd-inhibit" + ) + + def boom(_cmd: list[str], **_kwargs: object) -> MagicMock: + msg = "no fork for you" + raise OSError(msg) + + monkeypatch.setattr(_inhibit.subprocess, "Popen", boom) + with caplog.at_level("WARNING"): + assert _inhibit.take_power_key_inhibitor() is None + assert "rings unprotected" in caplog.text + + def test_clears_a_stale_inhibitor_first( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A blocker from a crashed alarm is cleared before a new one is taken.""" + monkeypatch.setattr( + _inhibit, "_systemd_inhibit_path", lambda: "/usr/bin/systemd-inhibit" + ) + monkeypatch.setattr(_inhibit.subprocess, "Popen", lambda *a, **k: MagicMock()) + cleared: list[str] = [] + + def fake_clear() -> int: + cleared.append("cleared") + return 0 + + monkeypatch.setattr(_inhibit, "kill_stale_inhibitors", fake_clear) + _inhibit.take_power_key_inhibitor() + assert cleared == ["cleared"] + + +class TestReleaseInhibitor: + """Releasing the inhibitor when the alarm ends.""" + + def test_none_is_a_no_op(self) -> None: + """on_close() runs on every path, including ones that took nothing.""" + _inhibit.release_power_key_inhibitor(None) + + def test_already_exited_is_a_no_op(self) -> None: + """A child that self-terminated is not signalled again.""" + process = MagicMock() + process.poll.return_value = 0 + _inhibit.release_power_key_inhibitor(process) + process.terminate.assert_not_called() + + def test_terminates_a_live_inhibitor(self) -> None: + """The ordinary path: SIGTERM, then reap.""" + process = MagicMock() + process.poll.return_value = None + _inhibit.release_power_key_inhibitor(process) + process.terminate.assert_called_once_with() + process.wait.assert_called_once_with(timeout=5) + process.kill.assert_not_called() + + def test_kills_an_inhibitor_that_ignores_sigterm( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """A stuck child must not outlive the alarm holding the power key.""" + process = MagicMock() + process.poll.return_value = None + process.wait.side_effect = subprocess.TimeoutExpired("systemd-inhibit", 5) + with caplog.at_level("WARNING"): + _inhibit.release_power_key_inhibitor(process) + process.kill.assert_called_once_with() + assert "ignored SIGTERM" in caplog.text + + +class TestStaleCleanup: + """Clearing an inhibitor orphaned by a crashed alarm.""" + + def test_identifies_our_own_child( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """A pid is judged by its argv, never by a command-line pattern. + + The earlier ``pkill -f`` version matched the shell that had just typed + the pattern and killed *it* -- substring matching is self-matching. + """ + proc = tmp_path / "77" + proc.mkdir() + (proc / "cmdline").write_bytes( + b"systemd-inhibit\x00--who=" + INHIBIT_WHO.encode() + b"\x00" + ) + monkeypatch.setattr( + _inhibit.pathlib, "Path", lambda *p: tmp_path / p[-2] / p[-1] + ) + assert _inhibit._is_our_inhibitor("77") is True + + def test_ignores_a_pid_that_vanished( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """A process that exited between listing and reading is not stale.""" + monkeypatch.setattr( + _inhibit.pathlib, "Path", lambda *p: tmp_path / "definitely-absent" + ) + assert _inhibit._is_our_inhibitor("99") is False + + def test_kills_only_matching_pids(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Someone else's systemd-inhibit is left strictly alone.""" + monkeypatch.setattr(_inhibit.shutil, "which", lambda _name: "/usr/bin/pgrep") + monkeypatch.setattr( + _inhibit.subprocess, + "run", + lambda *a, **k: MagicMock(stdout=b"11\n22\n"), + ) + monkeypatch.setattr(_inhibit, "_is_our_inhibitor", lambda pid: pid == "22") + signalled: list[int] = [] + monkeypatch.setattr( + _inhibit.os, "kill", lambda pid, _sig: signalled.append(pid) + ) + assert _inhibit.kill_stale_inhibitors() == 1 + assert signalled == [22] + + def test_survives_a_pid_that_exits_mid_kill( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A race between listing and killing is not an error.""" + monkeypatch.setattr(_inhibit.shutil, "which", lambda _name: "/usr/bin/pgrep") + monkeypatch.setattr( + _inhibit.subprocess, "run", lambda *a, **k: MagicMock(stdout=b"11\n") + ) + monkeypatch.setattr(_inhibit, "_is_our_inhibitor", lambda _pid: True) + + def gone(_pid: int, _sig: int) -> None: + raise ProcessLookupError + + monkeypatch.setattr(_inhibit.os, "kill", gone) + assert _inhibit.kill_stale_inhibitors() == 0 + + def test_no_pgrep_means_no_scan(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Without pgrep the cleanup is skipped rather than guessed at.""" + monkeypatch.setattr(_inhibit.shutil, "which", lambda _name: None) + assert _inhibit.kill_stale_inhibitors() == 0 + + def test_pgrep_failure_is_survivable( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """A failed scan must not stop the alarm from starting.""" + monkeypatch.setattr(_inhibit.shutil, "which", lambda _name: "/usr/bin/pgrep") + + def boom(*_a: object, **_k: object) -> None: + msg = "nope" + raise OSError(msg) + + monkeypatch.setattr(_inhibit.subprocess, "run", boom) + with caplog.at_level("WARNING"): + assert _inhibit.kill_stale_inhibitors() == 0 + assert "Could not check for a stale" in caplog.text diff --git a/wake_alarm/tests/test_trigger.py b/wake_alarm/tests/test_trigger.py index 313bd88..f51632e 100644 --- a/wake_alarm/tests/test_trigger.py +++ b/wake_alarm/tests/test_trigger.py @@ -3,19 +3,21 @@ from __future__ import annotations from datetime import datetime, timezone -from unittest.mock import MagicMock, patch +from typing import TYPE_CHECKING +from unittest.mock import patch import pytest +if TYPE_CHECKING: + from pathlib import Path + +from wake_alarm import _alarm_session from wake_alarm._sync import Alarm from wake_alarm._trigger import ( _decide, _fire_window, _now, - _parse_args, _skip_reason, - _start_target_unit, - main, ) @@ -31,10 +33,16 @@ def _at(hour: int, minute: int) -> datetime: @pytest.fixture(autouse=True) def _ungated() -> None: - """Open both "already handled today" gates unless a test closes them.""" + """Open every gate unless a test closes one. + + ``should_refire`` is stubbed here too: unstubbed it reads the *real* + session record on the host, so a test run started while an alarm was + actually ringing would silently invert half these assertions. + """ with ( patch("wake_alarm._trigger.was_alarm_dismissed_today", return_value=False), patch("wake_alarm._trigger.was_workout_logged_today", return_value=False), + patch("wake_alarm._trigger.should_refire", return_value=False), ): yield @@ -155,98 +163,69 @@ class TestDecide: assert fire -class TestStartTargetUnit: - """Tests for the systemctl invocation.""" +class TestRefireOnAnUnfinishedSession: + """A power-off mid-alarm must not be a way out.""" - def test_starts_the_orchestrator_not_the_alarm(self) -> None: - """The trigger must start morning-routine, never wake-alarm.""" - completed = MagicMock(returncode=0) - with patch("subprocess.run", return_value=completed) as run: - assert _start_target_unit() - assert run.call_args[0][0] == [ - "systemctl", - "--user", - "start", - "morning-routine.service", - ] + def test_refires_long_after_the_window_closed(self) -> None: + """The whole point: the machine may have been off for the window. - def test_nonzero_exit_is_reported(self) -> None: - """A systemctl failure returns False rather than raising.""" - completed = MagicMock(returncode=1, stderr=b"no such unit") - with patch("subprocess.run", return_value=completed): - assert not _start_target_unit() + Without this, holding the power button ends the alarm for good -- + the catch-up window closes while the PC is down and nothing on disk + ever said an alarm was ringing. + """ + with patch("wake_alarm._trigger.should_refire", return_value=True): + fire, reason = _decide(_at(20, 0), Alarm(6, 45, "remote")) + assert fire + assert "cut short" in reason - def test_missing_systemctl_is_swallowed(self) -> None: - """An OSError (no systemctl) returns False rather than raising.""" - with patch("subprocess.run", side_effect=OSError("not found")): - assert not _start_target_unit() - - def test_does_not_raise_on_failure(self) -> None: - """check=False: a failing start never raises CalledProcessError.""" - completed = MagicMock(returncode=1, stderr=b"") - with patch("subprocess.run", return_value=completed) as run: - _start_target_unit() - assert run.call_args.kwargs["check"] is False - - -class TestMain: - """Tests for the tick entry point.""" - - def test_parse_args_default(self) -> None: - """--dry-run defaults to off.""" - assert not _parse_args([]).dry_run - - def _run_main(self, argv: list[str], now: datetime, alarm: Alarm) -> MagicMock: + def test_dismissed_gate_still_beats_a_refire(self) -> None: + """The ordinary gates are checked first and still win.""" with ( - patch("sys.argv", ["_trigger", *argv]), - patch("wake_alarm._trigger._now", return_value=now), - patch("wake_alarm._trigger.get_alarm", return_value=alarm), - patch("wake_alarm._trigger._start_target_unit") as start, + patch("wake_alarm._trigger.should_refire", return_value=True), + patch( + "wake_alarm._trigger.was_alarm_dismissed_today", + return_value=True, + ), ): - main() - return start + fire, reason = _decide(_at(20, 0), Alarm(6, 45, "remote")) + assert not fire + assert reason == "alarm already dismissed today" - def test_fires_at_the_alarm_minute(self) -> None: - """On the alarm minute the target unit is started.""" - start = self._run_main([], _at(6, 45), Alarm(6, 45, "remote")) - start.assert_called_once() - - def test_does_not_fire_early(self) -> None: - """Before the alarm nothing is started.""" - start = self._run_main([], _at(5, 0), Alarm(6, 45, "remote")) - start.assert_not_called() - - def test_dry_run_never_starts_anything( - self, capsys: pytest.CaptureFixture[str] + def test_end_to_end_through_a_real_session_record( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - """--dry-run reports WOULD FIRE without touching systemd.""" - start = self._run_main(["--dry-run"], _at(6, 45), Alarm(6, 45, "remote")) - start.assert_not_called() - assert "WOULD FIRE" in capsys.readouterr().out + """_decide against a real record, with should_refire NOT patched. - def test_dry_run_reports_not_firing( - self, capsys: pytest.CaptureFixture[str] - ) -> None: - """--dry-run explains why it is not firing.""" - self._run_main(["--dry-run"], _at(5, 0), Alarm(6, 45, "remote")) - out = capsys.readouterr().out - assert "not firing" in out - assert "still ahead" in out + Every other test here stubs ``should_refire``, so the wiring between + the record on disk and the decision was untested -- and a manual run + found it broken while the whole suite stayed green. This is the + deterministic version of that manual check. + """ + monkeypatch.setattr( + _alarm_session, "ALARM_SESSION_FILE", tmp_path / "session.json" + ) + now = _at(20, 0) + # The record is stamped by the alarm process, which ran at fire time -- + # not at this tick. Pin its clock to the same fictional morning as the + # tick, or the two dates are years apart and the age check is + # meaningless. + monkeypatch.setattr(_alarm_session, "_now", lambda: _at(19, 30)) + alarm = Alarm(6, 45, "remote") - @pytest.mark.parametrize("source", ["fallback", "stale-cache"]) - def test_warns_when_firing_on_a_degraded_source( - self, caplog: pytest.LogCaptureFixture, source: str - ) -> None: - """Firing on a non-phone alarm time is logged loudly.""" - with caplog.at_level("WARNING"): - self._run_main([], _at(6, 45), Alarm(6, 45, source)) - assert "degraded alarm source" in caplog.text + with patch("wake_alarm._trigger.should_refire", _alarm_session.should_refire): + assert _decide(now, alarm)[0] is False # window closed, no record - @pytest.mark.parametrize("source", ["remote", "cache"]) - def test_quiet_when_firing_on_the_phone_alarm( - self, caplog: pytest.LogCaptureFixture, source: str - ) -> None: - """A normal fire does not emit the degraded-source warning.""" - with caplog.at_level("WARNING"): - self._run_main([], _at(6, 45), Alarm(6, 45, source)) - assert "degraded alarm source" not in caplog.text + _alarm_session.start_session("wake-20260719-0645") + fire, reason = _decide(now, alarm) + assert fire, "a cut-short alarm must ring again" + assert "cut short" in reason + + _alarm_session.finish_session() + assert _decide(now, alarm)[0] is False # dismissed stays dismissed + + def test_refire_is_asked_about_the_tick_instant(self) -> None: + """The session's age is measured against the tick, not wall-clock now.""" + with patch("wake_alarm._trigger.should_refire") as refire: + refire.return_value = False + _decide(_at(20, 0), Alarm(6, 45, "remote")) + refire.assert_called_once_with(now=_at(20, 0)) diff --git a/wake_alarm/tests/test_trigger_main.py b/wake_alarm/tests/test_trigger_main.py new file mode 100644 index 0000000..e575473 --- /dev/null +++ b/wake_alarm/tests/test_trigger_main.py @@ -0,0 +1,143 @@ +"""Tests for _trigger.py — the systemd invocation and the CLI entry point. + +Split from ``test_trigger.py`` for the 250-line cap. The header is copied +wholesale rather than trimmed: without the autouse ``_ungated`` fixture these +tests read the *real* gates on the host and would pass or fail by accident. +""" + +from __future__ import annotations + +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest + +from wake_alarm._sync import Alarm +from wake_alarm._trigger import ( + _parse_args, + _start_target_unit, + main, +) + + +def _at(hour: int, minute: int) -> datetime: + """Return a Sunday at the given local wall-clock time, timezone-aware. + + Built the same way production does — naive wall clock, then astimezone() + to attach the system offset — so the tests exercise the real comparison + semantics rather than a hand-picked fixed offset. + """ + return datetime(2026, 7, 19, hour, minute, tzinfo=None).astimezone() + + +@pytest.fixture(autouse=True) +def _ungated() -> None: + """Open every gate unless a test closes one. + + ``should_refire`` is stubbed here too: unstubbed it reads the *real* + session record on the host, so a test run started while an alarm was + actually ringing would silently invert half these assertions. + """ + with ( + patch("wake_alarm._trigger.was_alarm_dismissed_today", return_value=False), + patch("wake_alarm._trigger.was_workout_logged_today", return_value=False), + patch("wake_alarm._trigger.should_refire", return_value=False), + ): + yield + + +class TestStartTargetUnit: + """Tests for the systemctl invocation.""" + + def test_starts_the_orchestrator_not_the_alarm(self) -> None: + """The trigger must start morning-routine, never wake-alarm.""" + completed = MagicMock(returncode=0) + with patch("subprocess.run", return_value=completed) as run: + assert _start_target_unit() + assert run.call_args[0][0] == [ + "systemctl", + "--user", + "start", + "morning-routine.service", + ] + + def test_nonzero_exit_is_reported(self) -> None: + """A systemctl failure returns False rather than raising.""" + completed = MagicMock(returncode=1, stderr=b"no such unit") + with patch("subprocess.run", return_value=completed): + assert not _start_target_unit() + + def test_missing_systemctl_is_swallowed(self) -> None: + """An OSError (no systemctl) returns False rather than raising.""" + with patch("subprocess.run", side_effect=OSError("not found")): + assert not _start_target_unit() + + def test_does_not_raise_on_failure(self) -> None: + """check=False: a failing start never raises CalledProcessError.""" + completed = MagicMock(returncode=1, stderr=b"") + with patch("subprocess.run", return_value=completed) as run: + _start_target_unit() + assert run.call_args.kwargs["check"] is False + + +class TestMain: + """Tests for the tick entry point.""" + + def test_parse_args_default(self) -> None: + """--dry-run defaults to off.""" + assert not _parse_args([]).dry_run + + def _run_main(self, argv: list[str], now: datetime, alarm: Alarm) -> MagicMock: + with ( + patch("sys.argv", ["_trigger", *argv]), + patch("wake_alarm._trigger._now", return_value=now), + patch("wake_alarm._trigger.get_alarm", return_value=alarm), + patch("wake_alarm._trigger._start_target_unit") as start, + ): + main() + return start + + def test_fires_at_the_alarm_minute(self) -> None: + """On the alarm minute the target unit is started.""" + start = self._run_main([], _at(6, 45), Alarm(6, 45, "remote")) + start.assert_called_once() + + def test_does_not_fire_early(self) -> None: + """Before the alarm nothing is started.""" + start = self._run_main([], _at(5, 0), Alarm(6, 45, "remote")) + start.assert_not_called() + + def test_dry_run_never_starts_anything( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """--dry-run reports WOULD FIRE without touching systemd.""" + start = self._run_main(["--dry-run"], _at(6, 45), Alarm(6, 45, "remote")) + start.assert_not_called() + assert "WOULD FIRE" in capsys.readouterr().out + + def test_dry_run_reports_not_firing( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """--dry-run explains why it is not firing.""" + self._run_main(["--dry-run"], _at(5, 0), Alarm(6, 45, "remote")) + out = capsys.readouterr().out + assert "not firing" in out + assert "still ahead" in out + + @pytest.mark.parametrize("source", ["fallback", "stale-cache"]) + def test_warns_when_firing_on_a_degraded_source( + self, caplog: pytest.LogCaptureFixture, source: str + ) -> None: + """Firing on a non-phone alarm time is logged loudly.""" + with caplog.at_level("WARNING"): + self._run_main([], _at(6, 45), Alarm(6, 45, source)) + assert "degraded alarm source" in caplog.text + + @pytest.mark.parametrize("source", ["remote", "cache"]) + def test_quiet_when_firing_on_the_phone_alarm( + self, caplog: pytest.LogCaptureFixture, source: str + ) -> None: + """A normal fire does not emit the degraded-source warning.""" + with caplog.at_level("WARNING"): + self._run_main([], _at(6, 45), Alarm(6, 45, source)) + assert "degraded alarm source" not in caplog.text