kuhyx/wake-alarm
1
0
Fork 0
mirror of https://github.com/kuhyx/wake-alarm.git synced 2026-11-11 12:47:31 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Phase 5: the power button no longer ends the alarm (item 2)

Browse Source 
Two defences, because one press and one held press are different problems.

**Prevention — _inhibit.py.** A logind block-mode inhibitor on
handle-power-key:shutdown:idle, taken while the alarm is audible and released
in on_close(). Confirmed on the host before writing it that
PowerKeyIgnoreInhibited=no is in force, so no logind.conf drop-in is needed --
and none is added: a drop-in changes the machine permanently, an inhibitor is
scoped to the alarm.

Verified live: the inhibitor appears in `systemd-inhibit --list` as
`wake-alarm ... shutdown:idle:handle-power-key ... block` and disappears on
release.

The child runs `sleep 7200`, never `sleep infinity`. An orphaned block-mode
inhibitor would hold the power button down until reboot -- worse than the bug
being fixed -- so it self-terminates, and a stale one is cleared at startup.

That cleanup was wrong on the first attempt and the host run caught it:
`pkill -f "--who=wake-alarm"` matched the *shell that had just typed the
pattern* and killed it, leaving the real inhibitor alive (`killed 0`, one
inhibitor still listed). Substring matching is inherently self-matching. Now
it lists `pgrep -x systemd-inhibit` and confirms each pid via
/proc/<pid>/cmdline. Re-verified: orphan reaped, exactly one live inhibitor,
zero remaining after release.

**Consequence — _alarm_session.py.** The real gap was that nothing on disk
ever said "ringing, not yet dismissed": wake_state.json is written only on
dismissal, so a hard power-off left no trace and the catch-up window closed
while the machine was down. A signed record is now written at fire time, and
_trigger re-fires on one still marked ringing, ignoring the window.

Deliberately no date field. _state.py keys by UTC and
screen_locker/_wake_state.py mirrors that helper to read the very same file
from another repo, so the UTC basis there is load-bearing. The plan's
_today_str() flip is therefore NOT done -- it would have made any dismissal
between 00:00-02:00 local read as yesterday's in screen-locker and silently
lose an earned workout skip. This record carries only an aware fired_at and
compares elapsed time, so the midnight question never arises.

Age-bounded at 5400 s: an alarm that dies before dismissal leaves a record
nothing ever clears, and unbounded this would restart the alarm every minute
forever.

Two bugs the unit tests could not see, both found by running it:

- A future-stamped record returned None, i.e. "not ringing". But a backward
  clock jump is routine on a machine that hibernates nightly (NTP correcting
  after resume), so that failed *open* on exactly the escape it closes: ring
  on a fast clock, clock corrects, power-off works again. Now clamped to zero
  within a 900 s tolerance and logged.
- _alarm.py derived the correlation id from the alarm's *start* instant. On a
  catch-up or re-fire the PC would write wake-...-0735 while the phone wrote
  wake-...-0700 and the two log streams stop joining. Now
  correlation_id_for_today() derives it from the resolved alarm time, which is
  what _correlation.py always specified.

Every trigger test patched should_refire, so _decide against a real record was
untested -- which is why only the manual run found the first one. There is now
an end-to-end test that writes a real session and asserts the decision.

Splits carried: test_trigger.py 294 -> 197 + test_trigger_main.py,
test_alarm_part2.py 266 -> 170 + test_alarm_timers_flash.py. The inhibitor
handle moved into _AlarmHardware, which also settles pylint's attribute count
and puts it beside the other thing on_close() gives back.

No new suppressions: the one inline pylint disable was redundant, since
consider-using-with is already disabled repo-wide in pyproject.toml.

452 tests pass, 100% branch coverage.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EckWmyfetYq3Mi8MAnUudh
This commit is contained in:
Krzysztof Rudnicki 2026-08-27 20:26:37 +02:00
parent 931a4187cb
commit e8f50828e6
16 changed files with 1465 additions and 195 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
wake_alarm/_alarm.py
View File

@ -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()

227
wake_alarm/_alarm_session.py Normal file
View File

@ -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

10
wake_alarm/_alarm_state.py
View File

@ -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

37
wake_alarm/_constants.py
View File

@ -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

22
wake_alarm/_correlation.py
View File

@ -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())

191
wake_alarm/_inhibit.py Normal file
View File

@ -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"})

10
wake_alarm/_trigger.py
View File

@ -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"

14
wake_alarm/tests/conftest.py
View File

@ -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),

122
wake_alarm/tests/test_alarm_part2.py
View File

@ -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()

23
wake_alarm/tests/test_alarm_part4.py
View File

@ -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."""

230
wake_alarm/tests/test_alarm_session.py Normal file
View File

@ -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

192
wake_alarm/tests/test_alarm_timers_flash.py Normal file
View File

@ -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()

26
wake_alarm/tests/test_correlation.py
View File

@ -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")

224
wake_alarm/tests/test_inhibit.py Normal file
View File

@ -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

159
wake_alarm/tests/test_trigger.py
View File

@ -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))

143
wake_alarm/tests/test_trigger_main.py Normal file
View File

@ -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