From 9e0cd123ff29872698995251f389839736fec3ce Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Wed, 26 Aug 2026 23:46:10 +0200 Subject: [PATCH] Phase 3: remove the max-fan ramp, split _alarm.py, wire the new modules Item 14: the fan ramp is gone entirely. It caused anxiety without helping anyone wake up. Deletes _max_fans/_restore_fans/_find_fan_hwmon, the _AlarmHardware.fan_state field, both call sites, and their tests. wake-alarm-fans.sh and its NOPASSWD sudoers entry come out in the install.sh pass -- dropping a passwordless root helper is a bonus. _alarm.py was 478 lines and the 250-line gate fires on any commit that stages it, so this carries the split it has owed for a while. New modules, each a real seam rather than an arbitrary cut: _gating should-it-run-today + CLI (no time gate; that is _trigger's) _beep_loop the sounding thread, now driving _escalation's planner _alarm_timers self-rescheduling challenge refresh / skip window (mixin) _alarm_state the progress + hardware dataclasses _rounds challenge presentation, re-exporting the view helpers _countdown skip-window arithmetic and demo/production intervals _dismiss_view the post-dismissal banner _screen_flash the ~1.3 Hz attention flash (kept under the 3 Hz threshold) _lock_setup gatelock config + arbiter claim _startup wake display, brightness, plug _tone tone synthesis and cache (out of _audio) _brightness xrandr brightness (was in _audio, never belonged there) main() deliberately stays in _alarm.py: morning_routine/_orchestrator.py:39 hardcodes "wake_alarm._alarm", and moving it is the exact mistake that caused a month-long silent production failure once already. Verified after the split that the module still resolves and accepts --production. Also fixes the plug pre-roll found while researching speakers: main() fired turn_on_plug() and the first beep in the same instant, so an amplifier's soft-start relay swallowed the opening of the alarm. PLUG_PREROLL_SECONDS=4 now sits between them. Two bugs found by running the code rather than by lint: - ruff stripped `struct` from the extracted _tone.py because the import landed before its first usage; only executing it caught the NameError. - test fixtures patched _play_on_all_sinks/_set_max_brightness at wake_alarm._alarm, which silently stopped intercepting once those moved. 404 tests, 100% branch coverage, no new noqa/type-ignore suppressions. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01EckWmyfetYq3Mi8MAnUudh --- wake_alarm/_alarm.py | 328 +++--------------- wake_alarm/_alarm_state.py | 34 ++ wake_alarm/_alarm_timers.py | 109 ++++++ wake_alarm/_audio.py | 159 +-------- wake_alarm/_beep_loop.py | 99 ++++++ wake_alarm/_brightness.py | 48 +++ wake_alarm/_constants.py | 6 + wake_alarm/_countdown.py | 92 +++++ wake_alarm/_dismiss_view.py | 58 ++++ wake_alarm/_gating.py | 81 +++++ wake_alarm/_lock_setup.py | 55 +++ wake_alarm/_rounds.py | 120 +++++++ wake_alarm/_screen_flash.py | 42 +++ wake_alarm/_startup.py | 47 +++ wake_alarm/_tone.py | 48 +++ wake_alarm/tests/test_alarm.py | 215 +----------- wake_alarm/tests/test_alarm_audio.py | 186 ---------- wake_alarm/tests/test_alarm_beep_phases.py | 161 --------- wake_alarm/tests/test_alarm_dismiss.py | 232 +++++++++++++ wake_alarm/tests/test_alarm_fanout.py | 139 ++++++++ .../tests/test_alarm_flash_challenge.py | 20 +- wake_alarm/tests/test_alarm_main.py | 149 ++++++++ wake_alarm/tests/test_alarm_part2.py | 231 +----------- wake_alarm/tests/test_alarm_part3.py | 14 +- wake_alarm/tests/test_alarm_part4.py | 77 +--- wake_alarm/tests/test_alarm_sinks.py | 216 +----------- wake_alarm/tests/test_beep_loop.py | 151 ++++++++ wake_alarm/tests/test_brightness.py | 118 +++++++ wake_alarm/tests/test_escape_flow.py | 149 +------- wake_alarm/tests/test_escape_integration.py | 229 ++++++++++++ wake_alarm/tests/test_gating.py | 75 ++++ wake_alarm/tests/test_pcspkr.py | 40 +++ wake_alarm/tests/test_sinks_activate.py | 158 +++++++++ wake_alarm/tests/test_sinks_restore.py | 109 ++++++ wake_alarm/tests/test_sinks_warn.py | 83 +++++ wake_alarm/tests/test_tone.py | 53 +++ 36 files changed, 2470 insertions(+), 1661 deletions(-) create mode 100644 wake_alarm/_alarm_state.py create mode 100644 wake_alarm/_alarm_timers.py create mode 100644 wake_alarm/_beep_loop.py create mode 100644 wake_alarm/_brightness.py create mode 100644 wake_alarm/_countdown.py create mode 100644 wake_alarm/_dismiss_view.py create mode 100644 wake_alarm/_gating.py create mode 100644 wake_alarm/_lock_setup.py create mode 100644 wake_alarm/_rounds.py create mode 100644 wake_alarm/_screen_flash.py create mode 100644 wake_alarm/_startup.py create mode 100644 wake_alarm/_tone.py delete mode 100644 wake_alarm/tests/test_alarm_beep_phases.py create mode 100644 wake_alarm/tests/test_alarm_dismiss.py create mode 100644 wake_alarm/tests/test_alarm_fanout.py create mode 100644 wake_alarm/tests/test_alarm_main.py create mode 100644 wake_alarm/tests/test_beep_loop.py create mode 100644 wake_alarm/tests/test_brightness.py create mode 100644 wake_alarm/tests/test_escape_integration.py create mode 100644 wake_alarm/tests/test_gating.py create mode 100644 wake_alarm/tests/test_pcspkr.py create mode 100644 wake_alarm/tests/test_sinks_activate.py create mode 100644 wake_alarm/tests/test_sinks_restore.py create mode 100644 wake_alarm/tests/test_sinks_warn.py create mode 100644 wake_alarm/tests/test_tone.py diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index eac53f4..4f1eb34 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -1,103 +1,66 @@ -"""Weekend wake alarm daemon with escalating beep and dismiss challenge. +"""Wake alarm: escalating beeps behind a fullscreen dismiss challenge. -Run as a systemd service on boot. Checks if today is an alarm day, -plays escalating system beeps, and presents a fullscreen dismiss -challenge (random code typing). Dismissing within the window grants a -workout-free day via HMAC-signed wake state. +Started by ``morning-routine.service``, never directly (AGENTS.md). Gate: +``_gating``. Sound: ``_beep_loop``. Timers: ``_alarm_timers``. """ from __future__ import annotations -import argparse import contextlib -from dataclasses import dataclass, field from datetime import datetime, timezone import logging import sys import threading -import time import tkinter as tk from gatelock import ( - RANK_WAKE_ALARM, - Arbiter, GateRoot, - LockConfig, LockWindow, SurfaceInfo, ) -from wake_alarm._alarm_display import _restore_display, _wake_display +from wake_alarm._alarm_display import _restore_display +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 -from wake_alarm._audio import ( - _beep_loud, - _beep_medium, - _beep_soft, - _max_fans, - _play_on_all_sinks, - _restore_fans, - _set_max_brightness, -) +from wake_alarm._beep_loop import start_beep_thread from wake_alarm._challenges import ( - _Challenge, _make_challenge, ) from wake_alarm._constants import ( - ALARM_DAYS, - DISMISS_CODE_REFRESH_SECONDS, DISMISS_FLASH_SECONDS, DISMISS_ROUNDS_REQUIRED, - DISMISS_WINDOW_MINUTES, - LOUD_TOGGLE_INTERVAL, - MEDIUM_BEEP_INTERVAL, - PHASE_MEDIUM_END, - PHASE_SOFT_END, - SOFT_BEEP_INTERVAL, ) from wake_alarm._escape_flow import EscapeHatch, build_tracker +from wake_alarm._gating import parse_args, should_run_alarm +from wake_alarm._lock_setup import build_config, claim_arbiter from wake_alarm._logging_setup import configure_logging +from wake_alarm._rounds import ( + BANNER_LINGER_MS, + FLASH_PERIOD_MS, + apply_flash_colour, + hide_flash_code, + normalise, + show_challenge, + show_dismissal, + show_flash_countdown, + show_round_progress, + show_wrong_answer, +) from wake_alarm._sinks import ( _activate_all_outputs, - _AudioState, _restore_audio_state, - _warn_if_no_real_sink, ) -from wake_alarm._smart_plug import turn_off_plug, turn_on_plug +from wake_alarm._smart_plug import turn_off_plug +from wake_alarm._startup import prepare_hardware from wake_alarm._state import ( save_wake_state, - was_alarm_dismissed_today, - was_workout_logged_today, ) _logger = logging.getLogger(__name__) -def _is_alarm_day() -> bool: - """Check if today is an alarm day.""" - return datetime.now(tz=timezone.utc).weekday() in ALARM_DAYS - - -@dataclass -class _AlarmProgress: - """Mutable dismiss-challenge progress state.""" - - current_challenge: _Challenge - started_at: float = field(default_factory=time.monotonic) - skip_earnable: bool = True - rounds_completed: int = 0 - flash_remaining: int = 0 - flash_on: bool = False - - -@dataclass -class _AlarmHardware: - """Hardware state captured at alarm start, restored when it closes.""" - - fan_state: bool - audio_restore: _AudioState | None - - -class WakeAlarm: +class WakeAlarm(AlarmTimersMixin): """Fullscreen wake alarm with escalating beep and dismiss challenge.""" def __init__(self, *, demo_mode: bool = False) -> None: @@ -114,28 +77,10 @@ class WakeAlarm: self.root = GateRoot() self.root.on_callback_error = self.on_callback_error self.root.title("Wake Alarm" + (" [DEMO]" if demo_mode else "")) - # Production hard-locks as of gatelock v0.2.0. The old comment here - # deferred it for want of "a watchdog for a clean, silent lock - # failure" -- the recovery loop is that watchdog. Demo stays grab-free - # so a manual test cannot trap the developer, but takes - # overrideredirect anyway: per-output placement is impossible without - # it, since a window manager rewrites a managed window's geometry. - config = LockConfig( - mode="soft" if demo_mode else "hard", - overrideredirect=True, - app_name="wake_alarm", - rank=RANK_WAKE_ALARM, - ) + config = build_config(demo_mode=demo_mode) # A local, not an attribute: the arbiter is published here and then # owned by LockWindow, which is what releases it on close. - arbiter = Arbiter( - "wake_alarm", - RANK_WAKE_ALARM, - grab=config.resolved_grab(), - disable_vt=config.resolved_disable_vt(), - ) - arbiter.publish() - arbiter.acquire_holder() + arbiter = claim_arbiter(config) # Built before setup(): setup() calls build_surface for every live # output, and those need the challenge and the shared entry variable. @@ -150,11 +95,9 @@ class WakeAlarm: self._start_flash_countdown() self._schedule_code_refresh() self._schedule_skip_window_close() - self._start_beep_thread() - self._hardware = _AlarmHardware( - fan_state=_max_fans(), audio_restore=_activate_all_outputs() - ) - self._start_screen_flash() + start_beep_thread(self._stop_beep, self._progress.started_at) + self._hardware = _AlarmHardware(audio_restore=_activate_all_outputs()) + self._flash_step() self._lock.grab_input() def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None: @@ -199,9 +142,8 @@ class WakeAlarm: self._view.entry.focus_force() def on_close(self) -> None: - """Restore fans/audio/display/plug; runs on every exit path, even SIGTERM.""" + """Restore audio/display/plug; runs on every exit path, even SIGTERM.""" self._stop_beep.set() - _restore_fans(active=self._hardware.fan_state) _restore_audio_state(self._hardware.audio_restore) _restore_display() turn_off_plug() @@ -214,15 +156,10 @@ class WakeAlarm: correct round generates a new random challenge so the user must stay awake and re-engage each time. """ - entered = self._view.entry.get().strip().upper() + entered = normalise(self._view.entry.get()) if entered != self._progress.current_challenge.answer: - self._view.status_label.configure(text="Wrong! Try again.") - self._view.entry.delete(0, tk.END) + show_wrong_answer(self._view, self._progress.current_challenge) if self._progress.current_challenge.kind == "flash": - self._view.code_label.configure( - text=self._progress.current_challenge.display, - fg=self._view.config.success, - ) self._start_flash_countdown() return self._progress.rounds_completed += 1 @@ -230,18 +167,11 @@ class WakeAlarm: self._dismiss_alarm(earned_skip=self._progress.skip_earnable) return self._progress.current_challenge = _make_challenge() - self._view.code_label.configure( - text=self._progress.current_challenge.display, - fg=self._view.config.success, - ) - self._view.info_label.configure(text=self._progress.current_challenge.hint) - self._view.entry.delete(0, tk.END) - next_round = self._progress.rounds_completed + 1 - self._view.round_label.configure( - text=f"Round {next_round} / {DISMISS_ROUNDS_REQUIRED}", - ) - self._view.status_label.configure( - text=f"Round {self._progress.rounds_completed} done — keep going!", + show_challenge(self._view, self._progress.current_challenge) + show_round_progress( + self._view, + completed=self._progress.rounds_completed, + required=DISMISS_ROUNDS_REQUIRED, ) if self._progress.current_challenge.kind == "flash": self._start_flash_countdown() @@ -256,15 +186,11 @@ class WakeAlarm: if not self._active: return if self._progress.flash_remaining > 0: - self._view.status_label.configure( - text=f"Memorise! Hiding in {self._progress.flash_remaining}s…", - ) + show_flash_countdown(self._view, self._progress.flash_remaining) self._progress.flash_remaining -= 1 self.root.after(1000, self._flash_tick) else: - hidden = "?" * len(self._progress.current_challenge.display) - self._view.code_label.configure(text=hidden, fg=self._view.config.muted) - self._view.status_label.configure(text="Now type the code from memory!") + hide_flash_code(self._view, self._progress.current_challenge) def _dismiss_alarm(self, *, earned_skip: bool) -> None: """Dismiss the alarm and save state.""" @@ -273,187 +199,40 @@ class WakeAlarm: self._stop_beep.set() now_iso = datetime.now(tz=timezone.utc).isoformat() save_wake_state(dismissed_at=now_iso, skip_workout=earned_skip) - - for container in self._view.containers: - for widget in container.winfo_children(): - widget.destroy() - - msg = ( - "Workout skip earned! Enjoy your morning." - if earned_skip - else "Alarm dismissed. No workout skip." + show_dismissal( + self._view.containers, + self._view.config, + earned_skip=earned_skip, ) - cfg = self._view.config - color = cfg.success if earned_skip else cfg.warning - - for container in self._view.containers: - tk.Label( - container, - text=msg, - font=cfg.font("display", bold=True), - fg=color, - bg=cfg.bg, - ).pack(pady=cfg.space("lg")) - - self.root.after(3000, self._lock.close) - - def _schedule_code_refresh(self) -> None: - """Replace the current challenge periodically. - - Ensures the user can't simply wait out a hard challenge type — a new - random challenge is generated every DISMISS_CODE_REFRESH_SECONDS. - """ - if not self._active: - return - self._progress.current_challenge = _make_challenge() - self._view.code_label.configure( - text=self._progress.current_challenge.display, - fg=self._view.config.success, - ) - self._view.info_label.configure(text=self._progress.current_challenge.hint) - self._view.entry.delete(0, tk.END) - if self._progress.current_challenge.kind == "flash": - self._start_flash_countdown() - ms = DISMISS_CODE_REFRESH_SECONDS * 1000 if not self.demo_mode else 10_000 - self.root.after(ms, self._schedule_code_refresh) - - def _schedule_skip_window_close(self) -> None: - """Mark the workout-skip reward as expired after the allowed time.""" - ms = DISMISS_WINDOW_MINUTES * 60 * 1000 if not self.demo_mode else 30_000 - self.root.after(ms, self._on_skip_window_expired) - - def _on_skip_window_expired(self) -> None: - """Skip window closed: keep the alarm running, deny the workout skip. - - The alarm intentionally does NOT stop here - it keeps beeping and - flashing until the user actually types the code. Only the workout-skip - reward expires; dismissing now silences the alarm without earning a skip. - """ - if not self._active: - return - self._progress.skip_earnable = False - self._view.info_label.configure( - text="Skip window closed - type the code to stop the alarm", - ) - self._view.status_label.configure(text="No workout skip today.") - _logger.info("Skip window expired - alarm continues until dismissed.") - - def _update_timer(self) -> None: - """Show the skip-window countdown, then a keep-going silence prompt.""" - if not self._active: - return - elapsed = time.monotonic() - self._progress.started_at - window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30 - remaining = max(0, window - elapsed) - if self._progress.skip_earnable and remaining > 0: - minutes = int(remaining) // 60 - seconds = int(remaining) % 60 - self._view.timer_label.configure( - text=f"Skip window: {minutes:02d}:{seconds:02d}", - ) - else: - self._view.timer_label.configure( - text="No skip available - type the code to stop the alarm", - ) - self._escape.maybe_offer(elapsed) - self.root.after(1000, self._update_timer) + self.root.after(BANNER_LINGER_MS, self._lock.close) def _on_escape_granted(self) -> None: """The hatch was earned: silence the alarm, grant no workout skip.""" _logger.warning("alarm dismissed via the escape hatch, no skip earned") self._dismiss_alarm(earned_skip=False) - def _start_beep_thread(self) -> None: - """Start the background beep escalation thread.""" - thread = threading.Thread(target=self._beep_loop, daemon=True) - thread.start() - - def _start_screen_flash(self) -> None: - """Start flashing the screen background to attract attention.""" - self._flash_step() - def _flash_step(self) -> None: - """Alternate background colour every 750 ms (below seizure-risk 3 Hz).""" + """Alternate the background colour to keep the screen attention-grabbing.""" if not self._active: return cfg = self._view.config colour = cfg.danger if self._progress.flash_on else cfg.bg - self.root.configure(bg=colour) - # The per-output surfaces sit on top of the backdrop, so flashing - # only the root would be invisible on every monitor. - for container in self._view.containers: - with contextlib.suppress(tk.TclError): - container.configure(bg=colour) - container.winfo_toplevel().configure(bg=colour) + apply_flash_colour(self.root, self._view.containers, colour) self._progress.flash_on = not self._progress.flash_on - self.root.after(750, self._flash_step) - - def _beep_loop(self) -> None: - """Escalating beep loop running in background thread.""" - while not self._stop_beep.is_set(): - elapsed_minutes = (time.monotonic() - self._progress.started_at) / 60.0 - - if elapsed_minutes < PHASE_SOFT_END: - _play_on_all_sinks(440) - _beep_soft() - self._stop_beep.wait(SOFT_BEEP_INTERVAL) - elif elapsed_minutes < PHASE_MEDIUM_END: - _play_on_all_sinks(1000) - _beep_medium() - self._stop_beep.wait(MEDIUM_BEEP_INTERVAL) - else: - freq = 800 if int(elapsed_minutes * 10) % 2 == 0 else 1200 - _play_on_all_sinks(freq) - _beep_loud(freq) - self._stop_beep.wait(LOUD_TOGGLE_INTERVAL) + self.root.after(FLASH_PERIOD_MS, self._flash_step) def run(self) -> None: """Start the alarm main loop, restoring hardware on every exit path.""" self._lock.run() -def _should_run_alarm() -> bool: - """Determine if the alarm should run right now.""" - if not _is_alarm_day(): - _logger.info("Not an alarm day. Exiting.") - return False - if was_alarm_dismissed_today(): - _logger.info("Alarm already dismissed today. Exiting.") - return False - if was_workout_logged_today(): - _logger.info("Workout already logged today. Skipping alarm.") - return False - return True - - -def _parse_args(argv: list[str]) -> argparse.Namespace: - """Parse CLI arguments for the alarm daemon.""" - parser = argparse.ArgumentParser(description="Wake alarm daemon.") - parser.add_argument( - "--production", - action="store_true", - help="Production mode (default; kept for systemd compatibility).", - ) - parser.add_argument( - "--demo", - action="store_true", - help="Run with a smaller window and shorter timers.", - ) - parser.add_argument( - "--trigger-now", - action="store_true", - help="Bypass the day/dismiss gate and fire the alarm immediately.", - ) - return parser.parse_args(argv) - - def main() -> None: """Entry point for the wake alarm daemon.""" configure_logging() - args = _parse_args(sys.argv[1:]) + args = parse_args(sys.argv[1:]) - if not args.trigger_now and not _should_run_alarm(): + if not args.trigger_now and not should_run_alarm(): return _logger.warning( @@ -462,14 +241,7 @@ def main() -> None: args.demo, args.trigger_now, ) - _warn_if_no_real_sink() - # Wake the display, then proceed immediately — no sleep hoping the monitor - # is "up". The monitor powers on in parallel; brightness and audio don't - # block on it, and the beep loop's per-cycle sink check picks up the HDMI - # output the moment it enumerates. - _wake_display() - _set_max_brightness() - turn_on_plug() + prepare_hardware() alarm = WakeAlarm(demo_mode=args.demo) alarm.run() diff --git a/wake_alarm/_alarm_state.py b/wake_alarm/_alarm_state.py new file mode 100644 index 0000000..4ce0b84 --- /dev/null +++ b/wake_alarm/_alarm_state.py @@ -0,0 +1,34 @@ +"""Mutable state the alarm carries while it is ringing. + +Split out of ``_alarm.py`` for the 250-line cap. Kept as plain dataclasses so +the alarm's own module stays about behaviour rather than bookkeeping. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import time +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from wake_alarm._challenges import _Challenge + from wake_alarm._sinks import _AudioState + + +@dataclass +class _AlarmProgress: + """Mutable dismiss-challenge progress state.""" + + current_challenge: _Challenge + started_at: float = field(default_factory=time.monotonic) + skip_earnable: bool = True + rounds_completed: int = 0 + flash_remaining: int = 0 + flash_on: bool = False + + +@dataclass +class _AlarmHardware: + """Hardware state captured at alarm start, restored when it closes.""" + + audio_restore: _AudioState | None diff --git a/wake_alarm/_alarm_timers.py b/wake_alarm/_alarm_timers.py new file mode 100644 index 0000000..e5da5c3 --- /dev/null +++ b/wake_alarm/_alarm_timers.py @@ -0,0 +1,109 @@ +"""Timers that run for the whole life of the alarm. + +Split out of ``_alarm.py`` for the 250-line cap, as a mixin rather than free +functions because every one of these reschedules itself through ``root.after`` +and needs the alarm's own view and progress state. + +Note ``_on_skip_window_expired`` deliberately does **not** stop the alarm: only +the workout-skip reward expires. The alarm keeps sounding until the challenge +is actually typed. +""" + +from __future__ import annotations + +import logging +import time +from typing import TYPE_CHECKING + +from wake_alarm._challenges import _make_challenge +from wake_alarm._rounds import ( + NO_SKIP_STATUS, + WINDOW_CLOSED_INFO, + countdown_text, + refresh_interval_ms, + remaining_seconds, + show_challenge, + skip_window_ms, + skip_window_seconds, +) + +if TYPE_CHECKING: # pragma: no cover + import tkinter as tk + + from wake_alarm._alarm_state import _AlarmProgress + from wake_alarm._alarm_view import _ViewGroup + from wake_alarm._escape_flow import EscapeHatch + +_logger = logging.getLogger(__name__) + + +class AlarmTimersMixin: + """Self-rescheduling timers for the challenge refresh and skip window.""" + + # Provided by WakeAlarm; declared so type checking sees them. + demo_mode: bool + root: tk.Misc + _active: bool + _progress: _AlarmProgress + _view: _ViewGroup + _escape: EscapeHatch + + def _start_flash_countdown(self) -> None: + """Restart the flash countdown; implemented by ``WakeAlarm``.""" + raise NotImplementedError # pragma: no cover + + def _schedule_code_refresh(self) -> None: + """Replace the current challenge periodically. + + Ensures the user can't simply wait out a hard challenge type — a new + random challenge is generated every DISMISS_CODE_REFRESH_SECONDS. + """ + if not self._active: + return + self._progress.current_challenge = _make_challenge() + show_challenge(self._view, self._progress.current_challenge) + if self._progress.current_challenge.kind == "flash": + self._start_flash_countdown() + self.root.after( + refresh_interval_ms(demo_mode=self.demo_mode), + self._schedule_code_refresh, + ) + + def _schedule_skip_window_close(self) -> None: + """Mark the workout-skip reward as expired after the allowed time.""" + self.root.after( + skip_window_ms(demo_mode=self.demo_mode), + self._on_skip_window_expired, + ) + + def _on_skip_window_expired(self) -> None: + """Skip window closed: keep the alarm running, deny the workout skip. + + The alarm intentionally does NOT stop here - it keeps beeping and + flashing until the user actually types the code. Only the workout-skip + reward expires; dismissing now silences the alarm without earning a skip. + """ + if not self._active: + return + self._progress.skip_earnable = False + self._view.info_label.configure(text=WINDOW_CLOSED_INFO) + self._view.status_label.configure(text=NO_SKIP_STATUS) + _logger.info("Skip window expired - alarm continues until dismissed.") + + def _update_timer(self) -> None: + """Show the skip-window countdown, then a keep-going silence prompt.""" + if not self._active: + return + elapsed = time.monotonic() - self._progress.started_at + remaining = remaining_seconds( + elapsed, + skip_window_seconds(demo_mode=self.demo_mode), + ) + self._view.timer_label.configure( + text=countdown_text( + remaining, + earnable=self._progress.skip_earnable, + ), + ) + self._escape.maybe_offer(elapsed) + self.root.after(1000, self._update_timer) diff --git a/wake_alarm/_audio.py b/wake_alarm/_audio.py index 3b08a78..10c0187 100644 --- a/wake_alarm/_audio.py +++ b/wake_alarm/_audio.py @@ -1,31 +1,26 @@ -"""Audio playback, fan control, and PipeWire sink management for the wake alarm.""" +"""Audio playback and PipeWire sink management for the wake alarm.""" from __future__ import annotations import logging -import math from pathlib import Path import shutil import struct import subprocess import sys -import tempfile import time -import wave from wake_alarm._sinks import ( _ensure_sink_audible, _list_real_sinks, _pactl_path, ) +from wake_alarm._tone import _ensure_tone_wav _logger = logging.getLogger(__name__) -_TONE_CACHE: dict[int, Path] = {} _TONE_TIMEOUT_SECONDS: float = 6.0 _TONE_DURATION_SECONDS: float = 1.5 -_TONE_FRAMERATE: int = 48000 -_TONE_AMPLITUDE: int = 32760 # near s16 max for the loudest sine we can emit # Number of back-to-back PC-speaker beeps per _play_tone call. # pcspkr volume is hardware-fixed, so we lean on repetition + duration to be @@ -42,26 +37,6 @@ _PCSPKR_SND_TONE: int = 0x02 # struct input_event: timeval (long sec, long usec), u16 type, u16 code, s32 val _PCSPKR_EVENT_FMT: str = "llHHi" -# NCT Super I/O chip names that expose a single pwm1 fan control channel. -_NCT_CHIP_NAMES: frozenset[str] = frozenset( - { - "nct6775", - "nct6779", - "nct6791", - "nct6792", - "nct6793", - "nct6795", - "nct6796", - "nct6797", - "nct6798", - "nct6799", - } -) - -# Installed by install.sh, controlled via sudoers NOPASSWD entry. -_FAN_SCRIPT: str = "/usr/local/bin/wake-alarm-fans.sh" -_SUDO_BIN: str = "/usr/bin/sudo" - def _beep_soft() -> None: """Play a soft system beep via terminal bell.""" @@ -117,29 +92,6 @@ def _beep_pcspkr(frequency: int, duration_seconds: float) -> None: ) -def _ensure_tone_wav(frequency: int) -> Path: - """Generate (and cache) a mono 48 kHz sine WAV at *frequency* Hz.""" - cached = _TONE_CACHE.get(frequency) - if cached is not None and cached.exists(): - return cached - path = Path(tempfile.gettempdir()) / f"wake_alarm_tone_{frequency}.wav" - n_frames = int(_TONE_FRAMERATE * _TONE_DURATION_SECONDS) - with wave.open(str(path), "wb") as wav: - wav.setnchannels(1) - wav.setsampwidth(2) - wav.setframerate(_TONE_FRAMERATE) - frames = bytearray() - for i in range(n_frames): - sample = int( - _TONE_AMPLITUDE - * math.sin(2 * math.pi * frequency * i / _TONE_FRAMERATE), - ) - frames.extend(struct.pack(" bool: """Run *binary* on *wav* with a generous timeout. Return True on success.""" path = shutil.which(binary) @@ -288,113 +240,6 @@ def _reap_player(proc: subprocess.Popen[bytes]) -> None: proc.wait() -def _find_fan_hwmon() -> str | None: - """Return the hwmon directory for an NCT fan controller, or None.""" - for name_path in Path("/sys/class/hwmon").glob("hwmon*/name"): - try: - chip = name_path.read_text().strip() - except OSError: - _logger.warning("Could not read %s", name_path, exc_info=True) - continue - if chip in _NCT_CHIP_NAMES: - return str(name_path.parent) - _logger.warning( - "No NCT super-I/O hwmon entry found; fan ramp will be skipped", - ) - return None - - -def _max_fans() -> bool: - """Ramp every NCT pwm channel to 100% speed via the helper script. - - The helper records prior state under /run/wake-alarm-fans.state so - _restore_fans() can put things back without arguments. Safe: higher fan - speed only lowers temperatures, never damages hardware. - - Returns: - True when the ramp script ran successfully, False otherwise. - """ - if _find_fan_hwmon() is None: - return False - try: - result = subprocess.run( - [_SUDO_BIN, "-n", _FAN_SCRIPT, "max"], - check=False, - capture_output=True, - timeout=5, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning( - "Fan script %s not runnable; skipping fan ramp", - _FAN_SCRIPT, - exc_info=True, - ) - return False - if result.returncode != 0: - _logger.warning( - "Fan script %s exited %d: %s", - _FAN_SCRIPT, - result.returncode, - result.stderr.decode(errors="replace").strip(), - ) - return False - return True - - -def _restore_fans(*, active: bool) -> None: - """Restore fan speed if _max_fans() previously succeeded.""" - if not active: - return - try: - subprocess.run( - [_SUDO_BIN, "-n", _FAN_SCRIPT, "restore"], - check=False, - capture_output=True, - timeout=5, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning( - "Failed to restore fan state via %s", - _FAN_SCRIPT, - exc_info=True, - ) - - -def _set_max_brightness() -> None: - """Set all connected monitors to maximum brightness via xrandr.""" - xrandr = shutil.which("xrandr") - if xrandr is None: - _logger.warning("xrandr not on PATH; skipping max-brightness") - return - try: - result = subprocess.run( - [xrandr, "--query"], - capture_output=True, - text=True, - timeout=5, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning("xrandr --query failed; skipping max-brightness", exc_info=True) - return - for line in result.stdout.splitlines(): - if " connected" in line: - output = line.split()[0] - try: - subprocess.run( - [xrandr, "--output", output, "--brightness", "1.0"], - check=False, - capture_output=True, - timeout=5, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning( - "Failed to set brightness on %s", - output, - exc_info=True, - ) - - def _beep_medium(frequency: int = 1000) -> None: """Play a medium beep (sine tone via paplay/aplay/speaker-test).""" _play_tone(frequency) diff --git a/wake_alarm/_beep_loop.py b/wake_alarm/_beep_loop.py new file mode 100644 index 0000000..be6793c --- /dev/null +++ b/wake_alarm/_beep_loop.py @@ -0,0 +1,99 @@ +"""The background thread that sounds the alarm. + +Split out of ``_alarm.py`` so the loop can be exercised without a Tk window, +and so the escalation decision itself stays in ``_escalation`` as a pure +function. + +The loop plays on **every** real sink each cycle rather than choosing one +output: a speaker that is switched off cannot be revived by software, so the +answer is to drive every other output in parallel. +""" + +from __future__ import annotations + +from collections.abc import Callable +import logging +import threading +import time + +from wake_alarm._audio import ( + _beep_loud, + _beep_medium, + _beep_soft, + _play_on_all_sinks, +) +from wake_alarm._escalation import BeepPlan, plan_for + +_Monotonic = Callable[[], float] + +_logger = logging.getLogger(__name__) + + +def _sound(plan: BeepPlan) -> None: + """Emit one cycle of audio for *plan*. + + Args: + plan: The decision for this iteration. + """ + if plan.frequency is None: + return + _play_on_all_sinks(plan.frequency) + if plan.phase == "soft": + _beep_soft() + elif plan.phase == "medium": + _beep_medium() + else: + _beep_loud(plan.frequency) + + +def run_beep_loop( + stop: threading.Event, + started_at: float, + *, + monotonic: _Monotonic = time.monotonic, +) -> None: + """Sound the alarm until *stop* is set. + + Args: + stop: Set on dismissal (and by the caller's exit paths). + started_at: Monotonic timestamp when the ring began. + monotonic: Clock source, injected for tests. + + The failsafe is measured from this side's own monotonic clock, never from + a timestamp compared across devices: a skewed phone clock must not be able + to silence the PC early, nor keep it sounding forever. + """ + announced_silence = False + while not stop.is_set(): + elapsed = monotonic() - started_at + plan = plan_for(elapsed) + if plan.silenced and not announced_silence: + announced_silence = True + _logger.warning( + "Failsafe reached after %.0fs: audio stops, but the lock and " + "challenge remain until dismissed.", + elapsed, + extra={"event": "failsafe_silence", "elapsed_s": round(elapsed)}, + ) + _sound(plan) + stop.wait(plan.interval) + + +def start_beep_thread(stop: threading.Event, started_at: float) -> threading.Thread: + """Run :func:`run_beep_loop` on a daemon thread. + + Args: + stop: Set on dismissal. + started_at: Monotonic timestamp when the ring began. + + Returns: + The started thread. + """ + thread = threading.Thread( + target=run_beep_loop, + args=(stop, started_at), + daemon=True, + name="wake-alarm-beep", + ) + thread.start() + return thread diff --git a/wake_alarm/_brightness.py b/wake_alarm/_brightness.py new file mode 100644 index 0000000..3ee05ae --- /dev/null +++ b/wake_alarm/_brightness.py @@ -0,0 +1,48 @@ +"""Raise every monitor to full brightness when the alarm fires. + +Split out of ``_audio.py`` for the 250-line cap; it is display control, not +audio, and only ever lived there for historical reasons. +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess + +_logger = logging.getLogger(__name__) + + +def _set_max_brightness() -> None: + """Set all connected monitors to maximum brightness via xrandr.""" + xrandr = shutil.which("xrandr") + if xrandr is None: + _logger.warning("xrandr not on PATH; skipping max-brightness") + return + try: + result = subprocess.run( + [xrandr, "--query"], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("xrandr --query failed; skipping max-brightness", exc_info=True) + return + for line in result.stdout.splitlines(): + if " connected" in line: + output = line.split()[0] + try: + subprocess.run( + [xrandr, "--output", output, "--brightness", "1.0"], + check=False, + capture_output=True, + timeout=5, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning( + "Failed to set brightness on %s", + output, + exc_info=True, + ) diff --git a/wake_alarm/_constants.py b/wake_alarm/_constants.py index da0a169..c109b02 100644 --- a/wake_alarm/_constants.py +++ b/wake_alarm/_constants.py @@ -89,6 +89,12 @@ ESCAPE_HISTORY_FILE: Path = ( # exactly when the alarm had just stopped being annoying. ESCAPE_OFFER_AFTER_SECONDS: int = 900 +# Pause between powering the smart plug and the first beep. A powered +# speaker's soft-start / mute relay typically delays output 1-5 s after mains +# is applied, so firing the plug and the first tone together loses the opening +# of the alarm. Costs nothing when no plug is configured. +PLUG_PREROLL_SECONDS: float = 4.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/_countdown.py b/wake_alarm/_countdown.py new file mode 100644 index 0000000..8e748ab --- /dev/null +++ b/wake_alarm/_countdown.py @@ -0,0 +1,92 @@ +"""Format the skip-window countdown shown on the lock screen. + +Split out of ``_alarm.py`` for the 250-line cap, and because the wording is +load-bearing: once the window closes the label must stop implying a reward is +still available while making clear the alarm has *not* stopped. +""" + +from __future__ import annotations + +from wake_alarm._constants import DISMISS_CODE_REFRESH_SECONDS, DISMISS_WINDOW_MINUTES + +# Demo mode compresses the schedule so a manual test does not take half an +# hour, without changing any production timing. +DEMO_REFRESH_MS = 10_000 +DEMO_WINDOW_MS = 30_000 +DEMO_WINDOW_SECONDS = 30.0 + +NO_SKIP_TEXT = "No skip available - type the code to stop the alarm" +WINDOW_CLOSED_INFO = "Skip window closed - type the code to stop the alarm" +NO_SKIP_STATUS = "No workout skip today." + + +def remaining_seconds(elapsed: float, window: float) -> float: + """Seconds left in the skip window, never negative. + + Args: + elapsed: Seconds since the alarm started. + window: Total length of the skip window in seconds. + + Returns: + Remaining seconds, floored at zero. + """ + return max(0.0, window - elapsed) + + +def countdown_text(remaining: float, *, earnable: bool) -> str: + """Build the timer label for the current state. + + Args: + remaining: Seconds left in the skip window. + earnable: Whether a skip can still be earned. + + Returns: + The label text. + """ + if not earnable or remaining <= 0: + return NO_SKIP_TEXT + minutes = int(remaining) // 60 + seconds = int(remaining) % 60 + return f"Skip window: {minutes:02d}:{seconds:02d}" + + +def refresh_interval_ms(*, demo_mode: bool) -> int: + """Milliseconds between challenge refreshes. + + Args: + demo_mode: True for the shortened manual-test schedule. + + Returns: + The delay to pass to ``after()``. + """ + if demo_mode: + return DEMO_REFRESH_MS + return DISMISS_CODE_REFRESH_SECONDS * 1000 + + +def skip_window_ms(*, demo_mode: bool) -> int: + """Milliseconds until the workout-skip reward expires. + + Args: + demo_mode: True for the shortened manual-test schedule. + + Returns: + The delay to pass to ``after()``. + """ + if demo_mode: + return DEMO_WINDOW_MS + return DISMISS_WINDOW_MINUTES * 60 * 1000 + + +def skip_window_seconds(*, demo_mode: bool) -> float: + """Length of the skip window in seconds. + + Args: + demo_mode: True for the shortened manual-test schedule. + + Returns: + The window length. + """ + if demo_mode: + return DEMO_WINDOW_SECONDS + return float(DISMISS_WINDOW_MINUTES * 60) diff --git a/wake_alarm/_dismiss_view.py b/wake_alarm/_dismiss_view.py new file mode 100644 index 0000000..bbd1371 --- /dev/null +++ b/wake_alarm/_dismiss_view.py @@ -0,0 +1,58 @@ +"""Render the post-dismissal banner across every monitor. + +Split out of ``_alarm.py`` for the 250-line cap. Pure presentation: deciding +*whether* a skip was earned stays in the alarm, this only draws the outcome. +""" + +from __future__ import annotations + +import tkinter as tk +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable + + from gatelock import LockConfig + +SKIP_EARNED_MESSAGE = "Workout skip earned! Enjoy your morning." +NO_SKIP_MESSAGE = "Alarm dismissed. No workout skip." + +# How long the banner stays up before the lock closes itself. +BANNER_LINGER_MS = 3000 + + +def clear_surfaces(containers: Iterable[tk.Misc]) -> None: + """Remove every child widget from each per-output container. + + Args: + containers: Per-output container widgets. + """ + for container in containers: + for widget in container.winfo_children(): + widget.destroy() + + +def show_dismissal( + containers: Iterable[tk.Misc], + config: LockConfig, + *, + earned_skip: bool, +) -> None: + """Replace the challenge UI with the outcome message on every monitor. + + Args: + containers: Per-output container widgets. + config: Styling source for fonts, colours and spacing. + earned_skip: Whether the workout skip was earned. + """ + clear_surfaces(containers) + message = SKIP_EARNED_MESSAGE if earned_skip else NO_SKIP_MESSAGE + colour = config.success if earned_skip else config.warning + for container in containers: + tk.Label( + container, + text=message, + font=config.font("display", bold=True), + fg=colour, + bg=config.bg, + ).pack(pady=config.space("lg")) diff --git a/wake_alarm/_gating.py b/wake_alarm/_gating.py new file mode 100644 index 0000000..d2fb6fd --- /dev/null +++ b/wake_alarm/_gating.py @@ -0,0 +1,81 @@ +"""Decide whether the alarm should run, and parse its CLI. + +Split out of ``_alarm.py`` so the gate is testable without constructing a +window, and to keep that module under the 250-line cap. + +Note ``_should_run_alarm`` has deliberately **no time gate**: the trigger owns +"is it time?", and this owns "should it run at all today?". That separation is +why ``wake-alarm.service`` must never get an ``[Install]`` section -- with no +time gate, a ``WantedBy=`` would fire the alarm at login. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import logging + +from wake_alarm._constants import ALARM_DAYS +from wake_alarm._state import was_alarm_dismissed_today, was_workout_logged_today + +_logger = logging.getLogger(__name__) + + +def is_alarm_day(*, now: datetime | None = None) -> bool: + """Report whether today is configured as an alarm day. + + Args: + now: Instant to test; defaults to the current local time. + + Returns: + True when today's weekday is in ``ALARM_DAYS``. + """ + moment = now if now is not None else datetime.now(tz=timezone.utc).astimezone() + return moment.weekday() in ALARM_DAYS + + +def should_run_alarm() -> bool: + """Decide whether the alarm should sound at all today. + + Returns: + True unless today is not an alarm day, the alarm was already + dismissed, or the workout was already logged. + """ + if not is_alarm_day(): + _logger.info("Not an alarm day. Exiting.") + return False + if was_alarm_dismissed_today(): + _logger.info("Alarm already dismissed today. Exiting.") + return False + if was_workout_logged_today(): + _logger.info("Workout already logged today. Skipping alarm.") + return False + return True + + +def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse CLI arguments for the alarm daemon. + + Args: + argv: Argument list without the program name. + + Returns: + The parsed namespace. + """ + parser = argparse.ArgumentParser(description="Wake alarm daemon.") + parser.add_argument( + "--production", + action="store_true", + help="Production mode (default; kept for systemd compatibility).", + ) + parser.add_argument( + "--demo", + action="store_true", + help="Run with a smaller window and shorter timers.", + ) + parser.add_argument( + "--trigger-now", + action="store_true", + help="Bypass the day/dismiss gate and fire the alarm immediately.", + ) + return parser.parse_args(argv) diff --git a/wake_alarm/_lock_setup.py b/wake_alarm/_lock_setup.py new file mode 100644 index 0000000..ccc51c6 --- /dev/null +++ b/wake_alarm/_lock_setup.py @@ -0,0 +1,55 @@ +"""Build the gatelock configuration and claim the screen-lock arbiter. + +Split out of ``_alarm.py`` for the 250-line cap. + +Production hard-locks; demo stays grab-free so a manual test cannot trap the +developer. Demo still takes ``overrideredirect``, because per-output placement +is impossible without it -- a window manager rewrites a managed window's +geometry. +""" + +from __future__ import annotations + +from gatelock import RANK_WAKE_ALARM, Arbiter, LockConfig + +APP_NAME = "wake_alarm" + + +def build_config(*, demo_mode: bool) -> LockConfig: + """Build the lock configuration for this run. + + Args: + demo_mode: True for the closeable manual-test window. + + Returns: + The configuration to hand to :class:`LockWindow`. + """ + return LockConfig( + mode="soft" if demo_mode else "hard", + overrideredirect=True, + app_name=APP_NAME, + rank=RANK_WAKE_ALARM, + ) + + +def claim_arbiter(config: LockConfig) -> Arbiter: + """Publish and acquire the arbiter for *config*. + + The caller must hand the result to :class:`LockWindow`, which takes + ownership and releases it on close. + + Args: + config: The lock configuration being used. + + Returns: + The acquired arbiter. + """ + arbiter = Arbiter( + APP_NAME, + RANK_WAKE_ALARM, + grab=config.resolved_grab(), + disable_vt=config.resolved_disable_vt(), + ) + arbiter.publish() + arbiter.acquire_holder() + return arbiter diff --git a/wake_alarm/_rounds.py b/wake_alarm/_rounds.py new file mode 100644 index 0000000..7a7ad09 --- /dev/null +++ b/wake_alarm/_rounds.py @@ -0,0 +1,120 @@ +"""Present a dismiss challenge and judge the answer. + +Split out of ``_alarm.py`` for the 250-line cap. Everything here manipulates +the view for one round; the alarm keeps ownership of *how many* rounds are +needed and what happens when they are all done. +""" + +from __future__ import annotations + +import tkinter as tk +from typing import TYPE_CHECKING + +from wake_alarm._countdown import ( + NO_SKIP_STATUS, + WINDOW_CLOSED_INFO, + countdown_text, + refresh_interval_ms, + remaining_seconds, + skip_window_ms, + skip_window_seconds, +) +from wake_alarm._dismiss_view import BANNER_LINGER_MS, show_dismissal +from wake_alarm._screen_flash import FLASH_PERIOD_MS, apply_flash_colour + +__all__ = [ + "BANNER_LINGER_MS", + "FLASH_PERIOD_MS", + "NO_SKIP_STATUS", + "WINDOW_CLOSED_INFO", + "apply_flash_colour", + "countdown_text", + "hide_flash_code", + "normalise", + "refresh_interval_ms", + "remaining_seconds", + "show_challenge", + "show_dismissal", + "show_flash_countdown", + "show_round_progress", + "show_wrong_answer", + "skip_window_ms", + "skip_window_seconds", +] + +if TYPE_CHECKING: + from wake_alarm._alarm_view import _ViewGroup + from wake_alarm._challenges import _Challenge + + +def normalise(entered: str) -> str: + """Canonicalise typed input before comparison. + + Args: + entered: Raw contents of the entry widget. + + Returns: + Trimmed, upper-cased text. + """ + return entered.strip().upper() + + +def show_challenge(view: _ViewGroup, challenge: _Challenge) -> None: + """Display *challenge* and clear the entry ready for the answer. + + Args: + view: The multi-monitor view group. + challenge: The challenge to present. + """ + view.code_label.configure(text=challenge.display, fg=view.config.success) + view.info_label.configure(text=challenge.hint) + view.entry.delete(0, tk.END) + + +def show_wrong_answer(view: _ViewGroup, challenge: _Challenge) -> None: + """Report an incorrect answer and re-present the same challenge. + + Args: + view: The multi-monitor view group. + challenge: The challenge that was answered incorrectly. + """ + view.status_label.configure(text="Wrong! Try again.") + view.entry.delete(0, tk.END) + if challenge.kind == "flash": + view.code_label.configure(text=challenge.display, fg=view.config.success) + + +def show_flash_countdown(view: _ViewGroup, remaining: int) -> None: + """Show how long the flash challenge stays visible. + + Args: + view: The multi-monitor view group. + remaining: Seconds before the code is hidden. + """ + view.status_label.configure(text=f"Memorise! Hiding in {remaining}s…") + + +def hide_flash_code(view: _ViewGroup, challenge: _Challenge) -> None: + """Mask the flash code and prompt for it from memory. + + Args: + view: The multi-monitor view group. + challenge: The challenge whose display is being hidden. + """ + view.code_label.configure( + text="?" * len(challenge.display), + fg=view.config.muted, + ) + view.status_label.configure(text="Now type the code from memory!") + + +def show_round_progress(view: _ViewGroup, *, completed: int, required: int) -> None: + """Update the round counter and the encouragement line. + + Args: + view: The multi-monitor view group. + completed: How many rounds are done. + required: How many rounds are needed in total. + """ + view.round_label.configure(text=f"Round {completed + 1} / {required}") + view.status_label.configure(text=f"Round {completed} done — keep going!") diff --git a/wake_alarm/_screen_flash.py b/wake_alarm/_screen_flash.py new file mode 100644 index 0000000..5d0b239 --- /dev/null +++ b/wake_alarm/_screen_flash.py @@ -0,0 +1,42 @@ +"""Alternate the lock screen's background colour to hold attention. + +Split out of ``_alarm.py`` for the 250-line cap. The 750 ms period is +deliberate: it is ~1.3 Hz, comfortably under the ~3 Hz threshold associated +with photosensitive seizures. Do not shorten it to make the alarm "more +urgent" -- the escalating audio is the right lever for that. +""" + +from __future__ import annotations + +import contextlib +import tkinter as tk +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable + +# Milliseconds between colour swaps. ~1.3 Hz; see the module docstring. +FLASH_PERIOD_MS = 750 + + +def apply_flash_colour( + root: tk.Tk, + containers: Iterable[tk.Frame], + colour: str, +) -> None: + """Paint *colour* onto the backdrop and every per-output surface. + + The surfaces sit on top of the root backdrop, so recolouring only the root + would be invisible on every monitor. + + Args: + root: The root window holding the backdrop. + containers: Per-output container widgets. + colour: The background colour to apply. + """ + root.configure(bg=colour) + for container in containers: + # A surface torn down mid-flash must not stop the others repainting. + with contextlib.suppress(tk.TclError): + container.configure(bg=colour) + container.winfo_toplevel().configure(bg=colour) diff --git a/wake_alarm/_startup.py b/wake_alarm/_startup.py new file mode 100644 index 0000000..bbc582e --- /dev/null +++ b/wake_alarm/_startup.py @@ -0,0 +1,47 @@ +"""Bring the machine's hardware up before the alarm window appears. + +Split out of ``_alarm.py`` for the 250-line cap. + +Nothing here blocks waiting for the monitor. It powers things on and returns; +the beep loop's per-cycle sink check and ``_reactivate``'s retry schedule pick +up outputs as they enumerate, which is a real presence check rather than a +timed guess. +""" + +from __future__ import annotations + +from collections.abc import Callable +import logging +import time + +from wake_alarm._alarm_display import _wake_display +from wake_alarm._brightness import _set_max_brightness +from wake_alarm._constants import PLUG_PREROLL_SECONDS +from wake_alarm._sinks import _warn_if_no_real_sink +from wake_alarm._smart_plug import turn_on_plug + +_logger = logging.getLogger(__name__) + +_Sleep = Callable[[float], object] + + +def prepare_hardware(*, sleep: _Sleep = time.sleep) -> None: + """Wake the display, raise brightness and power the speakers. + + Args: + sleep: Delay function, injected so tests do not really wait. + """ + _warn_if_no_real_sink() + _wake_display() + _set_max_brightness() + # Power the speakers BEFORE the first beep. An amplifier's soft-start / + # mute relay swallows the first seconds of audio after mains is applied, + # so firing the plug and the first tone together loses the opening of the + # alarm entirely. Costs nothing when no plug is configured. + turn_on_plug() + _logger.info( + "Waiting %.1fs for speaker soft-start before the first beep", + PLUG_PREROLL_SECONDS, + extra={"event": "plug_preroll", "seconds": PLUG_PREROLL_SECONDS}, + ) + sleep(PLUG_PREROLL_SECONDS) diff --git a/wake_alarm/_tone.py b/wake_alarm/_tone.py new file mode 100644 index 0000000..7dd487e --- /dev/null +++ b/wake_alarm/_tone.py @@ -0,0 +1,48 @@ +"""Synthesise and cache the alarm tone. + +Split out of ``_audio.py`` for the 250-line cap. + +Note the pylint ``generated-members`` entries for ``wave.Wave_write`` in +pyproject.toml: pylint infers the read-mode overload of ``wave.open`` and +wrongly reports ``setnchannels`` and friends as missing. Keep them. +""" + +from __future__ import annotations + +import logging +import math +from pathlib import Path +import struct +import tempfile +import wave + +_logger = logging.getLogger(__name__) + +# Generated WAVs are reused for the life of the process. +_TONE_CACHE: dict[int, Path] = {} +_TONE_DURATION_SECONDS: float = 1.5 +_TONE_FRAMERATE: int = 48000 +_TONE_AMPLITUDE: int = 32760 # near s16 max for the loudest sine we can emit + + +def _ensure_tone_wav(frequency: int) -> Path: + """Generate (and cache) a mono 48 kHz sine WAV at *frequency* Hz.""" + cached = _TONE_CACHE.get(frequency) + if cached is not None and cached.exists(): + return cached + path = Path(tempfile.gettempdir()) / f"wake_alarm_tone_{frequency}.wav" + n_frames = int(_TONE_FRAMERATE * _TONE_DURATION_SECONDS) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(_TONE_FRAMERATE) + frames = bytearray() + for i in range(n_frames): + sample = int( + _TONE_AMPLITUDE + * math.sin(2 * math.pi * frequency * i / _TONE_FRAMERATE), + ) + frames.extend(struct.pack(" Generator[MagicMock]: 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", @@ -75,6 +70,7 @@ def mock_tk_module() -> Generator[MagicMock]: 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", @@ -232,204 +228,3 @@ class TestBeepFunctions: with patch("wake_alarm._audio._play_tone") as mock_play: _beep_loud(frequency=1200) mock_play.assert_called_once_with(1200) - - -class TestShouldRunAlarm: - """Tests for _should_run_alarm.""" - - def test_returns_false_on_non_alarm_day(self) -> None: - """Return False when today is not an alarm day.""" - with patch( - "wake_alarm._alarm._is_alarm_day", - return_value=False, - ): - assert _should_run_alarm() is False - - def test_returns_false_when_already_dismissed(self) -> None: - """Return False when alarm was already dismissed today.""" - with ( - patch( - "wake_alarm._alarm._is_alarm_day", - return_value=True, - ), - patch( - "wake_alarm._alarm.was_alarm_dismissed_today", - return_value=True, - ), - ): - assert _should_run_alarm() is False - - def test_returns_true_when_alarm_day_and_not_dismissed(self) -> None: - """Return True when today is alarm day and not yet dismissed.""" - with ( - patch( - "wake_alarm._alarm._is_alarm_day", - return_value=True, - ), - patch( - "wake_alarm._alarm.was_alarm_dismissed_today", - return_value=False, - ), - patch( - "wake_alarm._alarm.was_workout_logged_today", - return_value=False, - ), - ): - assert _should_run_alarm() is True - - def test_returns_false_when_workout_already_logged(self) -> None: - """Return False when workout was already logged today.""" - with ( - patch( - "wake_alarm._alarm._is_alarm_day", - return_value=True, - ), - patch( - "wake_alarm._alarm.was_alarm_dismissed_today", - return_value=False, - ), - patch( - "wake_alarm._alarm.was_workout_logged_today", - return_value=True, - ), - ): - assert _should_run_alarm() is False - - -class TestFindFanHwmon: - """Tests for _find_fan_hwmon.""" - - def test_returns_none_when_no_hwmon(self, monkeypatch: pytest.MonkeyPatch) -> None: - """No hwmon entries → returns None.""" - monkeypatch.setattr(pathlib.Path, "glob", lambda _s, _p: iter([])) - assert _find_fan_hwmon() is None - - def test_returns_none_for_unknown_chip( - self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Non-NCT chip name → returns None.""" - name_file = tmp_path / "name" - name_file.write_text("unknown_chip\n") - monkeypatch.setattr(pathlib.Path, "glob", lambda _s, _p: iter([name_file])) - assert _find_fan_hwmon() is None - - def test_returns_hwmon_dir_for_nct_chip( - self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch - ) -> None: - """NCT chip name → returns the hwmon directory path.""" - name_file = tmp_path / "name" - name_file.write_text("nct6799\n") - monkeypatch.setattr(pathlib.Path, "glob", lambda _s, _p: iter([name_file])) - result = _find_fan_hwmon() - assert result == str(tmp_path) - - def test_skips_unreadable_name_file(self, monkeypatch: pytest.MonkeyPatch) -> None: - """OSError on read → skips and returns None.""" - bad_path = MagicMock(spec=pathlib.Path) - bad_path.read_text.side_effect = OSError("unreadable") - monkeypatch.setattr(pathlib.Path, "glob", lambda _s, _p: iter([bad_path])) - assert _find_fan_hwmon() is None - - -class TestMaxFans: - """Tests for _max_fans.""" - - def test_returns_false_when_no_hwmon(self) -> None: - """No fan controller → returns False immediately.""" - with patch("wake_alarm._audio._find_fan_hwmon", return_value=None): - assert _max_fans() is False - - def test_returns_false_on_script_oserror(self, tmp_path: pathlib.Path) -> None: - """OSError running fan script → returns False.""" - with ( - patch( - "wake_alarm._audio._find_fan_hwmon", - return_value=str(tmp_path), - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=OSError("not found"), - ), - ): - assert _max_fans() is False - - def test_returns_false_on_script_timeout(self, tmp_path: pathlib.Path) -> None: - """TimeoutExpired running fan script → returns False.""" - with ( - patch( - "wake_alarm._audio._find_fan_hwmon", - return_value=str(tmp_path), - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=subprocess.TimeoutExpired("fan", 5), - ), - ): - assert _max_fans() is False - - def test_returns_false_on_nonzero_returncode(self, tmp_path: pathlib.Path) -> None: - """Fan script exits non-zero → returns False.""" - mock_result = MagicMock() - mock_result.returncode = 1 - with ( - patch( - "wake_alarm._audio._find_fan_hwmon", - return_value=str(tmp_path), - ), - patch( - "wake_alarm._audio.subprocess.run", - return_value=mock_result, - ), - ): - assert _max_fans() is False - - def test_returns_true_on_success(self, tmp_path: pathlib.Path) -> None: - """Successful run → returns True (state is saved by the helper).""" - mock_result = MagicMock() - mock_result.returncode = 0 - with ( - patch( - "wake_alarm._audio._find_fan_hwmon", - return_value=str(tmp_path), - ), - patch( - "wake_alarm._audio.subprocess.run", - return_value=mock_result, - ), - ): - assert _max_fans() is True - - -class TestRestoreFans: - """Tests for _restore_fans.""" - - def test_noop_when_inactive(self) -> None: - """False state → subprocess.run is never called.""" - with patch("wake_alarm._audio.subprocess.run") as mock_run: - _restore_fans(active=False) - mock_run.assert_not_called() - - def test_calls_fan_script_restore(self) -> None: - """Active state → fan script called with restore (no args).""" - with patch("wake_alarm._audio.subprocess.run") as mock_run: - mock_run.return_value.returncode = 0 - _restore_fans(active=True) - mock_run.assert_called_once() - args = mock_run.call_args[0][0] - assert "restore" in args - - def test_ignores_oserror_on_restore(self) -> None: - """OSError from fan script is silently suppressed.""" - with patch( - "wake_alarm._audio.subprocess.run", - side_effect=OSError("no script"), - ): - _restore_fans(active=True) # must not raise - - def test_ignores_timeout_on_restore(self) -> None: - """TimeoutExpired from fan script is silently suppressed.""" - with patch( - "wake_alarm._audio.subprocess.run", - side_effect=subprocess.TimeoutExpired("fan", 5), - ): - _restore_fans(active=True) # must not raise diff --git a/wake_alarm/tests/test_alarm_audio.py b/wake_alarm/tests/test_alarm_audio.py index bec9f7a..86bdc99 100644 --- a/wake_alarm/tests/test_alarm_audio.py +++ b/wake_alarm/tests/test_alarm_audio.py @@ -13,166 +13,11 @@ if TYPE_CHECKING: import pathlib from wake_alarm._audio import ( - _beep_pcspkr, - _ensure_tone_wav, _play_tone, - _set_max_brightness, _try_player, ) -class TestSetMaxBrightness: - """Tests for _set_max_brightness.""" - - def test_noop_when_xrandr_missing(self) -> None: - """No xrandr on PATH → subprocess.run never called.""" - with ( - patch("wake_alarm._audio.shutil.which", return_value=None), - patch("wake_alarm._audio.subprocess.run") as mock_run, - ): - _set_max_brightness() - mock_run.assert_not_called() - - def test_noop_on_oserror_from_query(self) -> None: - """OSError from xrandr --query is suppressed.""" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/xrandr", - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=OSError("no display"), - ), - ): - _set_max_brightness() # must not raise - - def test_noop_on_timeout_from_query(self) -> None: - """TimeoutExpired from xrandr --query is suppressed.""" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/xrandr", - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=subprocess.TimeoutExpired("xrandr", 5), - ), - ): - _set_max_brightness() # must not raise - - def test_sets_brightness_for_connected_displays(self) -> None: - """Connected displays each get an --output --brightness call.""" - mock_query_result = MagicMock() - mock_query_result.stdout = ( - "HDMI-0 connected 2560x1440+0+0\nDP-0 connected primary\n" - ) - call_args_list: list[list[str]] = [] - - def fake_run(args: list[str], **_kwargs: object) -> MagicMock: - call_args_list.append(args) - return mock_query_result - - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/xrandr", - ), - patch("wake_alarm._audio.subprocess.run", side_effect=fake_run), - ): - _set_max_brightness() - - # First call is --query; subsequent calls set brightness for each output. - brightness_calls = [a for a in call_args_list if "--brightness" in a] - expected_brightness_calls = 2 - assert len(brightness_calls) == expected_brightness_calls - - def test_skips_disconnected_outputs(self) -> None: - """Disconnected outputs do NOT get a brightness call.""" - mock_result = MagicMock() - mock_result.stdout = "Screen 0: minimum 320\nHDMI-0 disconnected\n" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/xrandr", - ), - patch( - "wake_alarm._audio.subprocess.run", - return_value=mock_result, - ) as mock_run, - ): - _set_max_brightness() - # Only the --query call, no brightness calls. - assert mock_run.call_count == 1 - - def test_warns_when_brightness_call_fails(self) -> None: - """OSError on per-output --brightness call is logged but swallowed.""" - query_result = MagicMock() - query_result.stdout = ( - "Screen 0: minimum 320\nHDMI-0 connected primary 1920x1080\n" - ) - - def _run_side_effect(args: list[str], **_kwargs: object) -> MagicMock: - if "--query" in args: - return query_result - msg = "permission denied" - raise OSError(msg) - - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/xrandr", - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=_run_side_effect, - ), - ): - _set_max_brightness() # must not raise - - -class TestEnsureToneWav: - """Tests for _ensure_tone_wav (sine WAV generator + cache).""" - - def test_generates_and_caches(self, tmp_path: pathlib.Path) -> None: - """First call generates the WAV; second call returns the cached path.""" - from wake_alarm import _audio as alarm_mod - - alarm_mod._TONE_CACHE.clear() - with patch( - "wake_alarm._audio.tempfile.gettempdir", - return_value=str(tmp_path), - ): - path1 = _ensure_tone_wav(440) - assert path1.exists() - size = path1.stat().st_size - assert size > 0 - # Second call must hit the cache (no regeneration). - with patch("wake_alarm._audio.wave.open") as mock_open: - path2 = _ensure_tone_wav(440) - mock_open.assert_not_called() - assert path2 == path1 - alarm_mod._TONE_CACHE.clear() - - def test_regenerates_when_cached_file_missing( - self, - tmp_path: pathlib.Path, - ) -> None: - """If the cached file was deleted, regenerate it.""" - from wake_alarm._audio import _TONE_CACHE - - _TONE_CACHE.clear() - with patch( - "wake_alarm._audio.tempfile.gettempdir", - return_value=str(tmp_path), - ): - path1 = _ensure_tone_wav(880) - path1.unlink() - path2 = _ensure_tone_wav(880) - assert path2.exists() - _TONE_CACHE.clear() - - class TestTryPlayer: """Tests for _try_player.""" @@ -262,37 +107,6 @@ class TestTryPlayer: assert _try_player("paplay", wav) is False -class TestBeepPcspkr: - """Tests for _beep_pcspkr (evdev PC speaker).""" - - def test_writes_tone_then_zero_to_device(self) -> None: - """Successful path writes start-frequency then stop event.""" - - mock_dev = MagicMock() - mock_open_ctx = MagicMock() - mock_open_ctx.__enter__.return_value = mock_dev - mock_open_ctx.__exit__.return_value = False - with ( - patch( - "wake_alarm._audio.Path.open", - return_value=mock_open_ctx, - ), - patch("wake_alarm._audio.time.sleep"), - ): - _beep_pcspkr(1000, 0.05) - # First write carries the frequency, second write carries 0 (stop). - assert mock_dev.write.call_count == 2 - - def test_oserror_is_swallowed(self) -> None: - """OSError opening the device must not raise.""" - - with patch( - "wake_alarm._audio.Path.open", - side_effect=OSError("no device"), - ): - _beep_pcspkr(1000, 0.05) # must not raise - - class TestPlayTone: """Tests for _play_tone.""" diff --git a/wake_alarm/tests/test_alarm_beep_phases.py b/wake_alarm/tests/test_alarm_beep_phases.py deleted file mode 100644 index e9e522a..0000000 --- a/wake_alarm/tests/test_alarm_beep_phases.py +++ /dev/null @@ -1,161 +0,0 @@ -"""Tests for the beep escalation loop's phase selection.""" - -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, -) -from wake_alarm._constants import ( - PHASE_MEDIUM_END, - PHASE_SOFT_END, -) - - -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._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 calls for extra ALSA devices and hardware.""" - with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), - patch("wake_alarm._alarm._restore_display"), - patch("wake_alarm._alarm._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._alarm.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._alarm_view.tk", mock), - patch( - "wake_alarm._alarm.GateRoot", - return_value=mock.Tk.return_value, - ), - ): - yield mock - - -class TestBeepLoopPhases: - """Tests for different beep loop escalation phases.""" - - def test_medium_phase( - self, - mock_tk_module: MagicMock, - ) -> None: - """Beep loop enters medium phase after PHASE_SOFT_END minutes.""" - alarm = WakeAlarm(demo_mode=True) - # Set alarm start to make elapsed > PHASE_SOFT_END minutes - import time as time_mod - - # Midpoint of the medium band, not "one past the soft boundary": the - # phases are tuned to fit inside the failsafe window, so an offset - # anchored to one edge can land in the next phase entirely. - midpoint = (PHASE_SOFT_END + PHASE_MEDIUM_END) / 2 - alarm._progress.started_at = time_mod.monotonic() - midpoint * 60 - - call_count = 0 - - def stop_after_one(*_args: object, **_kwargs: object) -> None: - nonlocal call_count - call_count += 1 - if call_count >= 1: - alarm._stop_beep.set() - - with ( - patch( - "wake_alarm._alarm._beep_medium", - side_effect=stop_after_one, - ) as mock_beep, - # Any other rung also ends the loop, so a mistuned phase boundary - # fails this test instead of spinning forever on the real audio - # stack -- which previously hung the entire suite. - patch("wake_alarm._alarm._beep_soft", side_effect=stop_after_one), - patch("wake_alarm._alarm._beep_loud", side_effect=stop_after_one), - patch("wake_alarm._alarm._play_on_all_sinks"), - ): - alarm._beep_loop() - - mock_beep.assert_called() - alarm._stop_beep.set() - - def test_loud_phase( - self, - mock_tk_module: MagicMock, - ) -> None: - """Beep loop enters loud phase after PHASE_MEDIUM_END minutes.""" - alarm = WakeAlarm(demo_mode=True) - import time as time_mod - - alarm._progress.started_at = time_mod.monotonic() - (PHASE_MEDIUM_END + 1) * 60 - - call_count = 0 - - def stop_after_one(*_args: object, **_kwargs: object) -> None: - nonlocal call_count - call_count += 1 - if call_count >= 1: - alarm._stop_beep.set() - - with ( - patch( - "wake_alarm._alarm._beep_loud", - side_effect=stop_after_one, - ) as mock_beep, - # See the note in test_medium_phase: any rung must end the loop. - patch("wake_alarm._alarm._beep_soft", side_effect=stop_after_one), - patch("wake_alarm._alarm._beep_medium", side_effect=stop_after_one), - patch("wake_alarm._alarm._play_on_all_sinks"), - ): - alarm._beep_loop() - - mock_beep.assert_called() - alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_dismiss.py b/wake_alarm/tests/test_alarm_dismiss.py new file mode 100644 index 0000000..763bbc4 --- /dev/null +++ b/wake_alarm/tests/test_alarm_dismiss.py @@ -0,0 +1,232 @@ +"""Tests for the dismiss-challenge rounds.""" + +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) +# --------------------------------------------------------------------------- + + +# 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 + # Distinct frames and buttons per call: a shared return_value would make + # "every monitor got one" indistinguishable from "one monitor got two". + mock.Frame.side_effect = lambda *_args, **_kwargs: MagicMock() + mock.Button.side_effect = lambda *_args, **_kwargs: 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 + + +class TestWakeAlarmDismiss: + """Tests for alarm dismiss logic.""" + + def test_correct_code_dismisses_after_all_rounds( + self, + mock_tk_module: MagicMock, + ) -> None: + """Entering the correct answer for every required round dismisses the alarm.""" + from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED + + alarm = WakeAlarm(demo_mode=True) + + with patch( + "wake_alarm._alarm.save_wake_state", + ) as mock_save: + for _ in range(DISMISS_ROUNDS_REQUIRED): + alarm._view.entry_var.get.return_value = ( + alarm._progress.current_challenge.answer + ) + alarm._on_submit() + + assert alarm.dismissed is True + mock_save.assert_called_once() + assert mock_save.call_args[1]["skip_workout"] is True + alarm._stop_beep.set() + + def test_first_round_correct_does_not_dismiss( + self, + mock_tk_module: MagicMock, + ) -> None: + """A single correct entry is not enough — DISMISS_ROUNDS_REQUIRED is 2+.""" + alarm = WakeAlarm(demo_mode=True) + alarm._view.entry_var.get.return_value = ( + alarm._progress.current_challenge.answer + ) + + alarm._on_submit() + + assert alarm.dismissed is False + assert alarm._progress.rounds_completed == 1 + alarm._stop_beep.set() + + def test_first_round_correct_non_flash_next_no_countdown( + self, + mock_tk_module: MagicMock, + ) -> None: + """When next challenge is math, no flash countdown is started.""" + from wake_alarm._challenges import _Challenge + + alarm = WakeAlarm(demo_mode=True) + alarm._view.entry_var.get.return_value = ( + alarm._progress.current_challenge.answer + ) + next_math = _Challenge(kind="math", display="2 + 2 = ?", answer="4", hint="x") + with patch("wake_alarm._alarm._make_challenge", return_value=next_math): + alarm._on_submit() + + assert alarm._progress.current_challenge.kind == "math" + assert alarm.dismissed is False + alarm._stop_beep.set() + + def test_wrong_code_does_not_dismiss( + self, + mock_tk_module: MagicMock, + ) -> None: + """Entering the wrong answer shows an error without dismissing.""" + from wake_alarm._challenges import _Challenge + + alarm = WakeAlarm(demo_mode=True) + # Use a pinned math challenge so the non-flash wrong-answer branch is covered. + alarm._progress.current_challenge = _Challenge( + kind="math", display="2 + 2 = ?", answer="4", hint="test" + ) + alarm._view.entry_var.get.return_value = "99" + + alarm._on_submit() + + assert alarm.dismissed is False + alarm._stop_beep.set() + + def test_skip_window_expired_keeps_alarm_running( + self, + mock_tk_module: MagicMock, + ) -> None: + """Skip-window expiry denies the skip but does NOT stop the alarm.""" + del mock_tk_module + alarm = WakeAlarm(demo_mode=True) + + with patch( + "wake_alarm._alarm.save_wake_state", + ) as mock_save: + alarm._on_skip_window_expired() + + # Alarm stays active and audible; only the skip reward is gone. + assert alarm._progress.skip_earnable is False + assert alarm._active is True + assert alarm.dismissed is False + assert not alarm._stop_beep.is_set() + mock_save.assert_not_called() + alarm._view._views[0].labels.info.configure.assert_called() + alarm._stop_beep.set() + + def test_skip_window_expired_noop_if_not_active( + self, + mock_tk_module: MagicMock, + ) -> None: + """Expiry is a no-op if alarm is no longer active.""" + del mock_tk_module + alarm = WakeAlarm(demo_mode=True) + alarm._active = False + + alarm._on_skip_window_expired() + + # skip_earnable stays at its initial True (method returned early). + assert alarm._progress.skip_earnable is True + alarm._stop_beep.set() + + def test_dismiss_after_skip_window_earns_no_skip( + self, + mock_tk_module: MagicMock, + ) -> None: + """Typing all rounds after the skip window stops the alarm without a skip.""" + from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED + + alarm = WakeAlarm(demo_mode=True) + alarm._progress.skip_earnable = False + + with patch( + "wake_alarm._alarm.save_wake_state", + ) as mock_save: + for _ in range(DISMISS_ROUNDS_REQUIRED): + alarm._view.entry_var.get.return_value = ( + alarm._progress.current_challenge.answer + ) + alarm._on_submit() + + assert alarm.dismissed is True + assert mock_save.call_args[1]["skip_workout"] is False + alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_fanout.py b/wake_alarm/tests/test_alarm_fanout.py new file mode 100644 index 0000000..df54fb4 --- /dev/null +++ b/wake_alarm/tests/test_alarm_fanout.py @@ -0,0 +1,139 @@ +"""Tests that every action reaches all monitors.""" + +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 +from wake_alarm.tests.conftest import TWO_OUTPUTS + +# --------------------------------------------------------------------------- +# Helpers (duplicated from part 1 so this file is self-contained) +# --------------------------------------------------------------------------- + + +# 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 + # Distinct frames and buttons per call: a shared return_value would make + # "every monitor got one" indistinguishable from "one monitor got two". + mock.Frame.side_effect = lambda *_args, **_kwargs: MagicMock() + mock.Button.side_effect = lambda *_args, **_kwargs: 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 + + +@pytest.mark.usefixtures("dual_output") +class TestFanOutAcrossMonitors: + """The three remaining v0.2.0 loops, on a desk that actually has two. + + Under the single-output default every one of these assertions holds even + if the loop body only ever ran for ``containers[0]`` -- which is exactly + the regression per-output locking exists to prevent. + """ + + def test_dismiss_clears_and_repaints_every_monitor( + self, mock_tk_module: MagicMock + ) -> None: + """The dismissal message replaces the challenge on both screens.""" + alarm = WakeAlarm(demo_mode=True) + containers = alarm._view.containers + assert len(containers) == len(TWO_OUTPUTS) + leftovers = {id(c): [MagicMock(), MagicMock()] for c in containers} + for container in containers: + container.winfo_children.return_value = leftovers[id(container)] + mock_tk_module.Label.reset_mock() + + with patch("wake_alarm._alarm.save_wake_state"): + alarm._dismiss_alarm(earned_skip=True) + + for widgets in leftovers.values(): + for widget in widgets: + widget.destroy.assert_called_once_with() + parents = [call.args[0] for call in mock_tk_module.Label.call_args_list] + assert {id(parent) for parent in parents} == {id(c) for c in containers} + alarm._stop_beep.set() + + def test_flash_recolours_every_monitor(self, mock_tk_module: MagicMock) -> None: + """A flash tick reaches both surfaces, not just the primary.""" + del mock_tk_module + alarm = WakeAlarm(demo_mode=True) + containers = alarm._view.containers + # __init__ already ran one flash tick; only the next one should count. + for container in containers: + container.configure.reset_mock() + container.winfo_toplevel.return_value.configure.reset_mock() + + alarm._flash_step() + + for container in containers: + container.configure.assert_called_once() + container.winfo_toplevel.return_value.configure.assert_called_once() + alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_flash_challenge.py b/wake_alarm/tests/test_alarm_flash_challenge.py index 1a8c299..919bf61 100644 --- a/wake_alarm/tests/test_alarm_flash_challenge.py +++ b/wake_alarm/tests/test_alarm_flash_challenge.py @@ -37,6 +37,7 @@ def _block_real_tk() -> Generator[MagicMock]: 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", @@ -50,16 +51,14 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess calls for extra ALSA devices and hardware.""" with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), + 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._alarm._warn_if_no_real_sink"), + 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._alarm.turn_on_plug"), + patch("wake_alarm._startup.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): yield mock @@ -71,6 +70,7 @@ def mock_tk_module() -> Generator[MagicMock]: 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", @@ -88,7 +88,7 @@ class TestFlashChallenge: mock_tk_module: MagicMock, ) -> None: """_flash_tick counts down per second and hides the code at zero.""" - from wake_alarm._alarm import _Challenge + from wake_alarm._challenges import _Challenge alarm = WakeAlarm(demo_mode=True) alarm._progress.current_challenge = _Challenge( @@ -137,7 +137,7 @@ class TestFlashChallenge: mock_tk_module: MagicMock, ) -> None: """Wrong flash answer restores the code and restarts the countdown.""" - from wake_alarm._alarm import _Challenge + from wake_alarm._challenges import _Challenge alarm = WakeAlarm(demo_mode=True) alarm._progress.current_challenge = _Challenge( @@ -161,7 +161,7 @@ class TestFlashChallenge: mock_tk_module: MagicMock, ) -> None: """When the next-round challenge is flash, the countdown starts immediately.""" - from wake_alarm._alarm import _Challenge + from wake_alarm._challenges import _Challenge alarm = WakeAlarm(demo_mode=True) alarm._progress.current_challenge = _Challenge( diff --git a/wake_alarm/tests/test_alarm_main.py b/wake_alarm/tests/test_alarm_main.py new file mode 100644 index 0000000..90799cb --- /dev/null +++ b/wake_alarm/tests/test_alarm_main.py @@ -0,0 +1,149 @@ +"""Tests for the alarm daemon entry point.""" + +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, + main, +) + +# --------------------------------------------------------------------------- +# Helpers (duplicated from part 1 so this file is self-contained) +# --------------------------------------------------------------------------- + + +# 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 + # Distinct frames and buttons per call: a shared return_value would make + # "every monitor got one" indistinguishable from "one monitor got two". + mock.Frame.side_effect = lambda *_args, **_kwargs: MagicMock() + mock.Button.side_effect = lambda *_args, **_kwargs: 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 + + +class TestMain: + """Tests for the main() entry point.""" + + def test_exits_when_not_alarm_day(self) -> None: + """main() returns early when not an alarm day.""" + with ( + patch( + "wake_alarm._alarm.should_run_alarm", + return_value=False, + ), + patch("wake_alarm._alarm.sys") as mock_sys, + ): + mock_sys.argv = ["alarm"] + main() # Should just return without error + + def test_creates_alarm_when_should_run( + self, + mock_tk_module: MagicMock, + ) -> None: + """main() creates a WakeAlarm when conditions are met.""" + del mock_tk_module + with ( + patch( + "wake_alarm._alarm.should_run_alarm", + return_value=True, + ), + patch( + "wake_alarm._alarm.sys", + ) as mock_sys, + patch.object(WakeAlarm, "run") as mock_run, + patch.object(WakeAlarm, "__init__", return_value=None), + ): + mock_sys.argv = ["alarm"] + main() + mock_run.assert_called_once() + + def test_trigger_now_bypasses_gate( + self, + mock_tk_module: MagicMock, + ) -> None: + """--trigger-now bypasses _should_run_alarm.""" + del mock_tk_module + with ( + patch( + "wake_alarm._alarm.should_run_alarm", + return_value=False, + ) as mock_gate, + patch("wake_alarm._alarm.sys") as mock_sys, + patch.object(WakeAlarm, "run") as mock_run, + patch.object(WakeAlarm, "__init__", return_value=None), + ): + mock_sys.argv = ["alarm", "--trigger-now"] + main() + mock_gate.assert_not_called() + mock_run.assert_called_once() diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index 27d5046..6757c99 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -13,7 +13,6 @@ if TYPE_CHECKING: from wake_alarm._alarm import ( WakeAlarm, - main, ) # --------------------------------------------------------------------------- @@ -42,6 +41,7 @@ def _block_real_tk() -> Generator[MagicMock]: 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", @@ -55,16 +55,14 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), + 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._alarm._warn_if_no_real_sink"), + 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._alarm.turn_on_plug"), + patch("wake_alarm._startup.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): yield mock @@ -76,6 +74,7 @@ def mock_tk_module() -> Generator[MagicMock]: 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", @@ -126,204 +125,6 @@ class TestWakeAlarmInit: alarm._stop_beep.set() -class TestWakeAlarmDismiss: - """Tests for alarm dismiss logic.""" - - def test_correct_code_dismisses_after_all_rounds( - self, - mock_tk_module: MagicMock, - ) -> None: - """Entering the correct answer for every required round dismisses the alarm.""" - from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED - - alarm = WakeAlarm(demo_mode=True) - - with patch( - "wake_alarm._alarm.save_wake_state", - ) as mock_save: - for _ in range(DISMISS_ROUNDS_REQUIRED): - alarm._view.entry_var.get.return_value = ( - alarm._progress.current_challenge.answer - ) - alarm._on_submit() - - assert alarm.dismissed is True - mock_save.assert_called_once() - assert mock_save.call_args[1]["skip_workout"] is True - alarm._stop_beep.set() - - def test_first_round_correct_does_not_dismiss( - self, - mock_tk_module: MagicMock, - ) -> None: - """A single correct entry is not enough — DISMISS_ROUNDS_REQUIRED is 2+.""" - alarm = WakeAlarm(demo_mode=True) - alarm._view.entry_var.get.return_value = ( - alarm._progress.current_challenge.answer - ) - - alarm._on_submit() - - assert alarm.dismissed is False - assert alarm._progress.rounds_completed == 1 - alarm._stop_beep.set() - - def test_first_round_correct_non_flash_next_no_countdown( - self, - mock_tk_module: MagicMock, - ) -> None: - """When next challenge is math, no flash countdown is started.""" - from wake_alarm._challenges import _Challenge - - alarm = WakeAlarm(demo_mode=True) - alarm._view.entry_var.get.return_value = ( - alarm._progress.current_challenge.answer - ) - next_math = _Challenge(kind="math", display="2 + 2 = ?", answer="4", hint="x") - with patch("wake_alarm._alarm._make_challenge", return_value=next_math): - alarm._on_submit() - - assert alarm._progress.current_challenge.kind == "math" - assert alarm.dismissed is False - alarm._stop_beep.set() - - def test_wrong_code_does_not_dismiss( - self, - mock_tk_module: MagicMock, - ) -> None: - """Entering the wrong answer shows an error without dismissing.""" - from wake_alarm._alarm import _Challenge - - alarm = WakeAlarm(demo_mode=True) - # Use a pinned math challenge so the non-flash wrong-answer branch is covered. - alarm._progress.current_challenge = _Challenge( - kind="math", display="2 + 2 = ?", answer="4", hint="test" - ) - alarm._view.entry_var.get.return_value = "99" - - alarm._on_submit() - - assert alarm.dismissed is False - alarm._stop_beep.set() - - def test_skip_window_expired_keeps_alarm_running( - self, - mock_tk_module: MagicMock, - ) -> None: - """Skip-window expiry denies the skip but does NOT stop the alarm.""" - del mock_tk_module - alarm = WakeAlarm(demo_mode=True) - - with patch( - "wake_alarm._alarm.save_wake_state", - ) as mock_save: - alarm._on_skip_window_expired() - - # Alarm stays active and audible; only the skip reward is gone. - assert alarm._progress.skip_earnable is False - assert alarm._active is True - assert alarm.dismissed is False - assert not alarm._stop_beep.is_set() - mock_save.assert_not_called() - alarm._view._views[0].labels.info.configure.assert_called() - alarm._stop_beep.set() - - def test_skip_window_expired_noop_if_not_active( - self, - mock_tk_module: MagicMock, - ) -> None: - """Expiry is a no-op if alarm is no longer active.""" - del mock_tk_module - alarm = WakeAlarm(demo_mode=True) - alarm._active = False - - alarm._on_skip_window_expired() - - # skip_earnable stays at its initial True (method returned early). - assert alarm._progress.skip_earnable is True - alarm._stop_beep.set() - - def test_dismiss_after_skip_window_earns_no_skip( - self, - mock_tk_module: MagicMock, - ) -> None: - """Typing all rounds after the skip window stops the alarm without a skip.""" - from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED - - alarm = WakeAlarm(demo_mode=True) - alarm._progress.skip_earnable = False - - with patch( - "wake_alarm._alarm.save_wake_state", - ) as mock_save: - for _ in range(DISMISS_ROUNDS_REQUIRED): - alarm._view.entry_var.get.return_value = ( - alarm._progress.current_challenge.answer - ) - alarm._on_submit() - - assert alarm.dismissed is True - assert mock_save.call_args[1]["skip_workout"] is False - alarm._stop_beep.set() - - -class TestMain: - """Tests for the main() entry point.""" - - def test_exits_when_not_alarm_day(self) -> None: - """main() returns early when not an alarm day.""" - with ( - patch( - "wake_alarm._alarm._should_run_alarm", - return_value=False, - ), - patch("wake_alarm._alarm.sys") as mock_sys, - ): - mock_sys.argv = ["alarm"] - main() # Should just return without error - - def test_creates_alarm_when_should_run( - self, - mock_tk_module: MagicMock, - ) -> None: - """main() creates a WakeAlarm when conditions are met.""" - del mock_tk_module - with ( - patch( - "wake_alarm._alarm._should_run_alarm", - return_value=True, - ), - patch( - "wake_alarm._alarm.sys", - ) as mock_sys, - patch.object(WakeAlarm, "run") as mock_run, - patch.object(WakeAlarm, "__init__", return_value=None), - ): - mock_sys.argv = ["alarm"] - main() - mock_run.assert_called_once() - - def test_trigger_now_bypasses_gate( - self, - mock_tk_module: MagicMock, - ) -> None: - """--trigger-now bypasses _should_run_alarm.""" - del mock_tk_module - with ( - patch( - "wake_alarm._alarm._should_run_alarm", - return_value=False, - ) as mock_gate, - patch("wake_alarm._alarm.sys") as mock_sys, - patch.object(WakeAlarm, "run") as mock_run, - patch.object(WakeAlarm, "__init__", return_value=None), - ): - mock_sys.argv = ["alarm", "--trigger-now"] - main() - mock_gate.assert_not_called() - mock_run.assert_called_once() - - class TestCodeRefreshAndTimer: """Tests for code refresh and timer update methods.""" @@ -363,24 +164,6 @@ class TestCodeRefreshAndTimer: alarm._stop_beep.set() -class TestBeepLoop: - """Tests for the beep loop thread.""" - - def test_beep_loop_stops_on_event( - self, - mock_tk_module: MagicMock, - ) -> None: - """Beep loop exits when stop event is set.""" - alarm = WakeAlarm(demo_mode=True) - alarm._stop_beep.set() - # Loop should exit immediately - with patch( - "wake_alarm._alarm._beep_soft", - ): - alarm._beep_loop() - alarm._stop_beep.set() - - class TestScreenFlash: """Tests for _start_screen_flash and _flash_step.""" diff --git a/wake_alarm/tests/test_alarm_part3.py b/wake_alarm/tests/test_alarm_part3.py index ba7423d..a4fb9d7 100644 --- a/wake_alarm/tests/test_alarm_part3.py +++ b/wake_alarm/tests/test_alarm_part3.py @@ -37,6 +37,7 @@ def _block_real_tk() -> Generator[MagicMock]: 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", @@ -50,16 +51,14 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess calls for extra ALSA devices and hardware.""" with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), + 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._alarm._warn_if_no_real_sink"), + 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._alarm.turn_on_plug"), + patch("wake_alarm._startup.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): yield mock @@ -71,6 +70,7 @@ def mock_tk_module() -> Generator[MagicMock]: 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", diff --git a/wake_alarm/tests/test_alarm_part4.py b/wake_alarm/tests/test_alarm_part4.py index b5036c8..ca6238d 100644 --- a/wake_alarm/tests/test_alarm_part4.py +++ b/wake_alarm/tests/test_alarm_part4.py @@ -14,7 +14,6 @@ if TYPE_CHECKING: from wake_alarm._alarm import WakeAlarm from wake_alarm._alarm_view import _EntryProxy from wake_alarm._sinks import _AudioState -from wake_alarm.tests.conftest import TWO_OUTPUTS # --------------------------------------------------------------------------- # Helpers (duplicated from part 1 so this file is self-contained) @@ -45,6 +44,7 @@ def _block_real_tk() -> Generator[MagicMock]: 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", @@ -58,16 +58,14 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), + 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._alarm._warn_if_no_real_sink"), + 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._alarm.turn_on_plug"), + patch("wake_alarm._startup.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): yield mock @@ -79,6 +77,7 @@ def mock_tk_module() -> Generator[MagicMock]: 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", @@ -133,19 +132,6 @@ class TestClose: assert alarm._stop_beep.is_set() alarm.root.destroy.assert_called() - def test_on_close_restores_fans( - self, - mock_tk_module: MagicMock, - ) -> None: - """on_close calls _restore_fans with the saved fan state.""" - del mock_tk_module - alarm = WakeAlarm(demo_mode=True) - alarm._hardware.fan_state = True - with patch("wake_alarm._alarm._restore_fans") as mock_restore: - alarm.on_close() - mock_restore.assert_called_once_with(active=True) - alarm._stop_beep.set() - def test_on_close_restores_audio( self, mock_tk_module: MagicMock, @@ -232,52 +218,3 @@ class TestMultiMonitorSurfaces: _EntryProxy([dead], MagicMock()).focus_force() # must not raise alarm._stop_beep.set() - - -@pytest.mark.usefixtures("dual_output") -class TestFanOutAcrossMonitors: - """The three remaining v0.2.0 loops, on a desk that actually has two. - - Under the single-output default every one of these assertions holds even - if the loop body only ever ran for ``containers[0]`` -- which is exactly - the regression per-output locking exists to prevent. - """ - - def test_dismiss_clears_and_repaints_every_monitor( - self, mock_tk_module: MagicMock - ) -> None: - """The dismissal message replaces the challenge on both screens.""" - alarm = WakeAlarm(demo_mode=True) - containers = alarm._view.containers - assert len(containers) == len(TWO_OUTPUTS) - leftovers = {id(c): [MagicMock(), MagicMock()] for c in containers} - for container in containers: - container.winfo_children.return_value = leftovers[id(container)] - mock_tk_module.Label.reset_mock() - - with patch("wake_alarm._alarm.save_wake_state"): - alarm._dismiss_alarm(earned_skip=True) - - for widgets in leftovers.values(): - for widget in widgets: - widget.destroy.assert_called_once_with() - parents = [call.args[0] for call in mock_tk_module.Label.call_args_list] - assert {id(parent) for parent in parents} == {id(c) for c in containers} - alarm._stop_beep.set() - - def test_flash_recolours_every_monitor(self, mock_tk_module: MagicMock) -> None: - """A flash tick reaches both surfaces, not just the primary.""" - del mock_tk_module - alarm = WakeAlarm(demo_mode=True) - containers = alarm._view.containers - # __init__ already ran one flash tick; only the next one should count. - for container in containers: - container.configure.reset_mock() - container.winfo_toplevel.return_value.configure.reset_mock() - - alarm._flash_step() - - for container in containers: - container.configure.assert_called_once() - container.winfo_toplevel.return_value.configure.assert_called_once() - alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_sinks.py b/wake_alarm/tests/test_alarm_sinks.py index 14aa2ed..3a3632b 100644 --- a/wake_alarm/tests/test_alarm_sinks.py +++ b/wake_alarm/tests/test_alarm_sinks.py @@ -5,13 +5,9 @@ from __future__ import annotations import subprocess from unittest.mock import MagicMock, patch -from wake_alarm._alarm import _parse_args -from wake_alarm._constants import ALARM_AUDIO_SINK +from wake_alarm._gating import parse_args as _parse_args from wake_alarm._sinks import ( - _activate_all_outputs, - _AudioState, _capture_audio_state, - _CardInfo, _current_default_sink, _list_cards, _list_real_sinks, @@ -19,8 +15,6 @@ from wake_alarm._sinks import ( _parse_cards, _parse_profile_line, _parse_sinks, - _restore_audio_state, - _warn_if_no_real_sink, ) # A representative `pactl list cards` block covering every parse branch: @@ -231,214 +225,6 @@ class TestCaptureAudioState: assert state.sink_mutes["alsa_output.analog"] is False -class TestActivateAllOutputs: - """Tests for _activate_all_outputs.""" - - def test_returns_none_when_pactl_missing(self) -> None: - """No pactl on PATH → returns None without touching audio.""" - with ( - patch("wake_alarm._sinks._pactl_path", return_value=None), - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - assert _activate_all_outputs() is None - mock_run.assert_not_called() - - def _sentinel_state(self) -> _AudioState: - return _AudioState( - default_sink="old", - card_profiles={}, - sink_volumes={}, - sink_mutes={}, - ) - - def test_activates_and_prefers_alarm_sink(self) -> None: - """Forces card profiles on, maxes sinks, defaults to ALARM_AUDIO_SINK.""" - state = self._sentinel_state() - cards = { - "card_hdmi": _CardInfo(None, ["output:hdmi-stereo"]), - "card_mic": _CardInfo(None, []), - } - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks._capture_audio_state", return_value=state), - patch("wake_alarm._sinks._pactl_text", return_value=""), - patch("wake_alarm._sinks._parse_cards", return_value=cards), - patch( - "wake_alarm._sinks._list_real_sinks", - return_value=[ALARM_AUDIO_SINK, "other"], - ), - patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - assert _activate_all_outputs() is state - cmds = [call.args[0] for call in mock_run.call_args_list] - assert [ - "/usr/bin/pactl", - "set-card-profile", - "card_hdmi", - "output:hdmi-stereo", - ] in cmds - # The input-only card (no output profile) is never switched. - assert not any(c[2] == "card_mic" for c in cmds if c[1] == "set-card-profile") - assert ["/usr/bin/pactl", "set-default-sink", ALARM_AUDIO_SINK] in cmds - assert mock_audible.call_count == 2 - - def test_does_not_block_on_sink_enumeration(self) -> None: - """Activation checks present sinks once — no polling/sleeping loop. - - The per-cycle presence check in _play_on_all_sinks is what covers a - slow monitor sink, so activation must not spin waiting for it. - """ - state = self._sentinel_state() - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks._capture_audio_state", return_value=state), - patch("wake_alarm._sinks._pactl_text", return_value=""), - patch("wake_alarm._sinks._parse_cards", return_value={}), - patch( - "wake_alarm._sinks._list_real_sinks", - return_value=[ALARM_AUDIO_SINK], - ) as mock_list, - patch("wake_alarm._sinks._ensure_sink_audible"), - patch("wake_alarm._sinks.subprocess.run"), - ): - assert _activate_all_outputs() is state - # Enumerated exactly once — no retry/poll loop. - mock_list.assert_called_once() - - def test_falls_back_to_first_sink_when_alarm_sink_absent(self) -> None: - """ALARM_AUDIO_SINK not present → default is the first present sink.""" - state = self._sentinel_state() - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks._capture_audio_state", return_value=state), - patch("wake_alarm._sinks._pactl_text", return_value=""), - patch("wake_alarm._sinks._parse_cards", return_value={}), - patch("wake_alarm._sinks._list_real_sinks", return_value=["other"]), - patch("wake_alarm._sinks._ensure_sink_audible"), - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - assert _activate_all_outputs() is state - cmds = [call.args[0] for call in mock_run.call_args_list] - assert ["/usr/bin/pactl", "set-default-sink", "other"] in cmds - - def test_returns_state_when_no_sink_present(self) -> None: - """No real sink present yet → returns state, no default, no maxing.""" - state = self._sentinel_state() - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks._capture_audio_state", return_value=state), - patch("wake_alarm._sinks._pactl_text", return_value=""), - patch("wake_alarm._sinks._parse_cards", return_value={}), - patch("wake_alarm._sinks._list_real_sinks", return_value=[]), - patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - assert _activate_all_outputs() is state - mock_audible.assert_not_called() - cmds = [call.args[0] for call in mock_run.call_args_list] - assert not any(c[1] == "set-default-sink" for c in cmds) - - -class TestRestoreAudioState: - """Tests for _restore_audio_state.""" - - def test_none_is_noop(self) -> None: - """None state → does nothing, no pactl lookup.""" - with patch("wake_alarm._sinks._pactl_path") as mock_path: - _restore_audio_state(None) - mock_path.assert_not_called() - - def test_no_pactl_returns_silently(self) -> None: - """State present but pactl missing → no raise, no run.""" - state = _AudioState( - default_sink="s", - card_profiles={}, - sink_volumes={}, - sink_mutes={}, - ) - with ( - patch("wake_alarm._sinks._pactl_path", return_value=None), - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - _restore_audio_state(state) - mock_run.assert_not_called() - - def test_restores_everything(self) -> None: - """Restores card profiles, sink volumes, mutes, and the default sink.""" - state = _AudioState( - default_sink="analog", - card_profiles={"card_hdmi": "off"}, - sink_volumes={"analog": "65536"}, - sink_mutes={"analog": True}, - ) - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - _restore_audio_state(state) - cmds = [call.args[0] for call in mock_run.call_args_list] - assert ["/usr/bin/pactl", "set-card-profile", "card_hdmi", "off"] in cmds - assert ["/usr/bin/pactl", "set-sink-volume", "analog", "65536"] in cmds - assert ["/usr/bin/pactl", "set-sink-mute", "analog", "1"] in cmds - assert ["/usr/bin/pactl", "set-default-sink", "analog"] in cmds - - def test_skips_default_when_none(self) -> None: - """default_sink None → no set-default-sink command.""" - state = _AudioState( - default_sink=None, - card_profiles={}, - sink_volumes={"analog": "100"}, - sink_mutes={"analog": False}, - ) - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks.subprocess.run") as mock_run, - ): - _restore_audio_state(state) - cmds = [call.args[0] for call in mock_run.call_args_list] - assert not any(c[1] == "set-default-sink" for c in cmds) - # Unmuted sink restores with "0". - assert ["/usr/bin/pactl", "set-sink-mute", "analog", "0"] in cmds - - -class TestWarnIfNoRealSink: - """Tests for _warn_if_no_real_sink.""" - - def test_logs_when_pactl_missing(self) -> None: - """No pactl on PATH → warns and returns (no sink lookup).""" - with ( - patch("wake_alarm._sinks._pactl_path", return_value=None), - patch("wake_alarm._sinks._list_real_sinks") as mock_list, - ): - _warn_if_no_real_sink() - mock_list.assert_not_called() - - def test_warns_when_no_real_sink(self) -> None: - """No real sink → warning is emitted.""" - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch("wake_alarm._sinks._list_real_sinks", return_value=[]), - patch("wake_alarm._sinks._logger") as mock_log, - ): - _warn_if_no_real_sink() - mock_log.warning.assert_called() - - def test_info_when_real_sink_present(self) -> None: - """A real sink → info log, no warning.""" - with ( - patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), - patch( - "wake_alarm._sinks._list_real_sinks", - return_value=["alsa_output.analog"], - ), - patch("wake_alarm._sinks._logger") as mock_log, - ): - _warn_if_no_real_sink() - mock_log.info.assert_called() - mock_log.warning.assert_not_called() - - class TestParseArgs: """Tests for _parse_args.""" diff --git a/wake_alarm/tests/test_beep_loop.py b/wake_alarm/tests/test_beep_loop.py new file mode 100644 index 0000000..ae717d0 --- /dev/null +++ b/wake_alarm/tests/test_beep_loop.py @@ -0,0 +1,151 @@ +"""Tests for the background beep loop.""" + +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING +from unittest.mock import patch + +from wake_alarm._beep_loop import run_beep_loop, start_beep_thread +from wake_alarm._constants import ( + FAILSAFE_SILENCE_SECONDS, + PHASE_MEDIUM_END, + PHASE_SOFT_END, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + import pytest + + +class _Clock: + """Monotonic stand-in that advances a fixed step on every read.""" + + def __init__(self, start: float, step: float) -> None: + self._now = start + self._step = step + + def __call__(self) -> float: + value = self._now + self._now += self._step + return value + + +def _run_once(elapsed: float) -> dict[str, int]: + """Run exactly one loop iteration at *elapsed* seconds and count the calls.""" + stop = threading.Event() + counts = {"soft": 0, "medium": 0, "loud": 0, "sinks": 0} + + def _count(key: str) -> Callable[..., None]: + def _inner(*_a: object, **_k: object) -> None: + counts[key] += 1 + stop.set() + + return _inner + + with ( + patch("wake_alarm._beep_loop._beep_soft", side_effect=_count("soft")), + patch("wake_alarm._beep_loop._beep_medium", side_effect=_count("medium")), + patch("wake_alarm._beep_loop._beep_loud", side_effect=_count("loud")), + patch( + "wake_alarm._beep_loop._play_on_all_sinks", + side_effect=lambda *_a: counts.__setitem__("sinks", counts["sinks"] + 1), + ), + ): + run_beep_loop(stop, 0.0, monotonic=lambda: elapsed) + return counts + + +def test_exits_immediately_when_already_stopped() -> None: + """A stop set before entry means no sound at all.""" + stop = threading.Event() + stop.set() + with patch("wake_alarm._beep_loop._play_on_all_sinks") as mock_play: + run_beep_loop(stop, 0.0, monotonic=lambda: 0.0) + mock_play.assert_not_called() + + +def test_soft_phase_uses_the_soft_beep() -> None: + """Early in the ring the gentle tone sounds.""" + assert _run_once(0.0)["soft"] == 1 + + +def test_medium_phase_uses_the_medium_beep() -> None: + """The middle band uses the medium tone.""" + midpoint = (PHASE_SOFT_END + PHASE_MEDIUM_END) / 2 * 60 + assert _run_once(midpoint)["medium"] == 1 + + +def test_loud_phase_uses_the_loud_beep() -> None: + """Past the medium boundary the alarm is at its most insistent.""" + assert _run_once(PHASE_MEDIUM_END * 60 + 1)["loud"] == 1 + + +def test_plays_on_every_sink_while_audible() -> None: + """Audio is driven to all outputs, not one chosen sink.""" + assert _run_once(0.0)["sinks"] == 1 + + +def test_failsafe_plays_no_audio() -> None: + """Past the failsafe nothing sounds -- on any output.""" + stop = threading.Event() + calls = {"n": 0} + + def _stop_after(*_a: object, **_k: object) -> None: + calls["n"] += 1 + stop.set() + + ticks = {"n": 0} + + def _wait(_timeout: float) -> bool: + # Nothing sounds during the failsafe, so no audio callback can end the + # loop; stop it from the wait instead. + ticks["n"] += 1 + if ticks["n"] >= 3: + stop.set() + return stop.is_set() + + with ( + patch("wake_alarm._beep_loop._play_on_all_sinks", side_effect=_stop_after), + patch("wake_alarm._beep_loop._beep_soft", side_effect=_stop_after), + patch("wake_alarm._beep_loop._beep_medium", side_effect=_stop_after), + patch("wake_alarm._beep_loop._beep_loud", side_effect=_stop_after), + patch.object(stop, "wait", _wait), + ): + run_beep_loop( + stop, + 0.0, + monotonic=_Clock(FAILSAFE_SILENCE_SECONDS, 1.0), + ) + assert calls["n"] == 0 + assert ticks["n"] == 3 + + +def test_failsafe_is_announced_once(caplog: pytest.LogCaptureFixture) -> None: + """Reaching the failsafe is logged loudly, and only once.""" + stop = threading.Event() + clock = _Clock(FAILSAFE_SILENCE_SECONDS, 1.0) + ticks = {"n": 0} + + def _wait(_timeout: float) -> bool: + ticks["n"] += 1 + if ticks["n"] >= 3: + stop.set() + return stop.is_set() + + with ( + caplog.at_level("WARNING"), + patch.object(stop, "wait", _wait), + ): + run_beep_loop(stop, 0.0, monotonic=clock) + assert caplog.text.count("Failsafe reached") == 1 + + +def test_start_beep_thread_is_a_daemon() -> None: + """A stuck beep thread must never keep the process alive.""" + stop = threading.Event() + stop.set() + thread = start_beep_thread(stop, 0.0) + thread.join(timeout=5) + assert thread.daemon diff --git a/wake_alarm/tests/test_brightness.py b/wake_alarm/tests/test_brightness.py new file mode 100644 index 0000000..388f32d --- /dev/null +++ b/wake_alarm/tests/test_brightness.py @@ -0,0 +1,118 @@ +"""Tests for raising every monitor to full brightness.""" + +from __future__ import annotations + +import subprocess +from unittest.mock import MagicMock, patch + +from wake_alarm._brightness import _set_max_brightness + + +class TestSetMaxBrightness: + """Tests for _set_max_brightness.""" + + def test_noop_when_xrandr_missing(self) -> None: + """No xrandr on PATH → subprocess.run never called.""" + with ( + patch("wake_alarm._audio.shutil.which", return_value=None), + patch("wake_alarm._audio.subprocess.run") as mock_run, + ): + _set_max_brightness() + mock_run.assert_not_called() + + def test_noop_on_oserror_from_query(self) -> None: + """OSError from xrandr --query is suppressed.""" + with ( + patch( + "wake_alarm._audio.shutil.which", + return_value="/usr/bin/xrandr", + ), + patch( + "wake_alarm._audio.subprocess.run", + side_effect=OSError("no display"), + ), + ): + _set_max_brightness() # must not raise + + def test_noop_on_timeout_from_query(self) -> None: + """TimeoutExpired from xrandr --query is suppressed.""" + with ( + patch( + "wake_alarm._audio.shutil.which", + return_value="/usr/bin/xrandr", + ), + patch( + "wake_alarm._audio.subprocess.run", + side_effect=subprocess.TimeoutExpired("xrandr", 5), + ), + ): + _set_max_brightness() # must not raise + + def test_sets_brightness_for_connected_displays(self) -> None: + """Connected displays each get an --output --brightness call.""" + mock_query_result = MagicMock() + mock_query_result.stdout = ( + "HDMI-0 connected 2560x1440+0+0\nDP-0 connected primary\n" + ) + call_args_list: list[list[str]] = [] + + def fake_run(args: list[str], **_kwargs: object) -> MagicMock: + call_args_list.append(args) + return mock_query_result + + with ( + patch( + "wake_alarm._audio.shutil.which", + return_value="/usr/bin/xrandr", + ), + patch("wake_alarm._audio.subprocess.run", side_effect=fake_run), + ): + _set_max_brightness() + + # First call is --query; subsequent calls set brightness for each output. + brightness_calls = [a for a in call_args_list if "--brightness" in a] + expected_brightness_calls = 2 + assert len(brightness_calls) == expected_brightness_calls + + def test_skips_disconnected_outputs(self) -> None: + """Disconnected outputs do NOT get a brightness call.""" + mock_result = MagicMock() + mock_result.stdout = "Screen 0: minimum 320\nHDMI-0 disconnected\n" + with ( + patch( + "wake_alarm._audio.shutil.which", + return_value="/usr/bin/xrandr", + ), + patch( + "wake_alarm._audio.subprocess.run", + return_value=mock_result, + ) as mock_run, + ): + _set_max_brightness() + # Only the --query call, no brightness calls. + assert mock_run.call_count == 1 + + def test_warns_when_brightness_call_fails(self) -> None: + """OSError on per-output --brightness call is logged but swallowed.""" + query_result = MagicMock() + query_result.stdout = ( + "Screen 0: minimum 320\nHDMI-0 connected primary 1920x1080\n" + ) + + def _run_side_effect(args: list[str], **_kwargs: object) -> MagicMock: + if "--query" in args: + return query_result + msg = "permission denied" + raise OSError(msg) + + with ( + patch( + "wake_alarm._audio.shutil.which", + return_value="/usr/bin/xrandr", + ), + patch( + "wake_alarm._audio.subprocess.run", + side_effect=_run_side_effect, + ), + ): + _set_max_brightness() # must not raise diff --git a/wake_alarm/tests/test_escape_flow.py b/wake_alarm/tests/test_escape_flow.py index bfd72a6..6f5141b 100644 --- a/wake_alarm/tests/test_escape_flow.py +++ b/wake_alarm/tests/test_escape_flow.py @@ -14,7 +14,6 @@ from unittest.mock import MagicMock, patch from gatelock import EscapePolicy, EscapeTracker import pytest -from wake_alarm._alarm import WakeAlarm from wake_alarm._constants import ESCAPE_OFFER_AFTER_SECONDS from wake_alarm._escape_flow import ( ALARM_ESCAPE_POLICY, @@ -23,7 +22,6 @@ from wake_alarm._escape_flow import ( build_tracker, is_offerable, ) -from wake_alarm.tests.conftest import TWO_OUTPUTS if TYPE_CHECKING: from collections.abc import Generator @@ -79,6 +77,7 @@ def _block_real_tk() -> Generator[MagicMock]: 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), ): @@ -89,16 +88,14 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( - patch("wake_alarm._alarm._play_on_all_sinks") as mock, - patch("wake_alarm._alarm._max_fans", return_value=False), - patch("wake_alarm._alarm._restore_fans"), - patch("wake_alarm._alarm._set_max_brightness"), - patch("wake_alarm._alarm._wake_display"), + 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._alarm._warn_if_no_real_sink"), + 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._alarm.turn_on_plug"), + patch("wake_alarm._startup.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): yield mock @@ -231,137 +228,3 @@ class TestEscapeDialog: granted.assert_not_called() dialog.frame.after.assert_called_once() - - -@pytest.mark.usefixtures("dual_output") -class TestAlarmIntegration: - """The hatch as the alarm actually uses it, on a two-monitor desk.""" - - def test_every_output_gets_its_own_button(self) -> None: - """The premise the rest of this class rests on: two distinct buttons.""" - alarm = WakeAlarm(demo_mode=True) - - buttons = alarm._view.escape_buttons - assert len(buttons) == len(TWO_OUTPUTS) - assert len({id(button) for button in buttons}) == len(TWO_OUTPUTS) - alarm._stop_beep.set() - - def test_button_hidden_until_earned(self) -> None: - """Straight after the alarm starts, the hatch is not offered.""" - alarm = WakeAlarm(demo_mode=True) - for button in alarm._view.escape_buttons: - button.pack.reset_mock() - - alarm._update_timer() - - for button in alarm._view.escape_buttons: - button.pack.assert_not_called() - alarm._stop_beep.set() - - def test_button_revealed_once_earned(self) -> None: - """After the threshold the button appears on every monitor. - - Driven through ``_update_timer`` rather than the hatch directly, so - the per-second tick is on the hook for ever offering the hatch at all. - """ - alarm = WakeAlarm(demo_mode=True) - alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10 - for button in alarm._view.escape_buttons: - button.pack.reset_mock() - - alarm._update_timer() - - for button in alarm._view.escape_buttons: - button.pack.assert_called() - alarm._stop_beep.set() - - def test_not_reoffered_while_open(self) -> None: - """With the form already up, the button is not re-packed.""" - alarm = WakeAlarm(demo_mode=True) - alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10 - alarm._escape.is_open = True - for button in alarm._view.escape_buttons: - button.pack.reset_mock() - - alarm._update_timer() - - for button in alarm._view.escape_buttons: - button.pack.assert_not_called() - alarm._stop_beep.set() - - def test_open_swaps_every_surface(self) -> None: - """Opening the form hides the challenge and builds a form per monitor.""" - alarm = WakeAlarm(demo_mode=True) - surfaces = alarm._view.surfaces - assert len(surfaces) == len(TWO_OUTPUTS) - - with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls: - alarm._escape.open() - - assert alarm._escape.is_open is True - assert dialog_cls.call_count == len(surfaces) - # One dialog *per* container, not two against the same parent. - parents = [call.args[0] for call in dialog_cls.call_args_list] - assert {id(parent) for parent in parents} == { - id(view.container) for view in surfaces - } - for view in surfaces: - view.challenge_frame.pack_forget.assert_called_once_with() - alarm._stop_beep.set() - - def test_a_dead_surface_does_not_block_the_others(self) -> None: - """A monitor that has gone away raises TclError; the rest still swap.""" - alarm = WakeAlarm(demo_mode=True) - dead, live = alarm._view.surfaces - dead.challenge_frame.pack_forget.side_effect = tk.TclError("bad path name") - - with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls: - alarm._escape.open() - - assert dialog_cls.call_count == 1 - assert dialog_cls.call_args.args[0] is live.container - alarm._stop_beep.set() - - def test_cancel_puts_the_challenge_back(self) -> None: - """Backing out destroys every form and re-packs every challenge. - - The bool alone is not the property that matters: ``open`` took the - challenge off the screen, so a cancel that only flips a flag leaves - the alarm showing a dead form on both monitors. - """ - alarm = WakeAlarm(demo_mode=True) - surfaces = alarm._view.surfaces - dialogs: list[MagicMock] = [] - - def make_dialog(*_args: object, **_kwargs: object) -> MagicMock: - dialog = MagicMock() - dialogs.append(dialog) - return dialog - - with patch("wake_alarm._escape_flow.EscapeDialog", side_effect=make_dialog): - alarm._escape.open() - # The frame was packed once when the view was built; only the re-pack - # that close() owes us should count. - for view in surfaces: - view.challenge_frame.pack.reset_mock() - - alarm._escape.close() - - assert alarm._escape.is_open is False - assert alarm.dismissed is False - assert len(dialogs) == len(TWO_OUTPUTS) - for dialog in dialogs: - dialog.frame.destroy.assert_called_once_with() - for view in surfaces: - view.challenge_frame.pack.assert_called_once_with() - alarm._stop_beep.set() - - def test_granted_dismisses_without_a_skip(self) -> None: - """Using the hatch silences the alarm but earns no workout skip.""" - alarm = WakeAlarm(demo_mode=True) - - with patch.object(alarm, "_dismiss_alarm") as dismiss: - alarm._on_escape_granted() - - dismiss.assert_called_once_with(earned_skip=False) - alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_escape_integration.py b/wake_alarm/tests/test_escape_integration.py new file mode 100644 index 0000000..b1e31cf --- /dev/null +++ b/wake_alarm/tests/test_escape_integration.py @@ -0,0 +1,229 @@ +"""Tests wiring the escape hatch into the running alarm. + +Split from ``test_escape_flow`` for the 250-line cap: that module covers the +hatch in isolation, this one covers it mounted on a real two-monitor alarm. +""" + +from __future__ import annotations + +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from gatelock import EscapePolicy, EscapeTracker +import pytest + +from wake_alarm._alarm import WakeAlarm +from wake_alarm._constants import ESCAPE_OFFER_AFTER_SECONDS +from wake_alarm.tests.conftest import TWO_OUTPUTS + +if TYPE_CHECKING: + from collections.abc import Generator + from pathlib import Path + +TODAY = "2026-07-25" + + +@pytest.fixture +def tracker(tmp_path: Path) -> EscapeTracker: + """A tracker over a throwaway history and key.""" + key = tmp_path / "hmac.key" + key.write_bytes(b"0" * 32) + return EscapeTracker( + EscapePolicy(name="cant-wake", label="Can't-wake"), + tmp_path / "escape.json", + key_file=key, + ) + + +# --------------------------------------------------------------------------- +# Tk stand-ins (duplicated from the part files so this one is self-contained; +# they are autouse there, and hoisting them into conftest would silently +# neuter the audio and sync suites) +# --------------------------------------------------------------------------- + + +def _make_mock_tk() -> MagicMock: + """Build a MagicMock that stands in for the tkinter module. + + ``Button`` gets a ``side_effect`` rather than a single ``return_value``: + every surface must end up with its *own* button, otherwise + ``escape_buttons`` is N references to one mock and "packed on every + monitor" cannot be told apart from "packed once". + """ + 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.side_effect = lambda *_args, **_kwargs: MagicMock() + mock.Label.return_value = MagicMock() + mock.Entry.return_value = MagicMock() + mock.Button.side_effect = lambda *_args, **_kwargs: 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.mark.usefixtures("dual_output") +class TestAlarmIntegration: + """The hatch as the alarm actually uses it, on a two-monitor desk.""" + + def test_every_output_gets_its_own_button(self) -> None: + """The premise the rest of this class rests on: two distinct buttons.""" + alarm = WakeAlarm(demo_mode=True) + + buttons = alarm._view.escape_buttons + assert len(buttons) == len(TWO_OUTPUTS) + assert len({id(button) for button in buttons}) == len(TWO_OUTPUTS) + alarm._stop_beep.set() + + def test_button_hidden_until_earned(self) -> None: + """Straight after the alarm starts, the hatch is not offered.""" + alarm = WakeAlarm(demo_mode=True) + for button in alarm._view.escape_buttons: + button.pack.reset_mock() + + alarm._update_timer() + + for button in alarm._view.escape_buttons: + button.pack.assert_not_called() + alarm._stop_beep.set() + + def test_button_revealed_once_earned(self) -> None: + """After the threshold the button appears on every monitor. + + Driven through ``_update_timer`` rather than the hatch directly, so + the per-second tick is on the hook for ever offering the hatch at all. + """ + alarm = WakeAlarm(demo_mode=True) + alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10 + for button in alarm._view.escape_buttons: + button.pack.reset_mock() + + alarm._update_timer() + + for button in alarm._view.escape_buttons: + button.pack.assert_called() + alarm._stop_beep.set() + + def test_not_reoffered_while_open(self) -> None: + """With the form already up, the button is not re-packed.""" + alarm = WakeAlarm(demo_mode=True) + alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10 + alarm._escape.is_open = True + for button in alarm._view.escape_buttons: + button.pack.reset_mock() + + alarm._update_timer() + + for button in alarm._view.escape_buttons: + button.pack.assert_not_called() + alarm._stop_beep.set() + + def test_open_swaps_every_surface(self) -> None: + """Opening the form hides the challenge and builds a form per monitor.""" + alarm = WakeAlarm(demo_mode=True) + surfaces = alarm._view.surfaces + assert len(surfaces) == len(TWO_OUTPUTS) + + with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls: + alarm._escape.open() + + assert alarm._escape.is_open is True + assert dialog_cls.call_count == len(surfaces) + # One dialog *per* container, not two against the same parent. + parents = [call.args[0] for call in dialog_cls.call_args_list] + assert {id(parent) for parent in parents} == { + id(view.container) for view in surfaces + } + for view in surfaces: + view.challenge_frame.pack_forget.assert_called_once_with() + alarm._stop_beep.set() + + def test_a_dead_surface_does_not_block_the_others(self) -> None: + """A monitor that has gone away raises TclError; the rest still swap.""" + alarm = WakeAlarm(demo_mode=True) + dead, live = alarm._view.surfaces + dead.challenge_frame.pack_forget.side_effect = tk.TclError("bad path name") + + with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls: + alarm._escape.open() + + assert dialog_cls.call_count == 1 + assert dialog_cls.call_args.args[0] is live.container + alarm._stop_beep.set() + + def test_cancel_puts_the_challenge_back(self) -> None: + """Backing out destroys every form and re-packs every challenge. + + The bool alone is not the property that matters: ``open`` took the + challenge off the screen, so a cancel that only flips a flag leaves + the alarm showing a dead form on both monitors. + """ + alarm = WakeAlarm(demo_mode=True) + surfaces = alarm._view.surfaces + dialogs: list[MagicMock] = [] + + def make_dialog(*_args: object, **_kwargs: object) -> MagicMock: + dialog = MagicMock() + dialogs.append(dialog) + return dialog + + with patch("wake_alarm._escape_flow.EscapeDialog", side_effect=make_dialog): + alarm._escape.open() + # The frame was packed once when the view was built; only the re-pack + # that close() owes us should count. + for view in surfaces: + view.challenge_frame.pack.reset_mock() + + alarm._escape.close() + + assert alarm._escape.is_open is False + assert alarm.dismissed is False + assert len(dialogs) == len(TWO_OUTPUTS) + for dialog in dialogs: + dialog.frame.destroy.assert_called_once_with() + for view in surfaces: + view.challenge_frame.pack.assert_called_once_with() + alarm._stop_beep.set() + + def test_granted_dismisses_without_a_skip(self) -> None: + """Using the hatch silences the alarm but earns no workout skip.""" + alarm = WakeAlarm(demo_mode=True) + + with patch.object(alarm, "_dismiss_alarm") as dismiss: + alarm._on_escape_granted() + + dismiss.assert_called_once_with(earned_skip=False) + alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_gating.py b/wake_alarm/tests/test_gating.py new file mode 100644 index 0000000..df72f77 --- /dev/null +++ b/wake_alarm/tests/test_gating.py @@ -0,0 +1,75 @@ +"""Tests for the day/dismissed/workout gate.""" + +from __future__ import annotations + +from unittest.mock import patch + +from wake_alarm._gating import ( + should_run_alarm as _should_run_alarm, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class TestShouldRunAlarm: + """Tests for _should_run_alarm.""" + + def test_returns_false_on_non_alarm_day(self) -> None: + """Return False when today is not an alarm day.""" + with patch( + "wake_alarm._gating.is_alarm_day", + return_value=False, + ): + assert _should_run_alarm() is False + + def test_returns_false_when_already_dismissed(self) -> None: + """Return False when alarm was already dismissed today.""" + with ( + patch( + "wake_alarm._gating.is_alarm_day", + return_value=True, + ), + patch( + "wake_alarm._gating.was_alarm_dismissed_today", + return_value=True, + ), + ): + assert _should_run_alarm() is False + + def test_returns_true_when_alarm_day_and_not_dismissed(self) -> None: + """Return True when today is alarm day and not yet dismissed.""" + with ( + patch( + "wake_alarm._gating.is_alarm_day", + return_value=True, + ), + patch( + "wake_alarm._gating.was_alarm_dismissed_today", + return_value=False, + ), + patch( + "wake_alarm._gating.was_workout_logged_today", + return_value=False, + ), + ): + assert _should_run_alarm() is True + + def test_returns_false_when_workout_already_logged(self) -> None: + """Return False when workout was already logged today.""" + with ( + patch( + "wake_alarm._gating.is_alarm_day", + return_value=True, + ), + patch( + "wake_alarm._gating.was_alarm_dismissed_today", + return_value=False, + ), + patch( + "wake_alarm._gating.was_workout_logged_today", + return_value=True, + ), + ): + assert _should_run_alarm() is False diff --git a/wake_alarm/tests/test_pcspkr.py b/wake_alarm/tests/test_pcspkr.py new file mode 100644 index 0000000..1c21e66 --- /dev/null +++ b/wake_alarm/tests/test_pcspkr.py @@ -0,0 +1,40 @@ +"""Tests for the PC-speaker fallback beep.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from wake_alarm._audio import ( + _beep_pcspkr, +) + + +class TestBeepPcspkr: + """Tests for _beep_pcspkr (evdev PC speaker).""" + + def test_writes_tone_then_zero_to_device(self) -> None: + """Successful path writes start-frequency then stop event.""" + + mock_dev = MagicMock() + mock_open_ctx = MagicMock() + mock_open_ctx.__enter__.return_value = mock_dev + mock_open_ctx.__exit__.return_value = False + with ( + patch( + "wake_alarm._tone.Path.open", + return_value=mock_open_ctx, + ), + patch("wake_alarm._audio.time.sleep"), + ): + _beep_pcspkr(1000, 0.05) + # First write carries the frequency, second write carries 0 (stop). + assert mock_dev.write.call_count == 2 + + def test_oserror_is_swallowed(self) -> None: + """OSError opening the device must not raise.""" + + with patch( + "wake_alarm._tone.Path.open", + side_effect=OSError("no device"), + ): + _beep_pcspkr(1000, 0.05) # must not raise diff --git a/wake_alarm/tests/test_sinks_activate.py b/wake_alarm/tests/test_sinks_activate.py new file mode 100644 index 0000000..12c8909 --- /dev/null +++ b/wake_alarm/tests/test_sinks_activate.py @@ -0,0 +1,158 @@ +"""Tests for forcing every card's output profile on.""" + +from __future__ import annotations + +from unittest.mock import patch + +from wake_alarm._constants import ALARM_AUDIO_SINK +from wake_alarm._sinks import ( + _activate_all_outputs, + _AudioState, + _CardInfo, +) + +# A representative `pactl list cards` block covering every parse branch: +# an HDMI card (off, with an available stereo output profile), an input-only +# card (no output profile), a card block with no Name line (must be skipped). +_CARDS_TEXT = """Card #54 +\tName: alsa_card.hdmi +\tDriver: module-alsa-card.c +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\t\toutput:hdmi-stereo: Stereo (sinks: 1, priority: 5900, available: yes) +\t\toutput:hdmi-surround: Surround (sinks: 1, priority: 600, available: no) +\t\tbroken (available: yes) +\tActive Profile: off +Card #55 +\tName: alsa_card.mic +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\tActive Profile: off +Card #56 +\tProfiles: +\t\toutput:noname: NoName (available: yes) +\tActive Profile: off +""" + +# A `pactl list sinks` block: a normal sink, plus one whose Volume line has +# no integer (to cover the "no digit" branch), plus a leading non-Name line. +_SINKS_TEXT = """Sink #62 +\tState: SUSPENDED +\tName: alsa_output.analog +\tMute: no +\tVolume: front-left: 65536 / 100% +\tDescription: Built-in Analog +Sink #63 +\tName: alsa_output.weird +\tMute: yes +\tVolume: n/a +""" + + +class TestActivateAllOutputs: + """Tests for _activate_all_outputs.""" + + def test_returns_none_when_pactl_missing(self) -> None: + """No pactl on PATH → returns None without touching audio.""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is None + mock_run.assert_not_called() + + def _sentinel_state(self) -> _AudioState: + return _AudioState( + default_sink="old", + card_profiles={}, + sink_volumes={}, + sink_mutes={}, + ) + + def test_activates_and_prefers_alarm_sink(self) -> None: + """Forces card profiles on, maxes sinks, defaults to ALARM_AUDIO_SINK.""" + state = self._sentinel_state() + cards = { + "card_hdmi": _CardInfo(None, ["output:hdmi-stereo"]), + "card_mic": _CardInfo(None, []), + } + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value=cards), + patch( + "wake_alarm._sinks._list_real_sinks", + return_value=[ALARM_AUDIO_SINK, "other"], + ), + patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is state + cmds = [call.args[0] for call in mock_run.call_args_list] + assert [ + "/usr/bin/pactl", + "set-card-profile", + "card_hdmi", + "output:hdmi-stereo", + ] in cmds + # The input-only card (no output profile) is never switched. + assert not any(c[2] == "card_mic" for c in cmds if c[1] == "set-card-profile") + assert ["/usr/bin/pactl", "set-default-sink", ALARM_AUDIO_SINK] in cmds + assert mock_audible.call_count == 2 + + def test_does_not_block_on_sink_enumeration(self) -> None: + """Activation checks present sinks once — no polling/sleeping loop. + + The per-cycle presence check in _play_on_all_sinks is what covers a + slow monitor sink, so activation must not spin waiting for it. + """ + state = self._sentinel_state() + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch( + "wake_alarm._sinks._list_real_sinks", + return_value=[ALARM_AUDIO_SINK], + ) as mock_list, + patch("wake_alarm._sinks._ensure_sink_audible"), + patch("wake_alarm._sinks.subprocess.run"), + ): + assert _activate_all_outputs() is state + # Enumerated exactly once — no retry/poll loop. + mock_list.assert_called_once() + + def test_falls_back_to_first_sink_when_alarm_sink_absent(self) -> None: + """ALARM_AUDIO_SINK not present → default is the first present sink.""" + state = self._sentinel_state() + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch("wake_alarm._sinks._list_real_sinks", return_value=["other"]), + patch("wake_alarm._sinks._ensure_sink_audible"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is state + cmds = [call.args[0] for call in mock_run.call_args_list] + assert ["/usr/bin/pactl", "set-default-sink", "other"] in cmds + + def test_returns_state_when_no_sink_present(self) -> None: + """No real sink present yet → returns state, no default, no maxing.""" + state = self._sentinel_state() + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch("wake_alarm._sinks._list_real_sinks", return_value=[]), + patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is state + mock_audible.assert_not_called() + cmds = [call.args[0] for call in mock_run.call_args_list] + assert not any(c[1] == "set-default-sink" for c in cmds) diff --git a/wake_alarm/tests/test_sinks_restore.py b/wake_alarm/tests/test_sinks_restore.py new file mode 100644 index 0000000..2ce174d --- /dev/null +++ b/wake_alarm/tests/test_sinks_restore.py @@ -0,0 +1,109 @@ +"""Tests for restoring the audio state captured at alarm start.""" + +from __future__ import annotations + +from unittest.mock import patch + +from wake_alarm._sinks import ( + _AudioState, + _restore_audio_state, +) + +# A representative `pactl list cards` block covering every parse branch: +# an HDMI card (off, with an available stereo output profile), an input-only +# card (no output profile), a card block with no Name line (must be skipped). +_CARDS_TEXT = """Card #54 +\tName: alsa_card.hdmi +\tDriver: module-alsa-card.c +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\t\toutput:hdmi-stereo: Stereo (sinks: 1, priority: 5900, available: yes) +\t\toutput:hdmi-surround: Surround (sinks: 1, priority: 600, available: no) +\t\tbroken (available: yes) +\tActive Profile: off +Card #55 +\tName: alsa_card.mic +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\tActive Profile: off +Card #56 +\tProfiles: +\t\toutput:noname: NoName (available: yes) +\tActive Profile: off +""" + +# A `pactl list sinks` block: a normal sink, plus one whose Volume line has +# no integer (to cover the "no digit" branch), plus a leading non-Name line. +_SINKS_TEXT = """Sink #62 +\tState: SUSPENDED +\tName: alsa_output.analog +\tMute: no +\tVolume: front-left: 65536 / 100% +\tDescription: Built-in Analog +Sink #63 +\tName: alsa_output.weird +\tMute: yes +\tVolume: n/a +""" + + +class TestRestoreAudioState: + """Tests for _restore_audio_state.""" + + def test_none_is_noop(self) -> None: + """None state → does nothing, no pactl lookup.""" + with patch("wake_alarm._sinks._pactl_path") as mock_path: + _restore_audio_state(None) + mock_path.assert_not_called() + + def test_no_pactl_returns_silently(self) -> None: + """State present but pactl missing → no raise, no run.""" + state = _AudioState( + default_sink="s", + card_profiles={}, + sink_volumes={}, + sink_mutes={}, + ) + with ( + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + _restore_audio_state(state) + mock_run.assert_not_called() + + def test_restores_everything(self) -> None: + """Restores card profiles, sink volumes, mutes, and the default sink.""" + state = _AudioState( + default_sink="analog", + card_profiles={"card_hdmi": "off"}, + sink_volumes={"analog": "65536"}, + sink_mutes={"analog": True}, + ) + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + _restore_audio_state(state) + cmds = [call.args[0] for call in mock_run.call_args_list] + assert ["/usr/bin/pactl", "set-card-profile", "card_hdmi", "off"] in cmds + assert ["/usr/bin/pactl", "set-sink-volume", "analog", "65536"] in cmds + assert ["/usr/bin/pactl", "set-sink-mute", "analog", "1"] in cmds + assert ["/usr/bin/pactl", "set-default-sink", "analog"] in cmds + + def test_skips_default_when_none(self) -> None: + """default_sink None → no set-default-sink command.""" + state = _AudioState( + default_sink=None, + card_profiles={}, + sink_volumes={"analog": "100"}, + sink_mutes={"analog": False}, + ) + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + _restore_audio_state(state) + cmds = [call.args[0] for call in mock_run.call_args_list] + assert not any(c[1] == "set-default-sink" for c in cmds) + # Unmuted sink restores with "0". + assert ["/usr/bin/pactl", "set-sink-mute", "analog", "0"] in cmds diff --git a/wake_alarm/tests/test_sinks_warn.py b/wake_alarm/tests/test_sinks_warn.py new file mode 100644 index 0000000..6aab23e --- /dev/null +++ b/wake_alarm/tests/test_sinks_warn.py @@ -0,0 +1,83 @@ +"""Tests for the no-real-sink warning.""" + +from __future__ import annotations + +from unittest.mock import patch + +from wake_alarm._sinks import ( + _warn_if_no_real_sink, +) + +# A representative `pactl list cards` block covering every parse branch: +# an HDMI card (off, with an available stereo output profile), an input-only +# card (no output profile), a card block with no Name line (must be skipped). +_CARDS_TEXT = """Card #54 +\tName: alsa_card.hdmi +\tDriver: module-alsa-card.c +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\t\toutput:hdmi-stereo: Stereo (sinks: 1, priority: 5900, available: yes) +\t\toutput:hdmi-surround: Surround (sinks: 1, priority: 600, available: no) +\t\tbroken (available: yes) +\tActive Profile: off +Card #55 +\tName: alsa_card.mic +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\tActive Profile: off +Card #56 +\tProfiles: +\t\toutput:noname: NoName (available: yes) +\tActive Profile: off +""" + +# A `pactl list sinks` block: a normal sink, plus one whose Volume line has +# no integer (to cover the "no digit" branch), plus a leading non-Name line. +_SINKS_TEXT = """Sink #62 +\tState: SUSPENDED +\tName: alsa_output.analog +\tMute: no +\tVolume: front-left: 65536 / 100% +\tDescription: Built-in Analog +Sink #63 +\tName: alsa_output.weird +\tMute: yes +\tVolume: n/a +""" + + +class TestWarnIfNoRealSink: + """Tests for _warn_if_no_real_sink.""" + + def test_logs_when_pactl_missing(self) -> None: + """No pactl on PATH → warns and returns (no sink lookup).""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks._list_real_sinks") as mock_list, + ): + _warn_if_no_real_sink() + mock_list.assert_not_called() + + def test_warns_when_no_real_sink(self) -> None: + """No real sink → warning is emitted.""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._list_real_sinks", return_value=[]), + patch("wake_alarm._sinks._logger") as mock_log, + ): + _warn_if_no_real_sink() + mock_log.warning.assert_called() + + def test_info_when_real_sink_present(self) -> None: + """A real sink → info log, no warning.""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch( + "wake_alarm._sinks._list_real_sinks", + return_value=["alsa_output.analog"], + ), + patch("wake_alarm._sinks._logger") as mock_log, + ): + _warn_if_no_real_sink() + mock_log.info.assert_called() + mock_log.warning.assert_not_called() diff --git a/wake_alarm/tests/test_tone.py b/wake_alarm/tests/test_tone.py new file mode 100644 index 0000000..061e6ff --- /dev/null +++ b/wake_alarm/tests/test_tone.py @@ -0,0 +1,53 @@ +"""Tests for alarm tone synthesis and caching.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +if TYPE_CHECKING: + import pathlib + +from wake_alarm._tone import _ensure_tone_wav + + +class TestEnsureToneWav: + """Tests for _ensure_tone_wav (sine WAV generator + cache).""" + + def test_generates_and_caches(self, tmp_path: pathlib.Path) -> None: + """First call generates the WAV; second call returns the cached path.""" + from wake_alarm import _tone as tone_mod + + tone_mod._TONE_CACHE.clear() + with patch( + "wake_alarm._tone.tempfile.gettempdir", + return_value=str(tmp_path), + ): + path1 = _ensure_tone_wav(440) + assert path1.exists() + size = path1.stat().st_size + assert size > 0 + # Second call must hit the cache (no regeneration). + with patch("wake_alarm._tone.wave.open") as mock_open: + path2 = _ensure_tone_wav(440) + mock_open.assert_not_called() + assert path2 == path1 + tone_mod._TONE_CACHE.clear() + + def test_regenerates_when_cached_file_missing( + self, + tmp_path: pathlib.Path, + ) -> None: + """If the cached file was deleted, regenerate it.""" + from wake_alarm._tone import _TONE_CACHE + + _TONE_CACHE.clear() + with patch( + "wake_alarm._tone.tempfile.gettempdir", + return_value=str(tmp_path), + ): + path1 = _ensure_tone_wav(880) + path1.unlink() + path2 = _ensure_tone_wav(880) + assert path2.exists() + _TONE_CACHE.clear()