morning_routine: unified alarm+lock orchestrator, fix alarm audio/reliability

- New python_pkg/morning_routine package: sequential orchestrator runs wake
  alarm then workout lock as blocking subprocesses (one fullscreen owner at
  a time). Deployed as morning-routine.service; sleep hook updated to start
  it on hibernate-resume instead of the standalone wake-alarm.service.

- wake_alarm: force G27Q HDMI card profile on at alarm time, poll up to 6s
  for sink to appear, set as default + unmute 100%. Alarm now persists until
  the typeable code is entered (no more silent 30-min give-up). Service gets
  DISPLAY=:0 + ExecStartPre sleep 1 to fix cold-boot Tkinter crash.

- phone_focus_mode/config.sh: whitelist Revolut, mObywatel, VaultKitBypass.

- 100% branch coverage maintained across wake_alarm and morning_routine.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-05-25 18:55:27 +02:00
parent 7aebf8e333
commit 8db7b8eb2a
7 changed files with 396 additions and 331 deletions

View File

@ -27,6 +27,11 @@ import tkinter as tk
import wave import wave
from python_pkg.wake_alarm._constants import ( from python_pkg.wake_alarm._constants import (
ALARM_AUDIO_CARD,
ALARM_AUDIO_PROFILE,
ALARM_AUDIO_SINK,
ALARM_AUDIO_SINK_POLL_SECONDS,
ALARM_AUDIO_SINK_WAIT_SECONDS,
ALARM_DAYS, ALARM_DAYS,
DISMISS_CODE_LENGTH, DISMISS_CODE_LENGTH,
DISMISS_CODE_REFRESH_SECONDS, DISMISS_CODE_REFRESH_SECONDS,
@ -461,12 +466,13 @@ class WakeAlarm:
self.root.update_idletasks() self.root.update_idletasks()
self._current_code = _generate_code() self._current_code = _generate_code()
self._skip_earnable: bool = True
self._build_ui() self._build_ui()
self._schedule_code_refresh() self._schedule_code_refresh()
self._schedule_dismiss_window_close() self._schedule_skip_window_close()
self._start_beep_thread() self._start_beep_thread()
self._fan_state: bool = _max_fans() self._fan_state: bool = _max_fans()
self._sink_volume_state: tuple[str, str, bool] | None = _max_sink_volume() self._audio_restore: str | None = _activate_alarm_audio()
self._flash_on: bool = False self._flash_on: bool = False
self._start_screen_flash() self._start_screen_flash()
@ -535,7 +541,7 @@ class WakeAlarm:
"""Handle code submission.""" """Handle code submission."""
entered = self._entry.get().strip() entered = self._entry.get().strip()
if entered == self._current_code: if entered == self._current_code:
self._dismiss_alarm(earned_skip=True) self._dismiss_alarm(earned_skip=self._skip_earnable)
else: else:
self._status_label.configure(text="Wrong code! Try again.") self._status_label.configure(text="Wrong code! Try again.")
self._entry.delete(0, tk.END) self._entry.delete(0, tk.END)
@ -572,7 +578,7 @@ class WakeAlarm:
"""Close the alarm window.""" """Close the alarm window."""
self._stop_beep.set() self._stop_beep.set()
_restore_fans(active=self._fan_state) _restore_fans(active=self._fan_state)
_restore_sink_volume(self._sink_volume_state) _restore_alarm_audio(self._audio_restore)
_restore_display() _restore_display()
turn_off_plug() turn_off_plug()
self.root.destroy() self.root.destroy()
@ -587,55 +593,45 @@ class WakeAlarm:
ms = DISMISS_CODE_REFRESH_SECONDS * 1000 if not self.demo_mode else 10_000 ms = DISMISS_CODE_REFRESH_SECONDS * 1000 if not self.demo_mode else 10_000
self.root.after(ms, self._schedule_code_refresh) self.root.after(ms, self._schedule_code_refresh)
def _schedule_dismiss_window_close(self) -> None: def _schedule_skip_window_close(self) -> None:
"""Close dismiss window after the allowed time.""" """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 ms = DISMISS_WINDOW_MINUTES * 60 * 1000 if not self.demo_mode else 30_000
self.root.after(ms, self._on_dismiss_window_expired) self.root.after(ms, self._on_skip_window_expired)
def _on_dismiss_window_expired(self) -> None: def _on_skip_window_expired(self) -> None:
"""Called when the dismiss window expires without valid dismissal.""" """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: if not self._active:
return return
self._active = False self._skip_earnable = False
self._stop_beep.set() self._info_label.configure(
save_wake_state(dismissed_at=None, skip_workout=False) text="Skip window closed - type the code to stop the alarm",
_logger.info("Dismiss window expired — no workout skip.") )
self._status_label.configure(text="No workout skip today.")
for widget in self._container.winfo_children(): _logger.info("Skip window expired - alarm continues until dismissed.")
widget.destroy()
tk.Label(
self._container,
text="Too late! No workout skip today.",
font=("Arial", 36, "bold"),
fg="#ff4444",
bg="#1a1a1a",
).pack(pady=30)
self.root.after(5000, self._close_and_schedule_fallback)
def _close_and_schedule_fallback(self) -> None:
"""Close the window and schedule the 1 PM fallback alarm."""
_restore_fans(active=self._fan_state)
_restore_sink_volume(self._sink_volume_state)
_restore_display()
turn_off_plug()
self.root.destroy()
def _update_timer(self) -> None: def _update_timer(self) -> None:
"""Update the remaining time display.""" """Show the skip-window countdown, then a keep-going silence prompt."""
if not self._active: if not self._active:
return return
elapsed = time.monotonic() - self._alarm_start elapsed = time.monotonic() - self._alarm_start
window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30 window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30
remaining = max(0, window - elapsed) remaining = max(0, window - elapsed)
minutes = int(remaining) // 60 if self._skip_earnable and remaining > 0:
seconds = int(remaining) % 60 minutes = int(remaining) // 60
self._timer_label.configure( seconds = int(remaining) % 60
text=f"Time remaining: {minutes:02d}:{seconds:02d}", self._timer_label.configure(
) text=f"Skip window: {minutes:02d}:{seconds:02d}",
if remaining > 0: )
self.root.after(1000, self._update_timer) else:
self._timer_label.configure(
text="No skip available - type the code to stop the alarm",
)
self.root.after(1000, self._update_timer)
def _start_beep_thread(self) -> None: def _start_beep_thread(self) -> None:
"""Start the background beep escalation thread.""" """Start the background beep escalation thread."""
@ -697,94 +693,99 @@ def _pactl_path() -> str | None:
return shutil.which("pactl") return shutil.which("pactl")
def _max_sink_volume() -> tuple[str, str, bool] | None: def _alarm_sink_present(pactl: str) -> bool:
"""Unmute the default sink, raise it to 100%, return state for restore. """Return True when the dedicated alarm HDMI sink exists in PipeWire."""
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: try:
sink_proc = subprocess.run( result = subprocess.run(
[pactl, "list", "short", "sinks"],
capture_output=True,
timeout=3,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl list sinks failed", exc_info=True)
return False
return ALARM_AUDIO_SINK in result.stdout.decode(errors="replace")
def _current_default_sink(pactl: str) -> str | None:
"""Return the current default sink name, or None on failure / empty."""
try:
result = subprocess.run(
[pactl, "get-default-sink"], [pactl, "get-default-sink"],
capture_output=True, capture_output=True,
timeout=3, timeout=3,
check=False, 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): except (OSError, subprocess.TimeoutExpired):
_logger.warning("pactl volume query failed", exc_info=True) _logger.warning("pactl get-default-sink failed", exc_info=True)
return None return None
# "Volume: front-left: 20641 / 31% / ..." — grab the first percent token. name = result.stdout.decode(errors="replace").strip()
vol_text = vol_proc.stdout.decode(errors="replace") return name or None
pct = "100%"
for tok in vol_text.replace(",", " ").split():
if tok.endswith("%"): def _activate_alarm_audio() -> str | None:
pct = tok """Force the monitor's HDMI output on and route the alarm to it.
At wake time the Bluetooth speaker is disconnected and PipeWire only has the
``auto_null`` sink, so the alarm is silent. This forces the HDMI card
profile on, waits for its sink to appear, makes it the default sink, and
raises it to full volume - empirically the only output audible on this
machine at wake time (the G27Q monitor's built-in speaker).
Returns:
The previous default sink name (to restore on close), or ``None`` when
the alarm audio sink could not be activated.
"""
pactl = _pactl_path()
if pactl is None:
_logger.warning("pactl not on PATH; cannot activate alarm audio")
return None
subprocess.run(
[pactl, "set-card-profile", ALARM_AUDIO_CARD, ALARM_AUDIO_PROFILE],
capture_output=True,
timeout=3,
check=False,
)
attempts = max(
1,
int(ALARM_AUDIO_SINK_WAIT_SECONDS / ALARM_AUDIO_SINK_POLL_SECONDS),
)
for _ in range(attempts):
if _alarm_sink_present(pactl):
break break
muted = b"yes" in mute_proc.stdout time.sleep(ALARM_AUDIO_SINK_POLL_SECONDS)
try: else:
subprocess.run( _logger.warning(
[pactl, "set-sink-mute", sink, "0"], "Alarm audio sink %s did not appear after %.0fs; alarm may be silent",
capture_output=True, ALARM_AUDIO_SINK,
timeout=3, ALARM_AUDIO_SINK_WAIT_SECONDS,
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 return None
_logger.info("Raised sink %s volume %s\u2192100%%", sink, pct) old_default = _current_default_sink(pactl)
return (sink, pct, muted) for cmd in (
[pactl, "set-default-sink", ALARM_AUDIO_SINK],
[pactl, "set-sink-mute", ALARM_AUDIO_SINK, "0"],
[pactl, "set-sink-volume", ALARM_AUDIO_SINK, "100%"],
):
subprocess.run(cmd, capture_output=True, timeout=3, check=False)
_logger.warning("Alarm audio routed to %s at 100%%", ALARM_AUDIO_SINK)
return old_default
def _restore_sink_volume(state: tuple[str, str, bool] | None) -> None: def _restore_alarm_audio(old_default: str | None) -> None:
"""Restore the sink volume + mute captured by :func:`_max_sink_volume`.""" """Restore the default sink captured by :func:`_activate_alarm_audio`."""
if state is None: if old_default is None:
return return
sink, pct, muted = state
pactl = _pactl_path() pactl = _pactl_path()
if pactl is None: if pactl is None:
return return
try: subprocess.run(
subprocess.run( [pactl, "set-default-sink", old_default],
[pactl, "set-sink-volume", sink, pct], capture_output=True,
capture_output=True, timeout=3,
timeout=3, check=False,
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: def _warn_if_no_real_sink() -> None:

View File

@ -38,6 +38,20 @@ WAKE_STATE_FILE: Path = Path(__file__).resolve().parent / "wake_state.json"
# rtcwake binary path # rtcwake binary path
RTCWAKE_BIN: str = "/usr/sbin/rtcwake" RTCWAKE_BIN: str = "/usr/sbin/rtcwake"
# Alarm audio output (machine-specific, empirically verified 2026-05-25).
# At wake time the Bluetooth speaker is disconnected and PipeWire only has the
# auto_null sink, so the alarm is silent unless we activate a real output. The
# only audible always-present output on this machine is the G27Q monitor's
# built-in speaker on the NVidia GPU's HDMI audio. WirePlumber leaves the card
# profile "off", so the alarm must force the profile on and wait for the sink.
ALARM_AUDIO_CARD: str = "alsa_card.pci-0000_01_00.1"
ALARM_AUDIO_PROFILE: str = "output:hdmi-stereo"
ALARM_AUDIO_SINK: str = "alsa_output.pci-0000_01_00.1.hdmi-stereo"
# Seconds to wait for the HDMI sink to appear after forcing the profile on.
ALARM_AUDIO_SINK_WAIT_SECONDS: float = 6.0
# Poll interval while waiting for the sink.
ALARM_AUDIO_SINK_POLL_SECONDS: float = 0.5
# TP-Link Tapo P110 smart-plug config file (JSON). # TP-Link Tapo P110 smart-plug config file (JSON).
# Create with mode 0600 and these keys: host, email, password. # Create with mode 0600 and these keys: host, email, password.
# Example contents: a JSON object mapping host -> "192.168.x.x", email -> # Example contents: a JSON object mapping host -> "192.168.x.x", email ->

View File

@ -1,18 +1,19 @@
#!/bin/bash #!/bin/bash
# systemd-sleep hook: restart wake-alarm.service after resume from hibernate. # systemd-sleep hook: start the unified morning routine after resume.
# #
# Installed to /usr/lib/systemd/system-sleep/wake-alarm.sh by install.sh. # Installed to /usr/lib/systemd/system-sleep/wake-alarm.sh by install.sh.
# #
# When the PC hibernates (rtcwake -m disk) and resumes the next morning, # When the PC hibernates (rtcwake -m disk) and resumes the next morning, the
# the user session is restored but wake-alarm.service is in a stopped state # user session is restored but no morning service is running. This hook starts
# (it ran at login the previous evening and exited with Restart=no). # morning-routine.service, which runs the wake alarm first (it owns the
# This hook restarts it so the alarm fires on the correct alarm day. # fullscreen until dismissed) and then the workout screen lock - one coherent
# flow, with the two never fighting for the screen.
if [[ "$1" != "post" ]]; then if [[ "$1" != "post" ]]; then
exit 0 exit 0
fi fi
logger -t wake-alarm-hook "Woke from sleep (type=$2) — restarting wake-alarm.service for active sessions" logger -t wake-alarm-hook "Woke from sleep (type=$2) - starting morning-routine.service for active sessions"
# Start wake-alarm.service for every logged-in user that has a running session # Start wake-alarm.service for every logged-in user that has a running session
# bus. Works with systemd >= 219. # bus. Works with systemd >= 219.
@ -20,10 +21,10 @@ while IFS= read -r uid; do
runtime_dir="/run/user/$uid" runtime_dir="/run/user/$uid"
[[ -d "$runtime_dir" ]] || continue [[ -d "$runtime_dir" ]] || continue
username=$(id -nu "$uid" 2>/dev/null) || continue username=$(id -nu "$uid" 2>/dev/null) || continue
logger -t wake-alarm-hook "Starting wake-alarm.service for user $username (uid=$uid)" logger -t wake-alarm-hook "Starting morning-routine.service for user $username (uid=$uid)"
XDG_RUNTIME_DIR="$runtime_dir" \ XDG_RUNTIME_DIR="$runtime_dir" \
DBUS_SESSION_BUS_ADDRESS="unix:path=${runtime_dir}/bus" \ DBUS_SESSION_BUS_ADDRESS="unix:path=${runtime_dir}/bus" \
runuser -u "$username" -- \ runuser -u "$username" -- \
systemctl --user start wake-alarm.service 2>/dev/null \ systemctl --user start morning-routine.service 2>/dev/null \
|| logger -t wake-alarm-hook "Failed to start wake-alarm.service for $username (non-fatal)" || logger -t wake-alarm-hook "Failed to start morning-routine.service for $username (non-fatal)"
done < <(loginctl list-sessions --no-legend 2>/dev/null | awk '{print $2}' | sort -u) done < <(loginctl list-sessions --no-legend 2>/dev/null | awk '{print $2}' | sort -u)

View File

@ -14,22 +14,24 @@ if TYPE_CHECKING:
from collections.abc import Generator, Iterator from collections.abc import Generator, Iterator
from python_pkg.wake_alarm._alarm import ( from python_pkg.wake_alarm._alarm import (
_activate_alarm_audio,
_alarm_sink_present,
_beep_loud, _beep_loud,
_beep_medium, _beep_medium,
_beep_pcspkr, _beep_pcspkr,
_beep_soft, _beep_soft,
_current_default_sink,
_ensure_tone_wav, _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, _parse_args,
_play_on_extra_devices, _play_on_extra_devices,
_play_tone, _play_tone,
_restore_alarm_audio,
_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,
@ -955,149 +957,173 @@ class TestWarnIfNoRealSink:
_warn_if_no_real_sink() # must not raise _warn_if_no_real_sink() # must not raise
class TestMaxSinkVolume: class TestAlarmSinkPresent:
"""Tests for _max_sink_volume and _restore_sink_volume.""" """Tests for _alarm_sink_present."""
def test_true_when_sink_listed(self) -> None:
"""Returns True when the alarm sink name appears in pactl output."""
from python_pkg.wake_alarm._constants import ALARM_AUDIO_SINK
proc = MagicMock(stdout=ALARM_AUDIO_SINK.encode() + b"\tPipeWire\n")
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=proc,
):
assert _alarm_sink_present("/usr/bin/pactl") is True
def test_false_when_sink_absent(self) -> None:
"""Returns False when the alarm sink is not in pactl output."""
proc = MagicMock(stdout=b"auto_null\tPipeWire\n")
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=proc,
):
assert _alarm_sink_present("/usr/bin/pactl") is False
def test_false_on_subprocess_error(self) -> None:
"""OSError while listing sinks → False, no raise."""
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=OSError("boom"),
):
assert _alarm_sink_present("/usr/bin/pactl") is False
class TestCurrentDefaultSink:
"""Tests for _current_default_sink."""
def test_returns_sink_name(self) -> None:
"""Returns the trimmed default sink name."""
proc = MagicMock(stdout=b"jbl_sink\n")
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=proc,
):
assert _current_default_sink("/usr/bin/pactl") == "jbl_sink"
def test_returns_none_when_empty(self) -> None:
"""Empty output → None."""
proc = MagicMock(stdout=b"\n")
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
return_value=proc,
):
assert _current_default_sink("/usr/bin/pactl") is None
def test_returns_none_on_error(self) -> None:
"""TimeoutExpired → None, no raise."""
with patch(
"python_pkg.wake_alarm._alarm.subprocess.run",
side_effect=subprocess.TimeoutExpired("pactl", 3),
):
assert _current_default_sink("/usr/bin/pactl") is None
class TestActivateAlarmAudio:
"""Tests for _activate_alarm_audio."""
def test_returns_none_when_pactl_missing(self) -> None: def test_returns_none_when_pactl_missing(self) -> None:
"""No pactl on PATH → returns None, logs warning.""" """No pactl on PATH → returns None without touching audio."""
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 ( with (
patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None), patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run, patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
): ):
_restore_sink_volume(("sink", "42%", False)) assert _activate_alarm_audio() is None
mock_run.assert_not_called() mock_run.assert_not_called()
def test_restores_volume_and_mute(self) -> None: def test_activates_and_returns_old_default(self) -> None:
"""Calls set-sink-volume and set-sink-mute with captured values.""" """Sink present → routes audio there and returns prior default sink."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm._alarm_sink_present",
return_value=True,
),
patch(
"python_pkg.wake_alarm._alarm._current_default_sink",
return_value="jbl_sink",
),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
):
result = _activate_alarm_audio()
assert result == "jbl_sink"
cmds = [call.args[0] for call in mock_run.call_args_list]
from python_pkg.wake_alarm._constants import (
ALARM_AUDIO_CARD,
ALARM_AUDIO_PROFILE,
ALARM_AUDIO_SINK,
)
assert [
"/usr/bin/pactl",
"set-card-profile",
ALARM_AUDIO_CARD,
ALARM_AUDIO_PROFILE,
] in cmds
assert ["/usr/bin/pactl", "set-default-sink", ALARM_AUDIO_SINK] in cmds
def test_returns_none_when_sink_never_appears(self) -> None:
"""Sink never shows up → returns None after polling (no raise)."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm._alarm_sink_present",
return_value=False,
),
patch("python_pkg.wake_alarm._alarm.time.sleep") as mock_sleep,
patch("python_pkg.wake_alarm._alarm.subprocess.run"),
):
assert _activate_alarm_audio() is None
mock_sleep.assert_called()
def test_waits_then_succeeds(self) -> None:
"""Sink absent then present → sleeps once, then routes audio."""
with (
patch(
"python_pkg.wake_alarm._alarm.shutil.which",
return_value="/usr/bin/pactl",
),
patch(
"python_pkg.wake_alarm._alarm._alarm_sink_present",
side_effect=[False, True],
),
patch(
"python_pkg.wake_alarm._alarm._current_default_sink",
return_value="old",
),
patch("python_pkg.wake_alarm._alarm.time.sleep") as mock_sleep,
patch("python_pkg.wake_alarm._alarm.subprocess.run"),
):
assert _activate_alarm_audio() == "old"
mock_sleep.assert_called_once()
class TestRestoreAlarmAudio:
"""Tests for _restore_alarm_audio."""
def test_none_is_noop(self) -> None:
"""None default → does nothing, no pactl lookup."""
with patch("python_pkg.wake_alarm._alarm.shutil.which") as mock_which:
_restore_alarm_audio(None)
mock_which.assert_not_called()
def test_no_pactl_returns_silently(self) -> None:
"""Default present but pactl missing → no raise, no run."""
with (
patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
):
_restore_alarm_audio("jbl_sink")
mock_run.assert_not_called()
def test_restores_default_sink(self) -> None:
"""Calls set-default-sink with the captured prior default."""
with ( with (
patch( patch(
"python_pkg.wake_alarm._alarm.shutil.which", "python_pkg.wake_alarm._alarm.shutil.which",
@ -1105,24 +1131,9 @@ class TestRestoreSinkVolume:
), ),
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run, patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
): ):
_restore_sink_volume(("sink", "42%", True)) _restore_alarm_audio("jbl_sink")
cmds = [call.args[0] for call in mock_run.call_args_list] 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-default-sink", "jbl_sink"] 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: class TestParseArgs:

View File

@ -58,8 +58,8 @@ def _block_extra_devices() -> Generator[MagicMock]:
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._wake_display"),
patch("python_pkg.wake_alarm._alarm._warn_if_no_real_sink"), 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._activate_alarm_audio", return_value=None),
patch("python_pkg.wake_alarm._alarm._restore_sink_volume"), patch("python_pkg.wake_alarm._alarm._restore_alarm_audio"),
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"),
): ):
@ -160,39 +160,61 @@ class TestWakeAlarmDismiss:
assert alarm.dismissed is False assert alarm.dismissed is False
alarm._stop_beep.set() alarm._stop_beep.set()
def test_dismiss_window_expired( def test_skip_window_expired_keeps_alarm_running(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Window expiry saves state with no skip.""" """Skip-window expiry denies the skip but does NOT stop the alarm."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
with patch( with patch(
"python_pkg.wake_alarm._alarm.save_wake_state", "python_pkg.wake_alarm._alarm.save_wake_state",
) as mock_save: ) as mock_save:
alarm._on_dismiss_window_expired() alarm._on_skip_window_expired()
# Alarm stays active and audible; only the skip reward is gone.
assert alarm._skip_earnable is False
assert alarm._active is True
assert alarm.dismissed is False assert alarm.dismissed is False
mock_save.assert_called_once_with( assert not alarm._stop_beep.is_set()
dismissed_at=None, mock_save.assert_not_called()
skip_workout=False, alarm._info_label.configure.assert_called()
)
alarm._stop_beep.set() alarm._stop_beep.set()
def test_dismiss_window_expired_noop_if_not_active( def test_skip_window_expired_noop_if_not_active(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Expiry is a no-op if alarm is no longer active.""" """Expiry is a no-op if alarm is no longer active."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
alarm._active = False alarm._active = False
alarm._on_skip_window_expired()
# skip_earnable stays at its initial True (method returned early).
assert alarm._skip_earnable is True
alarm._stop_beep.set()
def test_dismiss_after_skip_window_earns_no_skip(
self,
mock_tk_module: MagicMock,
) -> None:
"""Typing the code after the skip window stops the alarm w/o a skip."""
alarm = WakeAlarm(demo_mode=True)
alarm._skip_earnable = False
code = alarm._current_code
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = code
with patch( with patch(
"python_pkg.wake_alarm._alarm.save_wake_state", "python_pkg.wake_alarm._alarm.save_wake_state",
) as mock_save: ) as mock_save:
alarm._on_dismiss_window_expired() alarm._on_submit()
mock_save.assert_not_called() assert alarm.dismissed is True
assert mock_save.call_args[1]["skip_workout"] is False
alarm._stop_beep.set() alarm._stop_beep.set()
@ -312,14 +334,15 @@ class TestBeepLoop:
alarm._stop_beep.set() alarm._stop_beep.set()
class TestCloseAndFallback: class TestClose:
"""Tests for close and fallback scheduling.""" """Tests for the alarm close path."""
def test_close_stops_beep_and_destroys( def test_close_stops_beep_and_destroys(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""_close sets stop event and destroys root.""" """_close sets stop event and destroys root."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
alarm._close() alarm._close()
assert alarm._stop_beep.is_set() assert alarm._stop_beep.is_set()
@ -330,32 +353,26 @@ class TestCloseAndFallback:
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""_close calls _restore_fans with the saved fan state.""" """_close calls _restore_fans with the saved fan state."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
alarm._fan_state = True alarm._fan_state = True
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(active=True) mock_restore.assert_called_once_with(active=True)
def test_close_and_schedule_fallback( def test_close_restores_audio(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""_close_and_schedule_fallback destroys root.""" """_close restores the default sink captured at activation."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
alarm._close_and_schedule_fallback() alarm._audio_restore = "jbl_sink"
alarm.root.destroy.assert_called() with patch(
alarm._stop_beep.set() "python_pkg.wake_alarm._alarm._restore_alarm_audio",
) as mock_restore:
def test_close_and_schedule_fallback_restores_fans( alarm._close()
self, mock_restore.assert_called_once_with("jbl_sink")
mock_tk_module: MagicMock,
) -> None:
"""_close_and_schedule_fallback calls _restore_fans with the saved state."""
alarm = WakeAlarm(demo_mode=True)
alarm._fan_state = True
with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore:
alarm._close_and_schedule_fallback()
mock_restore.assert_called_once_with(active=True)
alarm._stop_beep.set() alarm._stop_beep.set()
@ -442,25 +459,22 @@ class TestDismissWithoutSkip:
alarm._stop_beep.set() alarm._stop_beep.set()
class TestDismissWindowExpiredWidgets: class TestSkipWindowExpiredMessage:
"""Tests for widget cleanup during dismiss window expiry.""" """Tests for the on-screen message when the skip window expires."""
def test_expired_creates_label( def test_expired_updates_status_label(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Expiry creates a 'Too late' label and destroys children.""" """Expiry updates the status label instead of closing the alarm."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
mock_widget = MagicMock()
alarm._container.winfo_children.return_value = [mock_widget]
with patch( alarm._on_skip_window_expired()
"python_pkg.wake_alarm._alarm.save_wake_state",
):
alarm._on_dismiss_window_expired()
mock_widget.destroy.assert_called_once() alarm._status_label.configure.assert_called_with(
mock_tk_module.Label.assert_called() text="No workout skip today.",
)
alarm._stop_beep.set() alarm._stop_beep.set()
@ -544,28 +558,45 @@ class TestRunMethod:
class TestUpdateTimerActive: class TestUpdateTimerActive:
"""Tests for timer update when alarm is active.""" """Tests for timer update when alarm is active."""
def test_update_timer_shows_remaining( def test_update_timer_shows_skip_window(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Timer update shows remaining time when not dismissed.""" """While the skip is earnable, the timer shows the skip-window count."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
alarm._update_timer() alarm._update_timer()
alarm._timer_label.configure.assert_called() text = alarm._timer_label.configure.call_args[1]["text"]
assert text.startswith("Skip window:")
alarm._stop_beep.set() alarm._stop_beep.set()
def test_update_timer_stops_at_zero( def test_update_timer_shows_prompt_after_window(
self, self,
mock_tk_module: MagicMock, mock_tk_module: MagicMock,
) -> None: ) -> None:
"""Timer stops scheduling when remaining time reaches zero.""" """After the window the timer shows the silence prompt and keeps going."""
import time as time_mod import time as time_mod
alarm = WakeAlarm(demo_mode=True) alarm = WakeAlarm(demo_mode=True)
# Set alarm start far in the past so remaining = 0 # Far in the past so remaining == 0 -> the else branch.
alarm._alarm_start = time_mod.monotonic() - 60 * 60 alarm._alarm_start = time_mod.monotonic() - 60 * 60
alarm.root.after.reset_mock()
alarm._update_timer() alarm._update_timer()
# root.after should NOT be called for re-scheduling text = alarm._timer_label.configure.call_args[1]["text"]
# (configure is still called to show 00:00) assert "type the code" in text
alarm._timer_label.configure.assert_called() # The alarm keeps nagging: it always reschedules while active.
alarm.root.after.assert_called_once()
alarm._stop_beep.set()
def test_update_timer_noop_when_not_active(
self,
mock_tk_module: MagicMock,
) -> None:
"""Timer update is a no-op once the alarm is no longer active."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._active = False
alarm._timer_label.configure.reset_mock()
alarm._update_timer()
alarm._timer_label.configure.assert_not_called()
alarm._stop_beep.set() alarm._stop_beep.set()

View File

@ -4,6 +4,13 @@ After=graphical-session.target
[Service] [Service]
Type=simple Type=simple
# DISPLAY/PYTHONPATH mirror workout-locker.service: without DISPLAY the alarm
# crashes on cold boot with "no display name and no $DISPLAY" (Tk can't open
# the X server) before systemd retries. The short sleep lets the X session
# export DISPLAY into the user environment first.
Environment=DISPLAY=:0
Environment=PYTHONPATH=%h/testsAndMisc
ExecStartPre=/bin/sleep 1
ExecStart=/usr/bin/python -m python_pkg.wake_alarm._alarm --production ExecStart=/usr/bin/python -m python_pkg.wake_alarm._alarm --production
WorkingDirectory=%h/testsAndMisc WorkingDirectory=%h/testsAndMisc
Restart=on-failure Restart=on-failure

View File

@ -1,6 +1,6 @@
{ {
"date": "2026-05-23", "date": "2026-05-25",
"dismissed_at": null, "dismissed_at": "2026-05-25T10:33:09.098156+00:00",
"skip_workout": false, "skip_workout": true,
"hmac": "17ae4173304d5f4b7c2376df2326162ae3bc4dfcf3a38ccb6f0647c81f61b5fa" "hmac": "49ae99880405c6e3f0b4948b07d398980e223a91d33dc7d0c7f0f9254463fa92"
} }