From 8db7b8eb2a9e7c21f7fc34040375da4411c26a47 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 25 May 2026 18:55:27 +0200 Subject: [PATCH] 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 --- wake_alarm/_alarm.py | 227 +++++++++---------- wake_alarm/_constants.py | 14 ++ wake_alarm/sleep-hook.sh | 19 +- wake_alarm/tests/test_alarm.py | 319 ++++++++++++++------------- wake_alarm/tests/test_alarm_part2.py | 133 ++++++----- wake_alarm/wake-alarm.service | 7 + wake_alarm/wake_state.json | 8 +- 7 files changed, 396 insertions(+), 331 deletions(-) diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index dbd7833..a305cc6 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -27,6 +27,11 @@ import tkinter as tk import wave 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, DISMISS_CODE_LENGTH, DISMISS_CODE_REFRESH_SECONDS, @@ -461,12 +466,13 @@ class WakeAlarm: self.root.update_idletasks() self._current_code = _generate_code() + self._skip_earnable: bool = True self._build_ui() self._schedule_code_refresh() - self._schedule_dismiss_window_close() + self._schedule_skip_window_close() self._start_beep_thread() 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._start_screen_flash() @@ -535,7 +541,7 @@ class WakeAlarm: """Handle code submission.""" entered = self._entry.get().strip() if entered == self._current_code: - self._dismiss_alarm(earned_skip=True) + self._dismiss_alarm(earned_skip=self._skip_earnable) else: self._status_label.configure(text="Wrong code! Try again.") self._entry.delete(0, tk.END) @@ -572,7 +578,7 @@ class WakeAlarm: """Close the alarm window.""" self._stop_beep.set() _restore_fans(active=self._fan_state) - _restore_sink_volume(self._sink_volume_state) + _restore_alarm_audio(self._audio_restore) _restore_display() turn_off_plug() self.root.destroy() @@ -587,55 +593,45 @@ class WakeAlarm: ms = DISMISS_CODE_REFRESH_SECONDS * 1000 if not self.demo_mode else 10_000 self.root.after(ms, self._schedule_code_refresh) - def _schedule_dismiss_window_close(self) -> None: - """Close dismiss window after the allowed time.""" + def _schedule_skip_window_close(self) -> None: + """Mark the workout-skip reward as expired after the allowed time.""" ms = DISMISS_WINDOW_MINUTES * 60 * 1000 if not self.demo_mode else 30_000 - self.root.after(ms, self._on_dismiss_window_expired) + self.root.after(ms, self._on_skip_window_expired) - def _on_dismiss_window_expired(self) -> None: - """Called when the dismiss window expires without valid dismissal.""" + def _on_skip_window_expired(self) -> None: + """Skip window closed: keep the alarm running, deny the workout skip. + + The alarm intentionally does NOT stop here - it keeps beeping and + flashing until the user actually types the code. Only the workout-skip + reward expires; dismissing now silences the alarm without earning a skip. + """ if not self._active: return - self._active = False - self._stop_beep.set() - save_wake_state(dismissed_at=None, skip_workout=False) - _logger.info("Dismiss window expired — no workout skip.") - - for widget in self._container.winfo_children(): - 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() + self._skip_earnable = False + self._info_label.configure( + text="Skip window closed - type the code to stop the alarm", + ) + self._status_label.configure(text="No workout skip today.") + _logger.info("Skip window expired - alarm continues until dismissed.") 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: return elapsed = time.monotonic() - self._alarm_start window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30 remaining = max(0, window - elapsed) - minutes = int(remaining) // 60 - seconds = int(remaining) % 60 - self._timer_label.configure( - text=f"Time remaining: {minutes:02d}:{seconds:02d}", - ) - if remaining > 0: - self.root.after(1000, self._update_timer) + if self._skip_earnable and remaining > 0: + minutes = int(remaining) // 60 + seconds = int(remaining) % 60 + self._timer_label.configure( + text=f"Skip window: {minutes:02d}:{seconds:02d}", + ) + 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: """Start the background beep escalation thread.""" @@ -697,94 +693,99 @@ def _pactl_path() -> str | None: 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 +def _alarm_sink_present(pactl: str) -> bool: + """Return True when the dedicated alarm HDMI sink exists in PipeWire.""" 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"], 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) + _logger.warning("pactl get-default-sink 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 + name = result.stdout.decode(errors="replace").strip() + return name or None + + +def _activate_alarm_audio() -> str | None: + """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 - muted = b"yes" in mute_proc.stdout - try: - subprocess.run( - [pactl, "set-sink-mute", sink, "0"], - capture_output=True, - timeout=3, - check=False, + time.sleep(ALARM_AUDIO_SINK_POLL_SECONDS) + else: + _logger.warning( + "Alarm audio sink %s did not appear after %.0fs; alarm may be silent", + ALARM_AUDIO_SINK, + ALARM_AUDIO_SINK_WAIT_SECONDS, ) - 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) + old_default = _current_default_sink(pactl) + 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: - """Restore the sink volume + mute captured by :func:`_max_sink_volume`.""" - if state is None: +def _restore_alarm_audio(old_default: str | None) -> None: + """Restore the default sink captured by :func:`_activate_alarm_audio`.""" + if old_default 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) + subprocess.run( + [pactl, "set-default-sink", old_default], + capture_output=True, + timeout=3, + check=False, + ) def _warn_if_no_real_sink() -> None: diff --git a/wake_alarm/_constants.py b/wake_alarm/_constants.py index c85763c..5cb51ab 100644 --- a/wake_alarm/_constants.py +++ b/wake_alarm/_constants.py @@ -38,6 +38,20 @@ WAKE_STATE_FILE: Path = Path(__file__).resolve().parent / "wake_state.json" # rtcwake binary path 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). # Create with mode 0600 and these keys: host, email, password. # Example contents: a JSON object mapping host -> "192.168.x.x", email -> diff --git a/wake_alarm/sleep-hook.sh b/wake_alarm/sleep-hook.sh index 29fd334..1cd90d3 100755 --- a/wake_alarm/sleep-hook.sh +++ b/wake_alarm/sleep-hook.sh @@ -1,18 +1,19 @@ #!/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. # -# When the PC hibernates (rtcwake -m disk) and resumes the next morning, -# the user session is restored but wake-alarm.service is in a stopped state -# (it ran at login the previous evening and exited with Restart=no). -# This hook restarts it so the alarm fires on the correct alarm day. +# When the PC hibernates (rtcwake -m disk) and resumes the next morning, the +# user session is restored but no morning service is running. This hook starts +# morning-routine.service, which runs the wake alarm first (it owns the +# fullscreen until dismissed) and then the workout screen lock - one coherent +# flow, with the two never fighting for the screen. if [[ "$1" != "post" ]]; then exit 0 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 # bus. Works with systemd >= 219. @@ -20,10 +21,10 @@ while IFS= read -r uid; do runtime_dir="/run/user/$uid" [[ -d "$runtime_dir" ]] || 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" \ DBUS_SESSION_BUS_ADDRESS="unix:path=${runtime_dir}/bus" \ runuser -u "$username" -- \ - systemctl --user start wake-alarm.service 2>/dev/null \ - || logger -t wake-alarm-hook "Failed to start wake-alarm.service for $username (non-fatal)" + systemctl --user start morning-routine.service 2>/dev/null \ + || 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) diff --git a/wake_alarm/tests/test_alarm.py b/wake_alarm/tests/test_alarm.py index b4567bf..7b99761 100644 --- a/wake_alarm/tests/test_alarm.py +++ b/wake_alarm/tests/test_alarm.py @@ -14,22 +14,24 @@ if TYPE_CHECKING: from collections.abc import Generator, Iterator from python_pkg.wake_alarm._alarm import ( + _activate_alarm_audio, + _alarm_sink_present, _beep_loud, _beep_medium, _beep_pcspkr, _beep_soft, + _current_default_sink, _ensure_tone_wav, _find_fan_hwmon, _generate_code, _is_alarm_day, _max_fans, - _max_sink_volume, _parse_args, _play_on_extra_devices, _play_tone, + _restore_alarm_audio, _restore_display, _restore_fans, - _restore_sink_volume, _set_max_brightness, _should_run_alarm, _speaker_test_path, @@ -955,149 +957,173 @@ class TestWarnIfNoRealSink: _warn_if_no_real_sink() # must not raise -class TestMaxSinkVolume: - """Tests for _max_sink_volume and _restore_sink_volume.""" +class TestAlarmSinkPresent: + """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: - """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.""" + """No pactl on PATH → returns None without touching audio.""" 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)) + assert _activate_alarm_audio() is None mock_run.assert_not_called() - def test_restores_volume_and_mute(self) -> None: - """Calls set-sink-volume and set-sink-mute with captured values.""" + def test_activates_and_returns_old_default(self) -> None: + """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 ( patch( "python_pkg.wake_alarm._alarm.shutil.which", @@ -1105,24 +1131,9 @@ class TestRestoreSinkVolume: ), 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] - 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 + assert ["/usr/bin/pactl", "set-default-sink", "jbl_sink"] in cmds class TestParseArgs: diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index 725e354..e525f15 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -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._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._activate_alarm_audio", return_value=None), + 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_off_plug"), ): @@ -160,39 +160,61 @@ class TestWakeAlarmDismiss: assert alarm.dismissed is False alarm._stop_beep.set() - def test_dismiss_window_expired( + def test_skip_window_expired_keeps_alarm_running( self, mock_tk_module: MagicMock, ) -> 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) with patch( "python_pkg.wake_alarm._alarm.save_wake_state", ) 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 - mock_save.assert_called_once_with( - dismissed_at=None, - skip_workout=False, - ) + assert not alarm._stop_beep.is_set() + mock_save.assert_not_called() + alarm._info_label.configure.assert_called() alarm._stop_beep.set() - def test_dismiss_window_expired_noop_if_not_active( + def test_skip_window_expired_noop_if_not_active( self, mock_tk_module: MagicMock, ) -> None: """Expiry is a no-op if alarm is no longer active.""" + del mock_tk_module alarm = WakeAlarm(demo_mode=True) alarm._active = False + alarm._on_skip_window_expired() + + # skip_earnable stays at its initial True (method returned early). + assert alarm._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( "python_pkg.wake_alarm._alarm.save_wake_state", ) 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() @@ -312,14 +334,15 @@ class TestBeepLoop: alarm._stop_beep.set() -class TestCloseAndFallback: - """Tests for close and fallback scheduling.""" +class TestClose: + """Tests for the alarm close path.""" def test_close_stops_beep_and_destroys( self, mock_tk_module: MagicMock, ) -> None: """_close sets stop event and destroys root.""" + del mock_tk_module alarm = WakeAlarm(demo_mode=True) alarm._close() assert alarm._stop_beep.is_set() @@ -330,32 +353,26 @@ class TestCloseAndFallback: mock_tk_module: MagicMock, ) -> None: """_close calls _restore_fans with the saved fan state.""" + del mock_tk_module alarm = WakeAlarm(demo_mode=True) alarm._fan_state = True with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore: alarm._close() mock_restore.assert_called_once_with(active=True) - def test_close_and_schedule_fallback( + def test_close_restores_audio( self, mock_tk_module: MagicMock, ) -> 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._close_and_schedule_fallback() - alarm.root.destroy.assert_called() - alarm._stop_beep.set() - - def test_close_and_schedule_fallback_restores_fans( - self, - 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._audio_restore = "jbl_sink" + with patch( + "python_pkg.wake_alarm._alarm._restore_alarm_audio", + ) as mock_restore: + alarm._close() + mock_restore.assert_called_once_with("jbl_sink") alarm._stop_beep.set() @@ -442,25 +459,22 @@ class TestDismissWithoutSkip: alarm._stop_beep.set() -class TestDismissWindowExpiredWidgets: - """Tests for widget cleanup during dismiss window expiry.""" +class TestSkipWindowExpiredMessage: + """Tests for the on-screen message when the skip window expires.""" - def test_expired_creates_label( + def test_expired_updates_status_label( self, mock_tk_module: MagicMock, ) -> 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) - mock_widget = MagicMock() - alarm._container.winfo_children.return_value = [mock_widget] - with patch( - "python_pkg.wake_alarm._alarm.save_wake_state", - ): - alarm._on_dismiss_window_expired() + alarm._on_skip_window_expired() - mock_widget.destroy.assert_called_once() - mock_tk_module.Label.assert_called() + alarm._status_label.configure.assert_called_with( + text="No workout skip today.", + ) alarm._stop_beep.set() @@ -544,28 +558,45 @@ class TestRunMethod: class TestUpdateTimerActive: """Tests for timer update when alarm is active.""" - def test_update_timer_shows_remaining( + def test_update_timer_shows_skip_window( self, mock_tk_module: MagicMock, ) -> 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._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() - def test_update_timer_stops_at_zero( + def test_update_timer_shows_prompt_after_window( self, mock_tk_module: MagicMock, ) -> 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 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.root.after.reset_mock() alarm._update_timer() - # root.after should NOT be called for re-scheduling - # (configure is still called to show 00:00) - alarm._timer_label.configure.assert_called() + text = alarm._timer_label.configure.call_args[1]["text"] + assert "type the code" in text + # 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() diff --git a/wake_alarm/wake-alarm.service b/wake_alarm/wake-alarm.service index 790fb44..fe1c045 100644 --- a/wake_alarm/wake-alarm.service +++ b/wake_alarm/wake-alarm.service @@ -4,6 +4,13 @@ After=graphical-session.target [Service] 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 WorkingDirectory=%h/testsAndMisc Restart=on-failure diff --git a/wake_alarm/wake_state.json b/wake_alarm/wake_state.json index ff8a0ad..94c00e5 100644 --- a/wake_alarm/wake_state.json +++ b/wake_alarm/wake_state.json @@ -1,6 +1,6 @@ { - "date": "2026-05-23", - "dismissed_at": null, - "skip_workout": false, - "hmac": "17ae4173304d5f4b7c2376df2326162ae3bc4dfcf3a38ccb6f0647c81f61b5fa" + "date": "2026-05-25", + "dismissed_at": "2026-05-25T10:33:09.098156+00:00", + "skip_workout": true, + "hmac": "49ae99880405c6e3f0b4948b07d398980e223a91d33dc7d0c7f0f9254463fa92" }