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

Phase 3: remove the max-fan ramp, split _alarm.py, wire the new modules

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EckWmyfetYq3Mi8MAnUudh
This commit is contained in:
Krzysztof Rudnicki 2026-08-26 23:46:10 +02:00
parent aa01e0e3e6
commit 9e0cd123ff
36 changed files with 2470 additions and 1661 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

328
wake_alarm/_alarm.py
View File

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

34
wake_alarm/_alarm_state.py Normal file
View File

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

109
wake_alarm/_alarm_timers.py Normal file
View File

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

159
wake_alarm/_audio.py
View File

@ -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("<h", sample))
wav.writeframesraw(bytes(frames))
_TONE_CACHE[frequency] = path
return path
def _try_player(binary: str, wav: Path) -> 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)

99
wake_alarm/_beep_loop.py Normal file
View File

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

48
wake_alarm/_brightness.py Normal file
View File

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

6
wake_alarm/_constants.py
View File

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

92
wake_alarm/_countdown.py Normal file
View File

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

58
wake_alarm/_dismiss_view.py Normal file
View File

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

81
wake_alarm/_gating.py Normal file
View File

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

55
wake_alarm/_lock_setup.py Normal file
View File

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

120
wake_alarm/_rounds.py Normal file
View File

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

42
wake_alarm/_screen_flash.py Normal file
View File

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

47
wake_alarm/_startup.py Normal file
View File

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

48
wake_alarm/_tone.py Normal file
View File

@ -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("<h", sample))
wav.writeframesraw(bytes(frames))
_TONE_CACHE[frequency] = path
return path

215
wake_alarm/tests/test_alarm.py
View File

@ -2,8 +2,6 @@
from __future__ import annotations
import pathlib
import subprocess
import tkinter as tk
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
@ -13,17 +11,10 @@ import pytest
if TYPE_CHECKING:
from collections.abc import Generator
from wake_alarm._alarm import (
_is_alarm_day,
_should_run_alarm,
)
from wake_alarm._audio import (
_beep_loud,
_beep_medium,
_beep_soft,
_find_fan_hwmon,
_max_fans,
_restore_fans,
_speaker_test_path,
)
from wake_alarm._challenges import (
@ -33,6 +24,9 @@ from wake_alarm._challenges import (
from wake_alarm._constants import (
DISMISS_CODE_LENGTH,
)
from wake_alarm._gating import (
is_alarm_day as _is_alarm_day,
)
# ---------------------------------------------------------------------------
# Helpers
@ -60,6 +54,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",
@ -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

186
wake_alarm/tests/test_alarm_audio.py
View File

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

161
wake_alarm/tests/test_alarm_beep_phases.py
View File

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

232
wake_alarm/tests/test_alarm_dismiss.py Normal file
View File

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

139
wake_alarm/tests/test_alarm_fanout.py Normal file
View File

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

20
wake_alarm/tests/test_alarm_flash_challenge.py
View File

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

149
wake_alarm/tests/test_alarm_main.py Normal file
View File

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

231
wake_alarm/tests/test_alarm_part2.py
View File

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

14
wake_alarm/tests/test_alarm_part3.py
View File

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

77
wake_alarm/tests/test_alarm_part4.py
View File

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

216
wake_alarm/tests/test_alarm_sinks.py
View File

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

151
wake_alarm/tests/test_beep_loop.py Normal file
View File

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

118
wake_alarm/tests/test_brightness.py Normal file
View File

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

149
wake_alarm/tests/test_escape_flow.py
View File

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

229
wake_alarm/tests/test_escape_integration.py Normal file
View File

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

75
wake_alarm/tests/test_gating.py Normal file
View File

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

40
wake_alarm/tests/test_pcspkr.py Normal file
View File

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

158
wake_alarm/tests/test_sinks_activate.py Normal file
View File

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

109
wake_alarm/tests/test_sinks_restore.py Normal file
View File

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

83
wake_alarm/tests/test_sinks_warn.py Normal file
View File

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

53
wake_alarm/tests/test_tone.py Normal file
View File

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