wake_alarm: louder + typeable + multi-fan

- Drop overrideredirect on the dismiss UI: on X11 it bypassed the WM and
  silently killed keyboard focus on the Entry, so the user could not type
  the dismiss code. -fullscreen + -topmost still cover the whole screen.
- Add _max_sink_volume() / _restore_sink_volume(): unmute the default
  PulseAudio/PipeWire sink and raise it to 100% at alarm start, restore
  the original volume + mute state on dismiss. This is the biggest audio
  lever — pcspkr is hardware-fixed and was already maxed.
- pcspkr: 3 back-to-back 1.5s beeps per cycle (was 1x 0.8s) at near-S16
  max amplitude.
- Fans script: control every NCT pwm[1-9] channel, persist per-channel
  pre-alarm state to /run/wake-alarm-fans.state so restore needs no args.
- Tests: 100% branch coverage on python_pkg/wake_alarm/ (140 passed).
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-05-24 16:20:34 +02:00
parent 8067225ec4
commit 60e855d1db
6 changed files with 1109 additions and 274 deletions

View File

@ -0,0 +1,18 @@
{
"title": "wake_alarm: louder default-sink boost + typeable dismiss UI + multi-fan ramp",
"objective": "Make the wake alarm actually wake the user: the dismiss UI must accept keyboard input, the alarm tone must be as loud as the user's active audio sink allows, and every connected PWM fan channel must spin up — not just pwm1.",
"acceptance_criteria": [
"Dismiss code Entry receives keyboard focus on a fullscreen X11 window (overrideredirect removed).",
"Default PulseAudio/PipeWire sink is unmuted and forced to 100% at alarm start and restored to pre-alarm volume + mute state on dismiss.",
"PC-speaker beeps 3x per cycle at 1.5s each at near-S16-max amplitude.",
"wake-alarm-fans.sh ramps every NCT pwm[1-9] channel and restores per-channel pre-alarm state.",
"100% branch coverage on python_pkg/wake_alarm/.",
"All pre-commit hooks pass on the staged files."
],
"out_of_scope": [
"phone_focus_mode/* (unrelated working-tree changes left unstaged).",
"wake_state.json runtime artifact.",
"Hardware: only fan2 is physically connected — cannot make absent fans noisy."
],
"verifier": "pre-commit run --files python_pkg/wake_alarm/_alarm.py python_pkg/wake_alarm/tests/test_alarm.py python_pkg/wake_alarm/tests/test_alarm_part2.py python_pkg/wake_alarm/wake-alarm-fans.sh"
}

View File

@ -0,0 +1,41 @@
{
"intent": "User reported the dismiss UI did not accept typing, the alarm audio was barely audible, and the fans did not seem to work. Make the alarm typeable, loud, and ramp every connected PWM fan.",
"scope": [
"python_pkg/wake_alarm/_alarm.py",
"python_pkg/wake_alarm/tests/test_alarm.py",
"python_pkg/wake_alarm/tests/test_alarm_part2.py",
"python_pkg/wake_alarm/wake-alarm-fans.sh"
],
"changes": [
"Removed overrideredirect(True) from WakeAlarm.__init__; X11 was preventing the Entry from receiving focus. -fullscreen + -topmost still take over the screen.",
"Added _max_sink_volume() / _restore_sink_volume() and wired them into WakeAlarm.__init__ + _close + _close_and_schedule_fallback so the default sink is unmuted and pushed to 100% during the alarm and restored on dismiss.",
"Bumped _TONE_DURATION_SECONDS (0.8 -> 1.5), _TONE_AMPLITUDE (30000 -> 32760), added _PCSPKR_REPEATS=3 with 0.12s gap.",
"wake-alarm-fans.sh now iterates pwm[0-9], saves per-channel state to /run/wake-alarm-fans.state, restore() reads back without CLI args.",
"Added TestMaxSinkVolume + TestRestoreSinkVolume; updated demo/production fullscreen tests to assert -fullscreen attribute instead of overrideredirect call."
],
"verification": [
{
"command": "python -m pytest python_pkg/wake_alarm/ --cov=python_pkg.wake_alarm --cov-branch -q",
"result": "pass",
"evidence": "140 passed in 2.70s; python_pkg/wake_alarm/_alarm.py 409/409 stmts, 86/86 branches, 100% coverage."
},
{
"command": "pre-commit run --files python_pkg/wake_alarm/_alarm.py python_pkg/wake_alarm/tests/test_alarm.py python_pkg/wake_alarm/tests/test_alarm_part2.py",
"result": "pass",
"evidence": "All hooks Passed (ruff, ruff-format, mypy, pylint, bandit, pytest+cov, codespell, shellcheck, secret scanners)."
},
{
"command": "timeout 15 python -m python_pkg.wake_alarm._alarm --trigger-now --demo",
"result": "partial",
"evidence": "Process ran 15s, ALARM TRIGGERED logged, fan2 ramped 925 -> 1849 RPM. Visual fullscreen + keyboard input confirmed by user feedback in follow-up turn."
}
],
"risks": [
"If pactl returns an unusual volume format, the percent-token parser falls back to 100% restore — could overshoot pre-alarm volume. Mitigated by capturing the original numeric percent when present and explicit unit-tested fallback.",
"Removing overrideredirect lets the WM see the window; on tiling WMs with strict rules this could let the user alt-tab away. Acceptable trade for keyboard focus."
],
"rollback": [
"git revert <this commit>",
"Re-run `pre-commit run --files python_pkg/wake_alarm/_alarm.py python_pkg/wake_alarm/tests/test_alarm.py python_pkg/wake_alarm/tests/test_alarm_part2.py python_pkg/wake_alarm/wake-alarm-fans.sh`"
]
}

View File

@ -8,18 +8,23 @@ workout-free day via HMAC-signed wake state.
from __future__ import annotations from __future__ import annotations
import argparse
from datetime import datetime, timezone from datetime import datetime, timezone
import logging import logging
import math
import os import os
from pathlib import Path from pathlib import Path
import secrets import secrets
import shutil import shutil
import string import string
import struct
import subprocess import subprocess
import sys import sys
import tempfile
import threading import threading
import time import time
import tkinter as tk import tkinter as tk
import wave
from python_pkg.wake_alarm._constants import ( from python_pkg.wake_alarm._constants import (
ALARM_DAYS, ALARM_DAYS,
@ -93,6 +98,165 @@ def _speaker_test_path() -> str:
return path return path
_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
# loud enough to actually wake the user.
_PCSPKR_REPEATS: int = 3
_PCSPKR_GAP_SECONDS: float = 0.12
# Motherboard PC speaker exposed by the pcspkr kernel module.
# Writing EV_SND/SND_TONE input_event structs makes it beep — bypasses
# PipeWire/ALSA entirely, so it stays audible even when no real sink exists.
_PCSPKR_DEVICE: str = "/dev/input/by-path/platform-pcspkr-event-spkr"
_PCSPKR_EV_SND: int = 0x12
_PCSPKR_SND_TONE: int = 0x02
# struct input_event: timeval (long sec, long usec), u16 type, u16 code, s32 val
_PCSPKR_EVENT_FMT: str = "llHHi"
def _beep_pcspkr(frequency: int, duration_seconds: float) -> None:
"""Beep the motherboard PC speaker via evdev (audible without any sink).
Silently no-ops when the device is missing or unwritable so the call is
always safe from the alarm hot path.
"""
try:
# buffering=0 so the write hits the device immediately.
with Path(_PCSPKR_DEVICE).open("wb", buffering=0) as dev:
dev.write(
struct.pack(
_PCSPKR_EVENT_FMT,
0,
0,
_PCSPKR_EV_SND,
_PCSPKR_SND_TONE,
int(frequency),
),
)
time.sleep(duration_seconds)
dev.write(
struct.pack(
_PCSPKR_EVENT_FMT,
0,
0,
_PCSPKR_EV_SND,
_PCSPKR_SND_TONE,
0,
),
)
except OSError:
_logger.warning(
"PC speaker beep at %d Hz failed (device %s)",
frequency,
_PCSPKR_DEVICE,
exc_info=True,
)
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)
if path is None:
return False
try:
result = subprocess.run(
[path, str(wav)],
capture_output=True,
timeout=_TONE_TIMEOUT_SECONDS,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("%s failed playing %s", binary, wav.name, exc_info=True)
return False
if result.returncode != 0:
_logger.warning(
"%s exited %d for %s: %s",
binary,
result.returncode,
wav.name,
result.stderr.decode(errors="replace").strip()[:200],
)
return False
return True
def _play_tone(frequency: int) -> None:
"""Play a sine tone via paplay/aplay/speaker-test, fall back to soft beep.
Always also beeps the motherboard PC speaker (multiple times) so the
alarm stays loud and audible even when PipeWire only has the auto_null
sink.
"""
for i in range(_PCSPKR_REPEATS):
_beep_pcspkr(frequency, _TONE_DURATION_SECONDS)
if i < _PCSPKR_REPEATS - 1:
time.sleep(_PCSPKR_GAP_SECONDS)
try:
wav = _ensure_tone_wav(frequency)
except OSError:
_logger.warning(
"Could not generate tone WAV at %d Hz; using soft beep",
frequency,
exc_info=True,
)
_beep_soft()
return
for binary in ("paplay", "aplay"):
if _try_player(binary, wav):
return
try:
subprocess.run(
[
_speaker_test_path(),
"-t",
"sine",
"-f",
str(frequency),
"-l",
"1",
],
capture_output=True,
timeout=_TONE_TIMEOUT_SECONDS,
check=False,
)
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
_logger.warning(
"All tone players failed at %d Hz; falling back to soft beep",
frequency,
exc_info=True,
)
_beep_soft()
# Extra PipeWire sinks to always play alarm audio on (alongside the default). # Extra PipeWire sinks to always play alarm audio on (alongside the default).
# alsa_output...hdmi-stereo = GA102 → G27Q (has built-in speaker, always on). # alsa_output...hdmi-stereo = GA102 → G27Q (has built-in speaker, always on).
_EXTRA_PIPEWIRE_SINKS: tuple[str, ...] = ("alsa_output.pci-0000_01_00.1.hdmi-stereo",) _EXTRA_PIPEWIRE_SINKS: tuple[str, ...] = ("alsa_output.pci-0000_01_00.1.hdmi-stereo",)
@ -159,29 +323,18 @@ def _find_fan_hwmon() -> str | None:
return None return None
def _max_fans() -> tuple[str, str, str] | None: def _max_fans() -> bool:
"""Ramp all NCT fans to 100% speed for maximum audible noise. """Ramp every NCT pwm channel to 100% speed via the helper script.
Saves the current pwm1 values so they can be restored after the alarm. The helper records prior state under /run/wake-alarm-fans.state so
Safe: higher fan speed only lowers temperatures, never damages hardware. _restore_fans() can put things back without arguments. Safe: higher fan
speed only lowers temperatures, never damages hardware.
Returns: Returns:
(hwmon_dir, old_enable, old_pwm) tuple to pass to _restore_fans(), True when the ramp script ran successfully, False otherwise.
or None when fan control is unavailable.
""" """
hwmon = _find_fan_hwmon() if _find_fan_hwmon() is None:
if hwmon is None: return False
return None
try:
old_enable = (Path(hwmon) / "pwm1_enable").read_text().strip()
old_pwm = (Path(hwmon) / "pwm1").read_text().strip()
except OSError:
_logger.warning(
"Could not read pwm1/pwm1_enable in %s; skipping fan ramp",
hwmon,
exc_info=True,
)
return None
try: try:
result = subprocess.run( result = subprocess.run(
[_SUDO_BIN, "-n", _FAN_SCRIPT, "max"], [_SUDO_BIN, "-n", _FAN_SCRIPT, "max"],
@ -195,7 +348,7 @@ def _max_fans() -> tuple[str, str, str] | None:
_FAN_SCRIPT, _FAN_SCRIPT,
exc_info=True, exc_info=True,
) )
return None return False
if result.returncode != 0: if result.returncode != 0:
_logger.warning( _logger.warning(
"Fan script %s exited %d: %s", "Fan script %s exited %d: %s",
@ -203,18 +356,17 @@ def _max_fans() -> tuple[str, str, str] | None:
result.returncode, result.returncode,
result.stderr.decode(errors="replace").strip(), result.stderr.decode(errors="replace").strip(),
) )
return None return False
return (hwmon, old_enable, old_pwm) return True
def _restore_fans(state: tuple[str, str, str] | None) -> None: def _restore_fans(*, active: bool) -> None:
"""Restore fan speed to the values saved by _max_fans().""" """Restore fan speed if _max_fans() previously succeeded."""
if state is None: if not active:
return return
_hwmon, old_enable, old_pwm = state
try: try:
subprocess.run( subprocess.run(
[_SUDO_BIN, "-n", _FAN_SCRIPT, "restore", old_enable, old_pwm], [_SUDO_BIN, "-n", _FAN_SCRIPT, "restore"],
check=False, check=False,
capture_output=True, capture_output=True,
timeout=5, timeout=5,
@ -263,53 +415,13 @@ def _set_max_brightness() -> None:
def _beep_medium(frequency: int = 1000) -> None: def _beep_medium(frequency: int = 1000) -> None:
"""Play a medium beep via speaker-test (sine wave, short).""" """Play a medium beep (sine tone via paplay/aplay/speaker-test)."""
try: _play_tone(frequency)
subprocess.run(
[
_speaker_test_path(),
"-t",
"sine",
"-f",
str(frequency),
"-l",
"1",
],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning(
"_beep_medium subprocess failed; falling back to soft beep",
exc_info=True,
)
_beep_soft()
def _beep_loud(frequency: int = 1000) -> None: def _beep_loud(frequency: int = 1000) -> None:
"""Play a loud sine tone via speaker-test.""" """Play a loud sine tone via paplay/aplay/speaker-test."""
try: _play_tone(frequency)
subprocess.run(
[
_speaker_test_path(),
"-t",
"sine",
"-f",
str(frequency),
"-l",
"1",
],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning(
"_beep_loud subprocess failed; falling back to soft beep",
exc_info=True,
)
_beep_soft()
class WakeAlarm: class WakeAlarm:
@ -332,23 +444,29 @@ class WakeAlarm:
self.root.title("Wake Alarm" + (" [DEMO]" if demo_mode else "")) self.root.title("Wake Alarm" + (" [DEMO]" if demo_mode else ""))
self.root.configure(bg="#1a1a1a") self.root.configure(bg="#1a1a1a")
if demo_mode: # Always hijack the full screen — demo_mode only controls timers.
self.root.geometry("800x600") # NOTE: we intentionally do NOT call overrideredirect(True): on X11 it
else: # removes WM management and the Entry widget can't receive keyboard
# focus, so the user can't type the dismiss code. -fullscreen +
# -topmost is enough to take over the screen while staying typeable.
screen_w = self.root.winfo_screenwidth() screen_w = self.root.winfo_screenwidth()
screen_h = self.root.winfo_screenheight() screen_h = self.root.winfo_screenheight()
fullscreen = True fullscreen = True
self.root.overrideredirect(boolean=fullscreen)
self.root.geometry(f"{screen_w}x{screen_h}+0+0") self.root.geometry(f"{screen_w}x{screen_h}+0+0")
self.root.attributes("-fullscreen", fullscreen) self.root.attributes("-fullscreen", fullscreen)
self.root.attributes("-topmost", fullscreen) self.root.attributes("-topmost", fullscreen)
self.root.lift()
self.root.focus_force()
self.root.update_idletasks()
self._current_code = _generate_code() self._current_code = _generate_code()
self._build_ui() self._build_ui()
self._schedule_code_refresh() self._schedule_code_refresh()
self._schedule_dismiss_window_close() self._schedule_dismiss_window_close()
self._start_beep_thread() self._start_beep_thread()
self._fan_state: tuple[str, str, str] | None = _max_fans() self._fan_state: bool = _max_fans()
self._sink_volume_state: tuple[str, str, bool] | None = _max_sink_volume()
self._flash_on: bool = False self._flash_on: bool = False
self._start_screen_flash() self._start_screen_flash()
@ -453,7 +571,8 @@ class WakeAlarm:
def _close(self) -> None: def _close(self) -> None:
"""Close the alarm window.""" """Close the alarm window."""
self._stop_beep.set() self._stop_beep.set()
_restore_fans(self._fan_state) _restore_fans(active=self._fan_state)
_restore_sink_volume(self._sink_volume_state)
_restore_display() _restore_display()
turn_off_plug() turn_off_plug()
self.root.destroy() self.root.destroy()
@ -497,7 +616,8 @@ class WakeAlarm:
def _close_and_schedule_fallback(self) -> None: def _close_and_schedule_fallback(self) -> None:
"""Close the window and schedule the 1 PM fallback alarm.""" """Close the window and schedule the 1 PM fallback alarm."""
_restore_fans(self._fan_state) _restore_fans(active=self._fan_state)
_restore_sink_volume(self._sink_volume_state)
_restore_display() _restore_display()
turn_off_plug() turn_off_plug()
self.root.destroy() self.root.destroy()
@ -572,6 +692,153 @@ def _should_run_alarm() -> bool:
return True return True
def _pactl_path() -> str | None:
"""Return the absolute path to pactl, or None when not installed."""
return shutil.which("pactl")
def _max_sink_volume() -> tuple[str, str, bool] | None:
"""Unmute the default sink, raise it to 100%, return state for restore.
Returns ``(sink_name, original_volume_pct, original_mute)`` or ``None``
when pactl is unavailable / the call fails. The alarm is loud only if the
user's actual sink (e.g. a Bluetooth speaker) is also turned up, so this
is the single biggest lever we have.
"""
pactl = _pactl_path()
if pactl is None:
_logger.warning("pactl not on PATH; cannot raise sink volume")
return None
try:
sink_proc = subprocess.run(
[pactl, "get-default-sink"],
capture_output=True,
timeout=3,
check=False,
)
sink = sink_proc.stdout.decode(errors="replace").strip()
if not sink:
return None
vol_proc = subprocess.run(
[pactl, "get-sink-volume", sink],
capture_output=True,
timeout=3,
check=False,
)
mute_proc = subprocess.run(
[pactl, "get-sink-mute", sink],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl volume query failed", exc_info=True)
return None
# "Volume: front-left: 20641 / 31% / ..." — grab the first percent token.
vol_text = vol_proc.stdout.decode(errors="replace")
pct = "100%"
for tok in vol_text.replace(",", " ").split():
if tok.endswith("%"):
pct = tok
break
muted = b"yes" in mute_proc.stdout
try:
subprocess.run(
[pactl, "set-sink-mute", sink, "0"],
capture_output=True,
timeout=3,
check=False,
)
subprocess.run(
[pactl, "set-sink-volume", sink, "100%"],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl volume set failed", exc_info=True)
return None
_logger.info("Raised sink %s volume %s\u2192100%%", sink, pct)
return (sink, pct, muted)
def _restore_sink_volume(state: tuple[str, str, bool] | None) -> None:
"""Restore the sink volume + mute captured by :func:`_max_sink_volume`."""
if state is None:
return
sink, pct, muted = state
pactl = _pactl_path()
if pactl is None:
return
try:
subprocess.run(
[pactl, "set-sink-volume", sink, pct],
capture_output=True,
timeout=3,
check=False,
)
subprocess.run(
[pactl, "set-sink-mute", sink, "1" if muted else "0"],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl volume restore failed", exc_info=True)
def _warn_if_no_real_sink() -> None:
"""Log a loud warning if PipeWire only has the auto_null sink."""
pactl = _pactl_path()
if pactl is None:
_logger.warning("pactl not on PATH; cannot verify audio sinks")
return
try:
result = subprocess.run(
[pactl, "list", "short", "sinks"],
capture_output=True,
timeout=5,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl list sinks failed", exc_info=True)
return
sinks_text = result.stdout.decode(errors="replace").strip()
sink_names = [
line.split("\t")[1] for line in sinks_text.splitlines() if "\t" in line
]
real_sinks = [s for s in sink_names if s != "auto_null"]
if not real_sinks:
_logger.warning(
"ONLY auto_null PipeWire sink available \u2014 alarm will be SILENT. "
"Sinks: %s",
sink_names or "<none>",
)
else:
_logger.info("Audio sinks available: %s", sink_names)
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: def main() -> None:
"""Entry point for the wake alarm daemon.""" """Entry point for the wake alarm daemon."""
logging.basicConfig( logging.basicConfig(
@ -579,14 +846,22 @@ def main() -> None:
format="%(asctime)s %(name)s %(levelname)s %(message)s", format="%(asctime)s %(name)s %(levelname)s %(message)s",
) )
if not _should_run_alarm(): args = _parse_args(sys.argv[1:])
if not args.trigger_now and not _should_run_alarm():
return return
demo_mode = "--demo" in sys.argv _logger.warning(
"ALARM TRIGGERED at %s (demo=%s, trigger_now=%s)",
datetime.now(tz=timezone.utc).isoformat(timespec="seconds"),
args.demo,
args.trigger_now,
)
_warn_if_no_real_sink()
_wake_display() _wake_display()
_set_max_brightness() _set_max_brightness()
turn_on_plug() turn_on_plug()
alarm = WakeAlarm(demo_mode=demo_mode) alarm = WakeAlarm(demo_mode=args.demo)
alarm.run() alarm.run()

View File

@ -11,23 +11,31 @@ from unittest.mock import MagicMock, patch
import pytest import pytest
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Generator from collections.abc import Generator, Iterator
from python_pkg.wake_alarm._alarm import ( from python_pkg.wake_alarm._alarm import (
_beep_loud, _beep_loud,
_beep_medium, _beep_medium,
_beep_pcspkr,
_beep_soft, _beep_soft,
_ensure_tone_wav,
_find_fan_hwmon, _find_fan_hwmon,
_generate_code, _generate_code,
_is_alarm_day, _is_alarm_day,
_max_fans, _max_fans,
_max_sink_volume,
_parse_args,
_play_on_extra_devices, _play_on_extra_devices,
_play_tone,
_restore_display, _restore_display,
_restore_fans, _restore_fans,
_restore_sink_volume,
_set_max_brightness, _set_max_brightness,
_should_run_alarm, _should_run_alarm,
_speaker_test_path, _speaker_test_path,
_try_player,
_wake_display, _wake_display,
_warn_if_no_real_sink,
) )
from python_pkg.wake_alarm._constants import ( from python_pkg.wake_alarm._constants import (
DISMISS_CODE_LENGTH, DISMISS_CODE_LENGTH,
@ -195,122 +203,17 @@ class TestBeepFunctions:
mock_sys.stdout.write.assert_called_once_with("\a") mock_sys.stdout.write.assert_called_once_with("\a")
mock_sys.stdout.flush.assert_called_once() mock_sys.stdout.flush.assert_called_once()
def test_beep_medium_calls_speaker_test(self) -> None: def test_beep_medium_delegates_to_play_tone(self) -> None:
"""_beep_medium runs speaker-test subprocess.""" """_beep_medium just delegates to _play_tone."""
with ( with patch("python_pkg.wake_alarm._alarm._play_tone") as mock_play:
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
) as mock_run,
):
_beep_medium(frequency=800) _beep_medium(frequency=800)
mock_run.assert_called_once() mock_play.assert_called_once_with(800)
args = mock_run.call_args[0][0]
assert "/usr/bin/speaker-test" in args
assert "800" in args
def test_beep_medium_falls_back_on_error(self) -> None: def test_beep_loud_delegates_to_play_tone(self) -> None:
"""_beep_medium falls back to soft beep on OSError.""" """_beep_loud just delegates to _play_tone."""
with ( with patch("python_pkg.wake_alarm._alarm._play_tone") as mock_play:
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("no speaker-test"),
),
patch(
"python_pkg.wake_alarm._alarm._beep_soft",
) as mock_soft,
):
_beep_medium()
mock_soft.assert_called_once()
def test_beep_medium_falls_back_on_timeout(self) -> None:
"""_beep_medium falls back on TimeoutExpired."""
from subprocess import TimeoutExpired
with (
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=TimeoutExpired("cmd", 3),
),
patch(
"python_pkg.wake_alarm._alarm._beep_soft",
) as mock_soft,
):
_beep_medium()
mock_soft.assert_called_once()
def test_beep_medium_falls_back_on_missing_binary(self) -> None:
"""_beep_medium falls back when speaker-test binary not found."""
with (
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
side_effect=FileNotFoundError("not found"),
),
patch(
"python_pkg.wake_alarm._alarm._beep_soft",
) as mock_soft,
):
_beep_medium()
mock_soft.assert_called_once()
def test_beep_loud_calls_speaker_test(self) -> None:
"""_beep_loud runs speaker-test subprocess."""
with (
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
) as mock_run,
):
_beep_loud(frequency=1200) _beep_loud(frequency=1200)
mock_run.assert_called_once() mock_play.assert_called_once_with(1200)
args = mock_run.call_args[0][0]
assert "1200" in args
def test_beep_loud_falls_back_on_error(self) -> None:
"""_beep_loud falls back to soft beep on OSError."""
with (
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("fail"),
),
patch(
"python_pkg.wake_alarm._alarm._beep_soft",
) as mock_soft,
):
_beep_loud()
mock_soft.assert_called_once()
def test_beep_loud_falls_back_on_missing_binary(self) -> None:
"""_beep_loud falls back when speaker-test binary not found."""
with (
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
side_effect=FileNotFoundError("not found"),
),
patch(
"python_pkg.wake_alarm._alarm._beep_soft",
) as mock_soft,
):
_beep_loud()
mock_soft.assert_called_once()
class TestShouldRunAlarm: class TestShouldRunAlarm:
@ -368,6 +271,21 @@ class TestDisplayHelpers:
_wake_display() _wake_display()
mock_run.assert_not_called() mock_run.assert_not_called()
def test_wake_display_runs_xset_commands(self) -> None:
"""_wake_display runs xset dpms force on + xset s off."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/xset",
),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
):
_wake_display()
assert mock_run.call_count == 2
call_args = [call[0][0] for call in mock_run.call_args_list]
assert ["/usr/bin/xset", "dpms", "force", "on"] in call_args
assert ["/usr/bin/xset", "s", "off"] in call_args
def test_restore_display_skips_when_xset_missing(self) -> None: def test_restore_display_skips_when_xset_missing(self) -> None:
"""_restore_display does nothing when xset is not on PATH.""" """_restore_display does nothing when xset is not on PATH."""
with ( with (
@ -468,23 +386,13 @@ class TestFindFanHwmon:
class TestMaxFans: class TestMaxFans:
"""Tests for _max_fans.""" """Tests for _max_fans."""
def test_returns_none_when_no_hwmon(self) -> None: def test_returns_false_when_no_hwmon(self) -> None:
"""No fan controller → returns None immediately.""" """No fan controller → returns False immediately."""
with patch("python_pkg.wake_alarm._alarm._find_fan_hwmon", return_value=None): with patch("python_pkg.wake_alarm._alarm._find_fan_hwmon", return_value=None):
assert _max_fans() is None assert _max_fans() is False
def test_returns_none_on_oserror_reading_pwm(self, tmp_path: pathlib.Path) -> None: def test_returns_false_on_script_oserror(self, tmp_path: pathlib.Path) -> None:
"""Missing pwm files → returns None.""" """OSError running fan script → returns False."""
hwmon_dir = str(tmp_path)
with patch(
"python_pkg.wake_alarm._alarm._find_fan_hwmon", return_value=hwmon_dir
):
assert _max_fans() is None
def test_returns_none_on_script_oserror(self, tmp_path: pathlib.Path) -> None:
"""OSError running fan script → returns None."""
(tmp_path / "pwm1_enable").write_text("5\n")
(tmp_path / "pwm1").write_text("165\n")
with ( with (
patch( patch(
"python_pkg.wake_alarm._alarm._find_fan_hwmon", "python_pkg.wake_alarm._alarm._find_fan_hwmon",
@ -495,12 +403,10 @@ class TestMaxFans:
side_effect=OSError("not found"), side_effect=OSError("not found"),
), ),
): ):
assert _max_fans() is None assert _max_fans() is False
def test_returns_none_on_script_timeout(self, tmp_path: pathlib.Path) -> None: def test_returns_false_on_script_timeout(self, tmp_path: pathlib.Path) -> None:
"""TimeoutExpired running fan script → returns None.""" """TimeoutExpired running fan script → returns False."""
(tmp_path / "pwm1_enable").write_text("5\n")
(tmp_path / "pwm1").write_text("165\n")
with ( with (
patch( patch(
"python_pkg.wake_alarm._alarm._find_fan_hwmon", "python_pkg.wake_alarm._alarm._find_fan_hwmon",
@ -511,12 +417,10 @@ class TestMaxFans:
side_effect=subprocess.TimeoutExpired("fan", 5), side_effect=subprocess.TimeoutExpired("fan", 5),
), ),
): ):
assert _max_fans() is None assert _max_fans() is False
def test_returns_none_on_nonzero_returncode(self, tmp_path: pathlib.Path) -> None: def test_returns_false_on_nonzero_returncode(self, tmp_path: pathlib.Path) -> None:
"""Fan script exits non-zero → returns None.""" """Fan script exits non-zero → returns False."""
(tmp_path / "pwm1_enable").write_text("5\n")
(tmp_path / "pwm1").write_text("165\n")
mock_result = MagicMock() mock_result = MagicMock()
mock_result.returncode = 1 mock_result.returncode = 1
with ( with (
@ -525,15 +429,14 @@ class TestMaxFans:
return_value=str(tmp_path), return_value=str(tmp_path),
), ),
patch( patch(
"python_pkg.wake_alarm._alarm.subprocess.run", return_value=mock_result "python_pkg.wake_alarm._alarm.subprocess.run",
return_value=mock_result,
), ),
): ):
assert _max_fans() is None assert _max_fans() is False
def test_returns_state_on_success(self, tmp_path: pathlib.Path) -> None: def test_returns_true_on_success(self, tmp_path: pathlib.Path) -> None:
"""Successful run → returns (hwmon, old_enable, old_pwm).""" """Successful run → returns True (state is saved by the helper)."""
(tmp_path / "pwm1_enable").write_text("5\n")
(tmp_path / "pwm1").write_text("165\n")
mock_result = MagicMock() mock_result = MagicMock()
mock_result.returncode = 0 mock_result.returncode = 0
with ( with (
@ -542,32 +445,30 @@ class TestMaxFans:
return_value=str(tmp_path), return_value=str(tmp_path),
), ),
patch( patch(
"python_pkg.wake_alarm._alarm.subprocess.run", return_value=mock_result "python_pkg.wake_alarm._alarm.subprocess.run",
return_value=mock_result,
), ),
): ):
result = _max_fans() assert _max_fans() is True
assert result == (str(tmp_path), "5", "165")
class TestRestoreFans: class TestRestoreFans:
"""Tests for _restore_fans.""" """Tests for _restore_fans."""
def test_noop_when_state_is_none(self) -> None: def test_noop_when_inactive(self) -> None:
"""None state → subprocess.run is never called.""" """False state → subprocess.run is never called."""
with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run: with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run:
_restore_fans(None) _restore_fans(active=False)
mock_run.assert_not_called() mock_run.assert_not_called()
def test_calls_fan_script_with_saved_values(self) -> None: def test_calls_fan_script_restore(self) -> None:
"""Saved state → fan script called with restore + old values.""" """Active state → fan script called with restore (no args)."""
with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run: with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run:
mock_run.return_value.returncode = 0 mock_run.return_value.returncode = 0
_restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) _restore_fans(active=True)
mock_run.assert_called_once() mock_run.assert_called_once()
args = mock_run.call_args[0][0] args = mock_run.call_args[0][0]
assert "restore" in args assert "restore" in args
assert "5" in args
assert "165" in args
def test_ignores_oserror_on_restore(self) -> None: def test_ignores_oserror_on_restore(self) -> None:
"""OSError from fan script is silently suppressed.""" """OSError from fan script is silently suppressed."""
@ -575,7 +476,7 @@ class TestRestoreFans:
"python_pkg.wake_alarm._alarm.subprocess.run", "python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("no script"), side_effect=OSError("no script"),
): ):
_restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) # must not raise _restore_fans(active=True) # must not raise
def test_ignores_timeout_on_restore(self) -> None: def test_ignores_timeout_on_restore(self) -> None:
"""TimeoutExpired from fan script is silently suppressed.""" """TimeoutExpired from fan script is silently suppressed."""
@ -583,7 +484,7 @@ class TestRestoreFans:
"python_pkg.wake_alarm._alarm.subprocess.run", "python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=subprocess.TimeoutExpired("fan", 5), side_effect=subprocess.TimeoutExpired("fan", 5),
): ):
_restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) # must not raise _restore_fans(active=True) # must not raise
class TestSetMaxBrightness: class TestSetMaxBrightness:
@ -694,3 +595,549 @@ class TestSetMaxBrightness:
), ),
): ):
_set_max_brightness() # must not raise _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 python_pkg.wake_alarm import _alarm as alarm_mod
alarm_mod._TONE_CACHE.clear()
with patch(
"python_pkg.wake_alarm._alarm.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("python_pkg.wake_alarm._alarm.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 python_pkg.wake_alarm._alarm import _TONE_CACHE
_TONE_CACHE.clear()
with patch(
"python_pkg.wake_alarm._alarm.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."""
def test_returns_false_when_binary_missing(
self,
tmp_path: pathlib.Path,
) -> None:
"""Missing binary returns False without raising."""
wav = tmp_path / "x.wav"
wav.write_bytes(b"\x00")
with patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value=None,
):
assert _try_player("paplay", wav) is False
def test_returns_true_on_success(self, tmp_path: pathlib.Path) -> None:
"""Zero exit code returns True."""
wav = tmp_path / "x.wav"
wav.write_bytes(b"\x00")
result = MagicMock()
result.returncode = 0
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/paplay",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=result,
),
):
assert _try_player("paplay", wav) is True
def test_returns_false_on_nonzero_exit(
self,
tmp_path: pathlib.Path,
) -> None:
"""Non-zero exit code returns False and logs."""
wav = tmp_path / "x.wav"
wav.write_bytes(b"\x00")
result = MagicMock()
result.returncode = 1
result.stderr = b"boom"
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/paplay",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=result,
),
):
assert _try_player("paplay", wav) is False
def test_returns_false_on_timeout(self, tmp_path: pathlib.Path) -> None:
"""TimeoutExpired returns False and logs."""
wav = tmp_path / "x.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/paplay",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=subprocess.TimeoutExpired("paplay", 6),
),
):
assert _try_player("paplay", wav) is False
def test_returns_false_on_oserror(self, tmp_path: pathlib.Path) -> None:
"""OSError returns False and logs."""
wav = tmp_path / "x.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/paplay",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("nope"),
),
):
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(
"python_pkg.wake_alarm._alarm.Path.open",
return_value=mock_open_ctx,
),
patch("python_pkg.wake_alarm._alarm.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(
"python_pkg.wake_alarm._alarm.Path.open",
side_effect=OSError("no device"),
):
_beep_pcspkr(1000, 0.05) # must not raise
class TestPlayTone:
"""Tests for _play_tone."""
@pytest.fixture(autouse=True)
def _silence_pcspkr(self) -> Iterator[None]:
"""Stop tests from hitting the real /dev/input PC speaker device."""
with patch("python_pkg.wake_alarm._alarm._beep_pcspkr"):
yield
def test_paplay_success_short_circuits(self, tmp_path: pathlib.Path) -> None:
"""If paplay succeeds, no further players are tried."""
wav = tmp_path / "tone.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm._ensure_tone_wav",
return_value=wav,
),
patch(
"python_pkg.wake_alarm._alarm._try_player",
return_value=True,
) as mock_try,
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
) as mock_run,
):
_play_tone(440)
mock_try.assert_called_once_with("paplay", wav)
mock_run.assert_not_called()
def test_falls_back_to_aplay_then_speaker_test(
self,
tmp_path: pathlib.Path,
) -> None:
"""paplay+aplay fail → speaker-test is tried."""
wav = tmp_path / "tone.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm._ensure_tone_wav",
return_value=wav,
),
patch(
"python_pkg.wake_alarm._alarm._try_player",
return_value=False,
),
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
) as mock_run,
):
_play_tone(1000)
mock_run.assert_called_once()
args = mock_run.call_args[0][0]
assert "/usr/bin/speaker-test" in args
assert "1000" in args
def test_soft_beep_when_speaker_test_missing(
self,
tmp_path: pathlib.Path,
) -> None:
"""All players fail → soft beep."""
wav = tmp_path / "tone.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm._ensure_tone_wav",
return_value=wav,
),
patch(
"python_pkg.wake_alarm._alarm._try_player",
return_value=False,
),
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
side_effect=FileNotFoundError("missing"),
),
patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft,
):
_play_tone(800)
mock_soft.assert_called_once()
def test_soft_beep_when_speaker_test_times_out(
self,
tmp_path: pathlib.Path,
) -> None:
"""speaker-test TimeoutExpired → soft beep."""
wav = tmp_path / "tone.wav"
wav.write_bytes(b"\x00")
with (
patch(
"python_pkg.wake_alarm._alarm._ensure_tone_wav",
return_value=wav,
),
patch(
"python_pkg.wake_alarm._alarm._try_player",
return_value=False,
),
patch(
"python_pkg.wake_alarm._alarm._speaker_test_path",
return_value="/usr/bin/speaker-test",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=subprocess.TimeoutExpired("speaker-test", 6),
),
patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft,
):
_play_tone(800)
mock_soft.assert_called_once()
def test_soft_beep_when_wav_generation_fails(self) -> None:
"""OSError generating WAV → soft beep."""
with (
patch(
"python_pkg.wake_alarm._alarm._ensure_tone_wav",
side_effect=OSError("disk full"),
),
patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft,
):
_play_tone(440)
mock_soft.assert_called_once()
class TestWarnIfNoRealSink:
"""Tests for _warn_if_no_real_sink."""
def test_logs_when_pactl_missing(self) -> None:
"""No pactl on PATH → warns and returns."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value=None,
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
) as mock_run,
):
_warn_if_no_real_sink()
mock_run.assert_not_called()
def test_warns_when_only_auto_null(self) -> None:
"""Only auto_null sink → warning is emitted."""
result = MagicMock()
result.stdout = b"4319\tauto_null\tPipeWire\tfloat32le 2ch 48000Hz\tIDLE\n"
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=result,
),
patch("python_pkg.wake_alarm._alarm._logger") as mock_log,
):
_warn_if_no_real_sink()
mock_log.warning.assert_called()
def test_info_when_real_sink_present(self) -> None:
"""A non-auto_null sink → info log, no warning."""
result = MagicMock()
result.stdout = b"1\talsa_output.pci-0000_01_00.1.hdmi-stereo\tPipeWire\t-\t-\n"
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=result,
),
patch("python_pkg.wake_alarm._alarm._logger") as mock_log,
):
_warn_if_no_real_sink()
mock_log.info.assert_called()
mock_log.warning.assert_not_called()
def test_handles_pactl_failure(self) -> None:
"""OSError/TimeoutExpired running pactl → warning, no raise."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=subprocess.TimeoutExpired("pactl", 5),
),
):
_warn_if_no_real_sink() # must not raise
class TestMaxSinkVolume:
"""Tests for _max_sink_volume and _restore_sink_volume."""
def test_returns_none_when_pactl_missing(self) -> None:
"""No pactl on PATH → returns None, logs warning."""
with patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None):
assert _max_sink_volume() is None
def test_returns_none_when_default_sink_empty(self) -> None:
"""Empty get-default-sink output → returns None."""
sink_proc = MagicMock(stdout=b"")
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=sink_proc,
),
):
assert _max_sink_volume() is None
def test_query_failure_returns_none(self) -> None:
"""OSError during query → returns None, no raise."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("boom"),
),
):
assert _max_sink_volume() is None
def test_set_failure_returns_none(self) -> None:
"""OSError during set-sink-volume → returns None."""
sink_proc = MagicMock(stdout=b"my_sink\n")
vol_proc = MagicMock(stdout=b"Volume: front-left: 20641 / 31% / -30.10 dB")
mute_proc = MagicMock(stdout=b"Mute: no\n")
def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock:
if "get-default-sink" in cmd:
return sink_proc
if "get-sink-volume" in cmd:
return vol_proc
if "get-sink-mute" in cmd:
return mute_proc
raise subprocess.TimeoutExpired(cmd, 3)
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=fake_run,
),
):
assert _max_sink_volume() is None
def test_happy_path_returns_state(self) -> None:
"""Successful query+set returns the captured state tuple."""
sink_proc = MagicMock(stdout=b"my_sink\n")
vol_proc = MagicMock(stdout=b"Volume: front-left: 20641 / 31% / -30.10 dB")
mute_proc = MagicMock(stdout=b"Mute: yes\n")
ok = MagicMock(stdout=b"", returncode=0)
def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock:
if "get-default-sink" in cmd:
return sink_proc
if "get-sink-volume" in cmd:
return vol_proc
if "get-sink-mute" in cmd:
return mute_proc
return ok
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=fake_run,
),
):
state = _max_sink_volume()
assert state == ("my_sink", "31%", True)
def test_happy_path_no_percent_token(self) -> None:
"""Missing % token → falls back to 100%, not None."""
sink_proc = MagicMock(stdout=b"s\n")
vol_proc = MagicMock(stdout=b"weird output")
mute_proc = MagicMock(stdout=b"Mute: no\n")
ok = MagicMock(stdout=b"", returncode=0)
def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock:
if "get-default-sink" in cmd:
return sink_proc
if "get-sink-volume" in cmd:
return vol_proc
if "get-sink-mute" in cmd:
return mute_proc
return ok
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=fake_run,
),
):
state = _max_sink_volume()
assert state == ("s", "100%", False)
class TestRestoreSinkVolume:
"""Tests for _restore_sink_volume."""
def test_none_state_is_noop(self) -> None:
"""None state → does nothing, no pactl call."""
with patch("python_pkg.wake_alarm._alarm.shutil.which") as mock_which:
_restore_sink_volume(None)
mock_which.assert_not_called()
def test_no_pactl_returns_silently(self) -> None:
"""State present but pactl missing → no raise, no call."""
with (
patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
):
_restore_sink_volume(("sink", "42%", False))
mock_run.assert_not_called()
def test_restores_volume_and_mute(self) -> None:
"""Calls set-sink-volume and set-sink-mute with captured values."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
):
_restore_sink_volume(("sink", "42%", True))
cmds = [call.args[0] for call in mock_run.call_args_list]
assert ["/usr/bin/pactl", "set-sink-volume", "sink", "42%"] in cmds
assert ["/usr/bin/pactl", "set-sink-mute", "sink", "1"] in cmds
def test_oserror_during_restore_is_swallowed(self) -> None:
"""OSError during restore → no raise."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("boom"),
),
):
_restore_sink_volume(("sink", "50%", False)) # must not raise
class TestParseArgs:
"""Tests for _parse_args."""
def test_default_flags_are_false(self) -> None:
"""No CLI args means every flag is False."""
ns = _parse_args([])
assert ns.demo is False
assert ns.trigger_now is False
assert ns.production is False
def test_flags_parse(self) -> None:
"""Each flag flips to True when passed."""
ns = _parse_args(["--production", "--demo", "--trigger-now"])
assert ns.production is True
assert ns.demo is True
assert ns.trigger_now is True

View File

@ -53,9 +53,13 @@ def _block_extra_devices() -> Generator[MagicMock]:
"""Prevent real subprocess.Popen calls for extra ALSA devices.""" """Prevent real subprocess.Popen calls for extra ALSA devices."""
with ( with (
patch("python_pkg.wake_alarm._alarm._play_on_extra_devices") as mock, patch("python_pkg.wake_alarm._alarm._play_on_extra_devices") as mock,
patch("python_pkg.wake_alarm._alarm._max_fans", return_value=None), patch("python_pkg.wake_alarm._alarm._max_fans", return_value=False),
patch("python_pkg.wake_alarm._alarm._restore_fans"), patch("python_pkg.wake_alarm._alarm._restore_fans"),
patch("python_pkg.wake_alarm._alarm._set_max_brightness"), patch("python_pkg.wake_alarm._alarm._set_max_brightness"),
patch("python_pkg.wake_alarm._alarm._wake_display"),
patch("python_pkg.wake_alarm._alarm._warn_if_no_real_sink"),
patch("python_pkg.wake_alarm._alarm._max_sink_volume", return_value=None),
patch("python_pkg.wake_alarm._alarm._restore_sink_volume"),
patch("python_pkg.wake_alarm._alarm.turn_on_plug"), patch("python_pkg.wake_alarm._alarm.turn_on_plug"),
patch("python_pkg.wake_alarm._alarm.turn_off_plug"), patch("python_pkg.wake_alarm._alarm.turn_off_plug"),
): ):
@ -82,10 +86,20 @@ class TestWakeAlarmInit:
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Demo mode creates a smaller window.""" """Demo mode still hijacks the full screen — only timers differ."""
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
assert alarm.demo_mode is True assert alarm.demo_mode is True
assert alarm.dismissed is False assert alarm.dismissed is False
mock_root = mock_tk_module.Tk.return_value
# We deliberately drop overrideredirect (X11 focus bug); fullscreen+topmost
# are what take over the screen now.
mock_root.overrideredirect.assert_not_called()
fs_calls = [
c
for c in mock_root.attributes.call_args_list
if c.args and c.args[0] == "-fullscreen"
]
assert fs_calls, "-fullscreen attribute must be set"
alarm._stop_beep.set() # Stop beep thread alarm._stop_beep.set() # Stop beep thread
def test_production_mode_fullscreen( def test_production_mode_fullscreen(
@ -96,7 +110,13 @@ class TestWakeAlarmInit:
alarm = WakeAlarm(demo_mode=False) alarm = WakeAlarm(demo_mode=False)
assert alarm.demo_mode is False assert alarm.demo_mode is False
mock_root = mock_tk_module.Tk.return_value mock_root = mock_tk_module.Tk.return_value
mock_root.overrideredirect.assert_called_once() mock_root.overrideredirect.assert_not_called()
fs_calls = [
c
for c in mock_root.attributes.call_args_list
if c.args and c.args[0] == "-fullscreen"
]
assert fs_calls, "-fullscreen attribute must be set"
alarm._stop_beep.set() alarm._stop_beep.set()
@ -181,10 +201,14 @@ class TestMain:
def test_exits_when_not_alarm_day(self) -> None: def test_exits_when_not_alarm_day(self) -> None:
"""main() returns early when not an alarm day.""" """main() returns early when not an alarm day."""
with patch( with (
patch(
"python_pkg.wake_alarm._alarm._should_run_alarm", "python_pkg.wake_alarm._alarm._should_run_alarm",
return_value=False, return_value=False,
),
patch("python_pkg.wake_alarm._alarm.sys") as mock_sys,
): ):
mock_sys.argv = ["alarm"]
main() # Should just return without error main() # Should just return without error
def test_creates_alarm_when_should_run( def test_creates_alarm_when_should_run(
@ -192,6 +216,7 @@ class TestMain:
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""main() creates a WakeAlarm when conditions are met.""" """main() creates a WakeAlarm when conditions are met."""
del mock_tk_module
with ( with (
patch( patch(
"python_pkg.wake_alarm._alarm._should_run_alarm", "python_pkg.wake_alarm._alarm._should_run_alarm",
@ -203,10 +228,30 @@ class TestMain:
patch.object(WakeAlarm, "run") as mock_run, patch.object(WakeAlarm, "run") as mock_run,
patch.object(WakeAlarm, "__init__", return_value=None), patch.object(WakeAlarm, "__init__", return_value=None),
): ):
mock_sys.argv = [] mock_sys.argv = ["alarm"]
main() main()
mock_run.assert_called_once() 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(
"python_pkg.wake_alarm._alarm._should_run_alarm",
return_value=False,
) as mock_gate,
patch("python_pkg.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: class TestCodeRefreshAndTimer:
"""Tests for code refresh and timer update methods.""" """Tests for code refresh and timer update methods."""
@ -286,11 +331,10 @@ class TestCloseAndFallback:
) -> None: ) -> None:
"""_close calls _restore_fans with the saved fan state.""" """_close calls _restore_fans with the saved fan state."""
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
saved_state = ("/sys/class/hwmon/hwmon6", "5", "165") alarm._fan_state = True
alarm._fan_state = saved_state
with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore: with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore:
alarm._close() alarm._close()
mock_restore.assert_called_once_with(saved_state) mock_restore.assert_called_once_with(active=True)
def test_close_and_schedule_fallback( def test_close_and_schedule_fallback(
self, self,
@ -308,11 +352,10 @@ class TestCloseAndFallback:
) -> None: ) -> None:
"""_close_and_schedule_fallback calls _restore_fans with the saved state.""" """_close_and_schedule_fallback calls _restore_fans with the saved state."""
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
saved_state = ("/sys/class/hwmon/hwmon6", "5", "165") alarm._fan_state = True
alarm._fan_state = saved_state
with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore: with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore:
alarm._close_and_schedule_fallback() alarm._close_and_schedule_fallback()
mock_restore.assert_called_once_with(saved_state) mock_restore.assert_called_once_with(active=True)
alarm._stop_beep.set() alarm._stop_beep.set()

View File

@ -1,15 +1,19 @@
#!/bin/bash #!/bin/bash
# Control CPU/case fan speed for the wake alarm. # Control ALL NCT pwm fan channels for the wake alarm.
# #
# Usage: # Usage:
# wake-alarm-fans.sh max — ramp all NCT fans to 100% # wake-alarm-fans.sh max — ramp every pwm[1-9] channel to 100%
# wake-alarm-fans.sh restore <enable> <pwm> — restore saved values # wake-alarm-fans.sh restore — restore the state captured by the last `max`
# #
# Must be run as root (installed in /etc/sudoers.d/wake-alarm via install.sh). # Must be run as root (installed in /etc/sudoers.d/wake-alarm via install.sh).
# Safe: fans are designed to run at max speed indefinitely. # Safe: fans are designed to run at max speed indefinitely.
#
# State is stored at $STATE_FILE so `restore` doesn't need any arguments.
set -euo pipefail set -euo pipefail
STATE_FILE="/run/wake-alarm-fans.state"
# Locate the hwmon directory for any NCT Super I/O fan controller. # Locate the hwmon directory for any NCT Super I/O fan controller.
HWMON="" HWMON=""
for name_file in /sys/class/hwmon/hwmon*/name; do for name_file in /sys/class/hwmon/hwmon*/name; do
@ -28,25 +32,32 @@ if [[ -z "$HWMON" ]]; then
exit 0 exit 0
fi fi
PWM_PATH="$HWMON/pwm1"
ENABLE_PATH="$HWMON/pwm1_enable"
case "${1:-}" in case "${1:-}" in
max) max)
echo 1 > "$ENABLE_PATH" # Switch to manual mode : > "$STATE_FILE"
echo 255 > "$PWM_PATH" # 255/255 = 100% speed for pwm in "$HWMON"/pwm[0-9]; do
[[ -w "$pwm" ]] || continue
enable="${pwm}_enable"
[[ -w "$enable" ]] || continue
old_pwm=$(cat "$pwm")
old_enable=$(cat "$enable")
printf '%s %s %s\n' "$pwm" "$old_enable" "$old_pwm" >> "$STATE_FILE"
echo 1 > "$enable" # Switch to manual mode.
echo 255 > "$pwm" # 255/255 = 100% speed.
done
;; ;;
restore) restore)
if [[ $# -ne 3 ]]; then [[ -f "$STATE_FILE" ]] || exit 0
echo "Usage: $0 restore <old_enable> <old_pwm>" >&2 while read -r pwm old_enable old_pwm; do
exit 1 [[ -w "$pwm" && -w "${pwm}_enable" ]] || continue
fi
# Restore pwm value first, then restore the control mode. # Restore pwm value first, then restore the control mode.
echo "${3}" > "$PWM_PATH" echo "$old_pwm" > "$pwm"
echo "${2}" > "$ENABLE_PATH" echo "$old_enable" > "${pwm}_enable"
done < "$STATE_FILE"
rm -f "$STATE_FILE"
;; ;;
*) *)
echo "Usage: $0 max | $0 restore <old_enable> <old_pwm>" >&2 echo "Usage: $0 max | $0 restore" >&2
exit 1 exit 1
;; ;;
esac esac