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

Migrate the alarm to gatelock v0.2.0 and give it an escape hatch

Browse Source 
Per-output locking: the alarm now paints one full challenge per live
monitor instead of one window on the primary. `_ViewGroup` presents the N
copies as one -- every label update fans out, and because all entries share
a single root-mastered StringVar, answering on whichever screen you are
actually looking at dismisses the alarm everywhere.

The escape hatch is the reason this repo could not stay unmigrated. The
alarm was the only locker with no way out but four correct rounds against a
30-second refresh, which is a lot to ask of someone who has just woken up
and nothing at all to fall back on if the alarm misfires at the wrong hour.
It is deliberately exactly as hard as sick mode: hidden for the first five
minutes so it is never the easy first move, gone entirely once a rolling
budget is spent, and costing a written justification plus a wait that
doubles with each recent use. It grants no workout skip.

`EscapeHatch` owns the whole flow so open and close stay symmetric. That
symmetry is the fix for a real bug: `_open_escape` used to pack_forget the
challenge widgets and cancel only flipped a bool, so "Back to the alarm"
left a dead form on a blank screen, leaked the dialog frame, and made the
next open sweep the dialog instead of the challenge. pack_forget does not
remember its options, so the challenge now lives in one `challenge_frame`
that is forgotten and re-packed as a unit.

Tests: the single-output default made every "on every monitor" assertion
degenerate into 1 == 1, and a shared MagicMock return_value made N surfaces
N references to one widget -- a loop cropped to containers[:1] passed the
whole suite. A `dual_output` fixture plus distinct-identity mocks now cover
all five fan-out loops; each was mutation-checked by cropping it and
confirming the suite goes red. The suite also no longer reads the real
escape history or signs with the production HMAC key.

326 tests, 100% branch coverage, pre-commit clean, no lint suppressions.
This commit is contained in:
Krzysztof Rudnicki 2026-07-25 15:35:44 +02:00
parent 191fa48b9d
commit c98055f8a2
9 changed files with 1261 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
wake_alarm/_alarm.py
View File

@ -10,7 +10,7 @@ from __future__ import annotations
import argparse
import contextlib
from dataclasses import dataclass
from dataclasses import dataclass, field
from datetime import datetime, timezone
import logging
import sys
@ -18,10 +18,17 @@ import threading
import time
import tkinter as tk
from gatelock import GateRoot, LockConfig, LockWindow
from gatelock import (
RANK_WAKE_ALARM,
Arbiter,
GateRoot,
LockConfig,
LockWindow,
SurfaceInfo,
)
from wake_alarm._alarm_display import _restore_display, _wake_display
from wake_alarm._alarm_view import _build_alarm_view
from wake_alarm._alarm_view import _build_alarm_view, _ViewGroup
from wake_alarm._audio import (
_beep_loud,
_beep_medium,
@ -47,6 +54,7 @@ from wake_alarm._constants import (
PHASE_SOFT_END,
SOFT_BEEP_INTERVAL,
)
from wake_alarm._escape_flow import EscapeHatch, build_tracker
from wake_alarm._logging_setup import configure_logging
from wake_alarm._sinks import (
_activate_all_outputs,
@ -74,6 +82,7 @@ class _AlarmProgress:
"""Mutable dismiss-challenge progress state."""
current_challenge: _Challenge
started_at: float = field(default_factory=time.monotonic)
skip_earnable: bool = True
rounds_completed: int = 0
flash_remaining: int = 0
@ -100,22 +109,42 @@ class WakeAlarm:
self.demo_mode = demo_mode
self.dismissed = False
self._stop_beep = threading.Event()
self._alarm_start: float = time.monotonic()
self._active = True
self.root = GateRoot()
self.root.on_callback_error = self.on_callback_error
self.root.title("Wake Alarm" + (" [DEMO]" if demo_mode else ""))
# mode="soft": no watchdog exists yet for a clean, silent lock
# failure, so hard-lock (overrideredirect + grab) is deferred.
config = LockConfig(mode="soft")
self._lock = LockWindow(self.root, config, hooks=self)
self._lock.setup()
self._progress = _AlarmProgress(current_challenge=_make_challenge())
self._view = _build_alarm_view(
self.root, self._progress.current_challenge, config, self._on_submit
# Production hard-locks as of gatelock v0.2.0. The old comment here
# deferred it for want of "a watchdog for a clean, silent lock
# failure" -- the recovery loop is that watchdog. Demo stays grab-free
# so a manual test cannot trap the developer, but takes
# overrideredirect anyway: per-output placement is impossible without
# it, since a window manager rewrites a managed window's geometry.
config = LockConfig(
mode="soft" if demo_mode else "hard",
overrideredirect=True,
app_name="wake_alarm",
rank=RANK_WAKE_ALARM,
)
# A local, not an attribute: the arbiter is published here and then
# owned by LockWindow, which is what releases it on close.
arbiter = Arbiter(
"wake_alarm",
RANK_WAKE_ALARM,
grab=config.resolved_grab(),
disable_vt=config.resolved_disable_vt(),
)
arbiter.publish()
arbiter.acquire_holder()
# Built before setup(): setup() calls build_surface for every live
# output, and those need the challenge and the shared entry variable.
self._progress = _AlarmProgress(current_challenge=_make_challenge())
self._view = _ViewGroup(config, tk.StringVar(master=self.root))
self._escape = EscapeHatch(build_tracker(), self._view, self._on_escape_granted)
self._lock = LockWindow(self.root, config, hooks=self, arbiter=arbiter)
self._lock.setup()
self._update_timer()
if self._progress.current_challenge.kind == "flash":
self._start_flash_countdown()
@ -128,8 +157,36 @@ class WakeAlarm:
self._start_screen_flash()
self._lock.grab_input()
def on_focus_ready(self) -> None:
"""Put keyboard focus on the dismiss-code entry once mapped."""
def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None:
"""Paint the full challenge on one monitor.
Called once per live output, and again whenever an output comes
back. Every copy shares one entry variable and one submit handler,
so answering on any monitor dismisses the alarm on all of them.
"""
view = _build_alarm_view(
parent,
self._progress.current_challenge,
self._view.config,
on_submit=self._on_submit,
entry_var=self._view.entry_var,
)
view.output_name = surface.output_name
view.escape_button.configure(command=self._escape.open)
self._view.add(view)
def teardown_surface(self, surface: SurfaceInfo) -> None:
"""Forget the view for an output that went dark."""
self._view.discard(surface.output_name)
def on_focus_ready(self, surface: SurfaceInfo | None) -> None:
"""Put keyboard focus on the dismiss-code entry once mapped.
``surface`` is None when no output is live -- the alarm is armed
and audible with nothing to show, so there is nothing to focus.
"""
if surface is None:
return
with contextlib.suppress(tk.TclError):
self._view.entry.focus_force()
@ -217,8 +274,9 @@ class WakeAlarm:
now_iso = datetime.now(tz=timezone.utc).isoformat()
save_wake_state(dismissed_at=now_iso, skip_workout=earned_skip)
for widget in self._view.container.winfo_children():
widget.destroy()
for container in self._view.containers:
for widget in container.winfo_children():
widget.destroy()
msg = (
"Workout skip earned! Enjoy your morning."
@ -228,13 +286,14 @@ class WakeAlarm:
cfg = self._view.config
color = cfg.success if earned_skip else cfg.warning
tk.Label(
self._view.container,
text=msg,
font=("Arial", 36, "bold"),
fg=color,
bg=cfg.bg,
).pack(pady=30)
for container in self._view.containers:
tk.Label(
container,
text=msg,
font=("Arial", 36, "bold"),
fg=color,
bg=cfg.bg,
).pack(pady=30)
self.root.after(3000, self._lock.close)
@ -283,7 +342,7 @@ class WakeAlarm:
"""Show the skip-window countdown, then a keep-going silence prompt."""
if not self._active:
return
elapsed = time.monotonic() - self._alarm_start
elapsed = time.monotonic() - self._progress.started_at
window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30
remaining = max(0, window - elapsed)
if self._progress.skip_earnable and remaining > 0:
@ -296,8 +355,14 @@ class WakeAlarm:
self._view.timer_label.configure(
text="No skip available - type the code to stop the alarm",
)
self._escape.maybe_offer(elapsed)
self.root.after(1000, self._update_timer)
def _on_escape_granted(self) -> None:
"""The hatch was earned: silence the alarm, grant no workout skip."""
_logger.warning("alarm dismissed via the escape hatch, no skip earned")
self._dismiss_alarm(earned_skip=False)
def _start_beep_thread(self) -> None:
"""Start the background beep escalation thread."""
thread = threading.Thread(target=self._beep_loop, daemon=True)
@ -312,14 +377,21 @@ class WakeAlarm:
if not self._active:
return
cfg = self._view.config
self.root.configure(bg=cfg.danger if self._progress.flash_on else cfg.bg)
colour = cfg.danger if self._progress.flash_on else cfg.bg
self.root.configure(bg=colour)
# The per-output surfaces sit on top of the backdrop, so flashing
# only the root would be invisible on every monitor.
for container in self._view.containers:
with contextlib.suppress(tk.TclError):
container.configure(bg=colour)
container.winfo_toplevel().configure(bg=colour)
self._progress.flash_on = not self._progress.flash_on
self.root.after(750, self._flash_step)
def _beep_loop(self) -> None:
"""Escalating beep loop running in background thread."""
while not self._stop_beep.is_set():
elapsed_minutes = (time.monotonic() - self._alarm_start) / 60.0
elapsed_minutes = (time.monotonic() - self._progress.started_at) / 60.0
if elapsed_minutes < PHASE_SOFT_END:
_play_on_all_sinks(440)

245
wake_alarm/_alarm_view.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import contextlib
from dataclasses import dataclass
import tkinter as tk
from typing import TYPE_CHECKING
@ -16,33 +17,183 @@ if TYPE_CHECKING:
from wake_alarm._challenges import _Challenge
@dataclass
class _AlarmLabels:
"""Every label on one monitor's challenge screen.
Grouped rather than spread across :class:`_AlarmView` so that fanning a
``configure`` call out by name (see :meth:`_ViewGroup._label`) has one
obvious namespace to look in.
"""
title: tk.Label
round_counter: tk.Label
info: tk.Label
code: tk.Label
status: tk.Label
timer: tk.Label
@dataclass
class _AlarmView:
"""The Tk widgets that make up the alarm's dismiss-challenge screen."""
"""The Tk widgets that make up the alarm's dismiss-challenge screen.
``container`` is placed and never unpacked; ``challenge_frame`` holds
everything the escape hatch has to take off the screen while its form is
up, so hiding and restoring the challenge is one call each way rather
than a per-widget ``pack_forget`` that cannot be undone.
"""
container: tk.Frame
title_label: tk.Label
round_label: tk.Label
info_label: tk.Label
code_label: tk.Label
challenge_frame: tk.Frame
labels: _AlarmLabels
entry: tk.Entry
status_label: tk.Label
timer_label: tk.Label
escape_button: tk.Button
config: LockConfig
output_name: str = ""
def _build_alarm_view(
root: tk.Tk,
class _LabelProxy:
"""Fans one ``configure`` call out to the same label on every monitor."""
def __init__(self, widgets: list[tk.Label]) -> None:
"""Wrap the per-monitor copies of one label."""
self._widgets = widgets
def configure(self, **kwargs: str) -> None:
"""Apply the same configuration to every copy."""
for widget in self._widgets:
with contextlib.suppress(tk.TclError):
widget.configure(**kwargs)
class _EntryProxy:
"""The dismiss-code entry, as seen across every monitor.
All copies share one root-mastered ``StringVar``, so typing on whichever
screen the user is actually looking at is the same text everywhere -- and
reading it needs no idea of which monitor that was.
"""
def __init__(self, widgets: list[tk.Entry], var: tk.StringVar) -> None:
"""Wrap the per-monitor entries and the variable behind them."""
self._widgets = widgets
self._var = var
def get(self) -> str:
"""Return the shared text, whichever monitor it was typed on."""
return self._var.get()
def delete(self, _first: object, _last: object = None) -> None:
"""Clear the shared text on every monitor."""
self._var.set("")
def focus_force(self) -> None:
"""Focus the first entry that still exists."""
for widget in self._widgets:
with contextlib.suppress(tk.TclError):
widget.focus_force()
return
class _ViewGroup:
"""Presents N per-monitor views as if they were one.
Every label update fans out to all monitors, and the challenge can be
answered on any of them -- solving it anywhere dismisses the alarm
everywhere, because every entry is the same variable and every submit
button is the same callable.
"""
def __init__(self, config: LockConfig, entry_var: tk.StringVar) -> None:
"""Start empty; surfaces are added as outputs come up."""
self.config = config
self._entry_var = entry_var
self._views: list[_AlarmView] = []
def add(self, view: _AlarmView) -> None:
"""Register a newly built per-monitor view."""
self._views.append(view)
def discard(self, output_name: str) -> None:
"""Forget the view for an output that went dark."""
self._views = [v for v in self._views if v.output_name != output_name]
@property
def surfaces(self) -> list[_AlarmView]:
"""Every per-monitor view, for callers needing more than one widget."""
return list(self._views)
@property
def containers(self) -> list[tk.Frame]:
"""Every per-monitor container frame."""
return [view.container for view in self._views]
@property
def entry_var(self) -> tk.StringVar:
"""The one variable every monitor's entry is bound to."""
return self._entry_var
@property
def entry(self) -> _EntryProxy:
"""The dismiss-code entry across all monitors."""
return _EntryProxy([v.entry for v in self._views], self._entry_var)
def _label(self, name: str) -> _LabelProxy:
"""Build a proxy over one named label on every monitor."""
return _LabelProxy([getattr(view.labels, name) for view in self._views])
@property
def title_label(self) -> _LabelProxy:
"""The "WAKE UP!" heading on every monitor."""
return self._label("title")
@property
def round_label(self) -> _LabelProxy:
"""The round counter on every monitor."""
return self._label("round_counter")
@property
def info_label(self) -> _LabelProxy:
"""The challenge hint on every monitor."""
return self._label("info")
@property
def code_label(self) -> _LabelProxy:
"""The challenge display on every monitor."""
return self._label("code")
@property
def status_label(self) -> _LabelProxy:
"""The status line on every monitor."""
return self._label("status")
@property
def timer_label(self) -> _LabelProxy:
"""The countdown on every monitor."""
return self._label("timer")
@property
def escape_buttons(self) -> list[tk.Button]:
"""The escape-hatch button on every monitor."""
return [view.escape_button for view in self._views]
def _build_challenge(
parent: tk.Misc,
challenge: _Challenge,
config: LockConfig,
on_submit: Callable[[object], None],
) -> _AlarmView:
"""Build the dismiss-challenge UI and return its widgets as a view."""
container = tk.Frame(root, bg=config.bg)
container.place(relx=0.5, rely=0.5, anchor="center")
entry_var: tk.StringVar,
) -> tuple[_AlarmLabels, tk.Entry]:
"""Fill ``parent`` with the challenge itself, top to bottom.
Split out from :func:`_build_alarm_view` so that neither function has to
hold the whole screen's worth of widgets at once; the pack order here is
the on-screen order, and the entry sits deliberately mid-stack.
"""
title_label = tk.Label(
container,
parent,
text="WAKE UP!",
font=("Arial", 48, "bold"),
fg=config.danger,
@ -51,7 +202,7 @@ def _build_alarm_view(
title_label.pack(pady=20)
round_label = tk.Label(
container,
parent,
text=f"Round 1 / {DISMISS_ROUNDS_REQUIRED}",
font=("Arial", 24, "bold"),
fg=config.warning,
@ -60,7 +211,7 @@ def _build_alarm_view(
round_label.pack(pady=5)
info_label = tk.Label(
container,
parent,
text=challenge.hint,
font=("Arial", 18),
fg=config.fg,
@ -73,7 +224,7 @@ def _build_alarm_view(
48 if challenge.kind in ("math", "sort", "sequence", "reverse") else 72
)
code_label = tk.Label(
container,
parent,
text=challenge.display,
font=("Courier", code_font_size, "bold"),
fg=config.success,
@ -82,17 +233,18 @@ def _build_alarm_view(
code_label.pack(pady=30)
entry = tk.Entry(
container,
parent,
font=("Courier", 36),
justify="center",
width=12,
textvariable=entry_var,
)
entry.pack(pady=10)
entry.focus_set()
entry.bind("<Return>", on_submit)
status_label = tk.Label(
container,
parent,
text="",
font=("Arial", 18),
fg=config.danger,
@ -101,7 +253,7 @@ def _build_alarm_view(
status_label.pack(pady=10)
timer_label = tk.Label(
container,
parent,
text="",
font=("Arial", 14),
fg=config.muted,
@ -109,14 +261,55 @@ def _build_alarm_view(
)
timer_label.pack(pady=5)
return (
_AlarmLabels(
title=title_label,
round_counter=round_label,
info=info_label,
code=code_label,
status=status_label,
timer=timer_label,
),
entry,
)
def _build_alarm_view(
root: tk.Misc,
challenge: _Challenge,
config: LockConfig,
*,
on_submit: Callable[[object], None],
entry_var: tk.StringVar,
) -> _AlarmView:
"""Build the dismiss-challenge UI and return its widgets as a view."""
container = tk.Frame(root, bg=config.bg)
container.place(relx=0.5, rely=0.5, anchor="center")
# One frame around the whole challenge: the escape hatch hides it wholesale
# and packs it straight back on cancel. Forgetting the widgets individually
# would be irreversible -- pack_forget does not remember its options.
challenge_frame = tk.Frame(container, bg=config.bg)
challenge_frame.pack()
labels, entry = _build_challenge(
challenge_frame, challenge, config, on_submit, entry_var
)
# Hidden until the alarm has been up long enough; see _escape_flow.
escape_button = tk.Button(
challenge_frame,
text="I can't do this right now",
font=(config.font_family, 14),
bg=config.field_bg,
fg=config.muted,
)
return _AlarmView(
container=container,
title_label=title_label,
round_label=round_label,
info_label=info_label,
code_label=code_label,
challenge_frame=challenge_frame,
labels=labels,
entry=entry,
status_label=status_label,
timer_label=timer_label,
escape_button=escape_button,
config=config,
)

9
wake_alarm/_constants.py
View File

@ -68,6 +68,15 @@ WORKOUT_LOG_FILE: Path = (
# hibernate fallback — keep the two readers in step if any of this changes.
# GitHub token config, key "token" (PAT with "repo" scope: the repo is private).
# The alarm's escape-hatch history. Lives beside the sync config rather
# than in the package dir, so a reinstall cannot silently reset the budget.
ESCAPE_HISTORY_FILE: Path = (
Path.home() / ".config" / "wake_alarm" / "escape_history.json"
)
# How long the alarm must have been up before the hatch is offered, so it
# is never the easy first move.
ESCAPE_OFFER_AFTER_SECONDS: int = 300
SYNC_CONFIG_FILE: Path = Path.home() / ".config" / "wake_alarm" / "sync.json"
# Fallback token locations, tried in order when SYNC_CONFIG_FILE is absent.

289
wake_alarm/_escape_flow.py Normal file
View File

@ -0,0 +1,289 @@
"""The alarm's sanctioned way out when the challenge genuinely isn't happening.
Until now the alarm had exactly one exit: four correct challenge rounds against
a 30-second refresh. That is a lot to ask of someone who has just woken up, and
it is *nothing* to fall back on if the alarm misfires at the wrong hour. Every
other locker here has a bounded escape -- screen-locker's sick mode,
diet-guard's log-a-meal -- so the alarm now has the same, built on the shared
:mod:`gatelock._escape` core rather than a private copy.
The shape, deliberately identical to sick mode:
* the button only appears after the alarm has been up a while, so it is never
the *easy* first move;
* it disappears entirely once any rolling budget is spent;
* it costs a written justification, with the last ten shown back to you;
* it costs a wait that doubles with each recent use;
* every entry is HMAC-signed, so editing the history is detectable.
"""
from __future__ import annotations
import contextlib
from dataclasses import dataclass
import logging
import tkinter as tk
from typing import TYPE_CHECKING
from gatelock import EscapeDraft, EscapePolicy, EscapeTracker
from wake_alarm._constants import ESCAPE_HISTORY_FILE, ESCAPE_OFFER_AFTER_SECONDS
if TYPE_CHECKING:
from collections.abc import Callable
from gatelock import LockConfig
from wake_alarm._alarm_view import _ViewGroup
_logger = logging.getLogger(__name__)
ALARM_ESCAPE_POLICY = EscapePolicy(
name="cant-wake",
label="Can't-wake",
)
"""Budgets left at the shared defaults, so this is exactly as hard as sick mode."""
def build_tracker() -> EscapeTracker:
"""Build the alarm's escape tracker over its own history file."""
tracker = EscapeTracker(ALARM_ESCAPE_POLICY, ESCAPE_HISTORY_FILE)
tracker.load()
return tracker
def is_offerable(tracker: EscapeTracker, elapsed_seconds: float) -> bool:
"""Whether the hatch should be shown at all.
Two independent gates: the alarm must have been up long enough that this
is not the first thing you reach for, and the budget must not be spent.
"""
if elapsed_seconds < ESCAPE_OFFER_AFTER_SECONDS:
return False
return not tracker.is_budget_exhausted()
@dataclass(frozen=True)
class EscapeCallbacks:
"""Who to tell how the form ended."""
on_granted: Callable[[], None]
on_cancel: Callable[[], None]
@dataclass
class _FormFields:
"""The four inputs the justification is read back out of."""
reason: tk.Entry
onset: tk.Entry
severity: tk.Entry
description: tk.Text
class EscapeDialog:
"""The justification form and its escalating wait."""
def __init__(
self,
parent: tk.Misc,
config: LockConfig,
tracker: EscapeTracker,
callbacks: EscapeCallbacks,
) -> None:
"""Build the form inside ``parent``."""
self._tracker = tracker
self._callbacks = callbacks
self._remaining = 0
self.frame = tk.Frame(parent, bg=config.bg)
self._status = tk.Label(
self.frame,
text=tracker.budget_summary(),
font=(config.font_family, 16),
fg=config.muted,
bg=config.bg,
)
self._status.pack(pady=6)
tk.Label(
self.frame,
text="Why can you not do this right now?",
font=(config.font_family, 22, "bold"),
fg=config.fg,
bg=config.bg,
).pack(pady=8)
tk.Label(
self.frame,
text=tracker.format_recent(),
font=("Courier", 11),
fg=config.muted,
bg=config.bg,
justify="left",
).pack(pady=6)
description = tk.Text(
self.frame,
height=4,
width=60,
wrap="word",
bg=config.field_bg,
fg=config.fg,
insertbackground=config.fg,
)
self._fields = _FormFields(
reason=self._field("What is wrong", config),
onset=self._field("When did it start", config),
severity=self._field("Severity 1-10", config),
description=description,
)
description.pack(pady=6)
self._complaint = tk.Label(
self.frame,
text="",
font=(config.font_family, 14),
fg=config.danger,
bg=config.bg,
)
self._complaint.pack(pady=4)
self._submit = tk.Button(
self.frame,
text="Submit",
font=(config.font_family, 16, "bold"),
bg=config.warning,
fg=config.on_fill,
command=self.submit,
)
self._submit.pack(pady=6)
tk.Button(
self.frame,
text="Back to the alarm",
font=(config.font_family, 14),
bg=config.field_bg,
fg=config.fg,
command=callbacks.on_cancel,
).pack(pady=4)
def _field(self, label: str, config: LockConfig) -> tk.Entry:
"""Add a labelled single-line field and return its entry."""
tk.Label(
self.frame,
text=label,
font=(config.font_family, 13),
fg=config.muted,
bg=config.bg,
).pack()
entry = tk.Entry(self.frame, width=40)
entry.pack(pady=2)
return entry
def _draft(self) -> EscapeDraft:
"""Read the form into a draft."""
try:
severity = int(self._fields.severity.get().strip() or "0")
except ValueError:
severity = 0
return EscapeDraft(
reason=self._fields.reason.get(),
onset=self._fields.onset.get(),
severity=severity,
description=self._fields.description.get("1.0", "end-1c"),
)
def submit(self) -> None:
"""Validate, then start the escalating wait."""
draft = self._draft()
complaint = self._tracker.validate(draft)
if complaint is not None:
self._complaint.configure(text=complaint)
return
if self._tracker.is_budget_exhausted():
# Re-checked at submit time: the budget could have been spent by
# another surface, or the day could have rolled over mid-form.
self._complaint.configure(text="No budget left. Solve the challenge.")
return
self._tracker.record(draft)
self._remaining = self._tracker.compute_lockout_seconds()
self._submit.configure(state="disabled")
self._tick()
def _tick(self) -> None:
"""Count the wait down, then grant."""
if self._remaining <= 0:
_logger.warning(
"alarm escape hatch used -- %s", self._tracker.budget_summary()
)
self._callbacks.on_granted()
return
self._complaint.configure(text=f"Wait {self._remaining}s...")
self._remaining -= 1
self.frame.after(1000, self._tick)
class EscapeHatch:
"""The alarm's way out, from the hidden button to putting the screen back.
The alarm delegates the whole flow here so that opening and closing the
form stay symmetric: every ``open`` that hides a challenge and builds a
dialog has a ``close`` that destroys the dialog and packs the challenge
back. Splitting those halves across two objects is what previously left
"Back to the alarm" showing a blank screen.
"""
def __init__(
self,
tracker: EscapeTracker,
views: _ViewGroup,
on_granted: Callable[[], None],
) -> None:
"""Bind the hatch to the surfaces it swaps and the alarm it releases."""
self._tracker = tracker
self._views = views
self._on_granted = on_granted
self._dialogs: list[EscapeDialog] = []
self.is_open = False
def maybe_offer(self, elapsed_seconds: float) -> None:
"""Reveal the button on every monitor once it is earned, and only then."""
if self.is_open:
return
if not is_offerable(self._tracker, elapsed_seconds):
return
for button in self._views.escape_buttons:
with contextlib.suppress(tk.TclError):
button.pack(pady=8)
def open(self) -> None:
"""Swap every surface over to the justification form."""
self.is_open = True
callbacks = EscapeCallbacks(
on_granted=self._on_granted,
on_cancel=self.close,
)
for view in self._views.surfaces:
with contextlib.suppress(tk.TclError):
view.challenge_frame.pack_forget()
dialog = EscapeDialog(
view.container,
self._views.config,
self._tracker,
callbacks,
)
dialog.frame.pack()
self._dialogs.append(dialog)
def close(self) -> None:
"""Back out of the form and put the challenge back on every monitor."""
for dialog in self._dialogs:
with contextlib.suppress(tk.TclError):
dialog.frame.destroy()
self._dialogs.clear()
for view in self._views.surfaces:
with contextlib.suppress(tk.TclError):
view.challenge_frame.pack()
self.is_open = False

108
wake_alarm/tests/conftest.py Normal file
View File

@ -0,0 +1,108 @@
"""Shared fixtures keeping the wake_alarm suite off the real machine.
Since gatelock v0.2.0 the lock builds one real ``tk.Toplevel`` per live
output. Two consequences for tests, both of which bite silently:
* ``tk.Toplevel`` over a ``MagicMock`` master sends real tkinter into infinite
mock recursion -- the run **hangs** rather than failing, which is far harder
to diagnose than an error.
* An unpatched lock setup shells out to ``setxkbmap`` and really disables VT
switching on the machine running the tests, and reaches the real
``$XDG_RUNTIME_DIR``, where it could stand a *production* locker down.
The escape hatch adds a third: ``WakeAlarm.__init__`` builds its tracker over
the real ``ESCAPE_HISTORY_FILE``, so an unpatched run both reads a *production*
budget -- making "the button appears" tests fail on any day the real budget is
spent -- and could write to it.
So this is autouse and deliberately broad.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from gatelock import Output, OutputRect
import pytest
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
FAKE_OUTPUTS = (
Output(
"VIRTUAL-1",
connected=True,
rect=OutputRect(0, 0, 1920, 1080),
primary=True,
),
)
"""One live output, so the Tk-screen fallback (which would call int() on a
MagicMock) is never reached."""
TWO_OUTPUTS = (
Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True),
Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)),
)
"""The real desk. Opt into it with the ``dual_output`` fixture for anything
asserting that work fans out: against ``FAKE_OUTPUTS`` a loop that only ever
touched the first surface would pass every assertion."""
def make_toplevel(_parent: object = None, **_kwargs: object) -> MagicMock:
"""A Toplevel stand-in that remembers the geometry it was given."""
window = MagicMock()
state = {"x": 0, "y": 0, "w": 0, "h": 0, "mapped": False}
def geometry(spec: str) -> None:
size, x, y = spec.split("+")
width, height = size.split("x")
state.update(x=int(x), y=int(y), w=int(width), h=int(height))
window.geometry.side_effect = geometry
window.deiconify.side_effect = lambda: state.update(mapped=True)
window.withdraw.side_effect = lambda: state.update(mapped=False)
window.winfo_ismapped.side_effect = lambda: state["mapped"]
window.winfo_rootx.side_effect = lambda: state["x"]
window.winfo_rooty.side_effect = lambda: state["y"]
window.winfo_width.side_effect = lambda: state["w"]
window.winfo_height.side_effect = lambda: state["h"]
return window
@pytest.fixture(autouse=True)
def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None]:
"""Block every path from the lock to the real display or machine."""
monkeypatch.setenv("GATELOCK_RUNTIME_DIR", str(tmp_path / "gatelock-runtime"))
key_file = tmp_path / "hmac.key"
key_file.write_bytes(b"0" * 32)
with (
# The real key is root-owned but world-readable, so an unpatched run
# signs test entries with the *production* key on this machine and
# with none at all anywhere else. Pin it so the suite is portable.
patch("gatelock.log_integrity.DEFAULT_HMAC_KEY_FILE", key_file),
patch(
"wake_alarm._escape_flow.ESCAPE_HISTORY_FILE",
tmp_path / "escape_history.json",
),
patch("gatelock._surfaces.tk.Toplevel", side_effect=make_toplevel),
patch("gatelock._outputs.RandrBackend.create", return_value=None),
patch("gatelock._outputs.scan_xrandr", return_value=FAKE_OUTPUTS),
patch("gatelock._detect._RandrEventSource.start", return_value=False),
patch("gatelock._vt.shutil.which", return_value="/usr/bin/setxkbmap"),
patch("gatelock._vt.subprocess.run"),
):
yield
@pytest.fixture
def dual_output(_hermetic: None) -> Generator[None]:
"""Re-scan as a two-monitor desk, layered over the single-output default.
Depends on ``_hermetic`` so pytest is obliged to apply it second; the
``scan_xrandr`` patch here has to be the one that wins.
"""
with patch("gatelock._outputs.scan_xrandr", return_value=TWO_OUTPUTS):
yield

52
wake_alarm/tests/test_alarm_part2.py
View File

@ -102,15 +102,12 @@ class TestWakeAlarmInit:
assert alarm.demo_mode is True
assert alarm.dismissed is False
mock_root = mock_tk_module.Tk.return_value
# LockConfig(mode="soft") never sets 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.kwargs.get("fullscreen") is True
]
assert fs_calls, "fullscreen attribute must be set"
# Since gatelock v0.2.0 even demo takes overrideredirect: a window
# manager rewrites a managed window's geometry, so per-output
# placement is impossible without it. attributes(fullscreen=True)
# is gone -- explicit per-output geometry replaces it.
mock_root.overrideredirect.assert_called_with(boolean=True)
assert alarm._view._views, "a surface must exist per live output"
alarm._stop_beep.set() # Stop beep thread
def test_production_mode_fullscreen(
@ -121,13 +118,11 @@ class TestWakeAlarmInit:
alarm = WakeAlarm(demo_mode=False)
assert alarm.demo_mode is False
mock_root = mock_tk_module.Tk.return_value
mock_root.overrideredirect.assert_not_called()
fs_calls = [
c
for c in mock_root.attributes.call_args_list
if c.kwargs.get("fullscreen") is True
]
assert fs_calls, "fullscreen attribute must be set"
mock_root.overrideredirect.assert_called_with(boolean=True)
# attributes(fullscreen=True) is deliberately gone: it is an EWMH
# request an override-redirect window is invisible to, and where
# it applies it snaps to one monitor or the whole bounding box.
assert alarm._view._views, "a surface must exist per live output"
alarm._stop_beep.set()
@ -142,13 +137,14 @@ class TestWakeAlarmDismiss:
from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED
alarm = WakeAlarm(demo_mode=True)
mock_entry = mock_tk_module.Entry.return_value
with patch(
"wake_alarm._alarm.save_wake_state",
) as mock_save:
for _ in range(DISMISS_ROUNDS_REQUIRED):
mock_entry.get.return_value = alarm._progress.current_challenge.answer
alarm._view.entry_var.get.return_value = (
alarm._progress.current_challenge.answer
)
alarm._on_submit()
assert alarm.dismissed is True
@ -162,8 +158,9 @@ class TestWakeAlarmDismiss:
) -> None:
"""A single correct entry is not enough — DISMISS_ROUNDS_REQUIRED is 2+."""
alarm = WakeAlarm(demo_mode=True)
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = alarm._progress.current_challenge.answer
alarm._view.entry_var.get.return_value = (
alarm._progress.current_challenge.answer
)
alarm._on_submit()
@ -179,8 +176,9 @@ class TestWakeAlarmDismiss:
from wake_alarm._challenges import _Challenge
alarm = WakeAlarm(demo_mode=True)
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = alarm._progress.current_challenge.answer
alarm._view.entry_var.get.return_value = (
alarm._progress.current_challenge.answer
)
next_math = _Challenge(kind="math", display="2 + 2 = ?", answer="4", hint="x")
with patch("wake_alarm._alarm._make_challenge", return_value=next_math):
alarm._on_submit()
@ -201,8 +199,7 @@ class TestWakeAlarmDismiss:
alarm._progress.current_challenge = _Challenge(
kind="math", display="2 + 2 = ?", answer="4", hint="test"
)
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = "99"
alarm._view.entry_var.get.return_value = "99"
alarm._on_submit()
@ -228,7 +225,7 @@ class TestWakeAlarmDismiss:
assert alarm.dismissed is False
assert not alarm._stop_beep.is_set()
mock_save.assert_not_called()
alarm._view.info_label.configure.assert_called()
alarm._view._views[0].labels.info.configure.assert_called()
alarm._stop_beep.set()
def test_skip_window_expired_noop_if_not_active(
@ -255,13 +252,14 @@ class TestWakeAlarmDismiss:
alarm = WakeAlarm(demo_mode=True)
alarm._progress.skip_earnable = False
mock_entry = mock_tk_module.Entry.return_value
with patch(
"wake_alarm._alarm.save_wake_state",
) as mock_save:
for _ in range(DISMISS_ROUNDS_REQUIRED):
mock_entry.get.return_value = alarm._progress.current_challenge.answer
alarm._view.entry_var.get.return_value = (
alarm._progress.current_challenge.answer
)
alarm._on_submit()
assert alarm.dismissed is True

38
wake_alarm/tests/test_alarm_part3.py
View File

@ -96,7 +96,7 @@ class TestBeepLoopPhases:
# Set alarm start to make elapsed > PHASE_SOFT_END minutes
import time as time_mod
alarm._alarm_start = time_mod.monotonic() - (PHASE_SOFT_END + 1) * 60
alarm._progress.started_at = time_mod.monotonic() - (PHASE_SOFT_END + 1) * 60
call_count = 0
@ -125,7 +125,7 @@ class TestBeepLoopPhases:
alarm = WakeAlarm(demo_mode=True)
import time as time_mod
alarm._alarm_start = time_mod.monotonic() - (PHASE_MEDIUM_END + 1) * 60
alarm._progress.started_at = time_mod.monotonic() - (PHASE_MEDIUM_END + 1) * 60
call_count = 0
@ -178,7 +178,7 @@ class TestUpdateTimerActive:
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._update_timer()
text = alarm._view.timer_label.configure.call_args[1]["text"]
text = alarm._view._views[0].labels.timer.configure.call_args[1]["text"]
assert text.startswith("Skip window:")
alarm._stop_beep.set()
@ -191,10 +191,10 @@ class TestUpdateTimerActive:
alarm = WakeAlarm(demo_mode=True)
# Far in the past so remaining == 0 -> the else branch.
alarm._alarm_start = time_mod.monotonic() - 60 * 60
alarm._progress.started_at = time_mod.monotonic() - 60 * 60
alarm.root.after.reset_mock()
alarm._update_timer()
text = alarm._view.timer_label.configure.call_args[1]["text"]
text = alarm._view._views[0].labels.timer.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()
@ -208,9 +208,9 @@ class TestUpdateTimerActive:
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._active = False
alarm._view.timer_label.configure.reset_mock()
alarm._view._views[0].labels.timer.configure.reset_mock()
alarm._update_timer()
alarm._view.timer_label.configure.assert_not_called()
alarm._view._views[0].labels.timer.configure.assert_not_called()
alarm._stop_beep.set()
@ -232,11 +232,11 @@ class TestFlashChallenge:
hint="Memorise",
)
alarm._progress.flash_remaining = 2
alarm._view.status_label.configure.reset_mock()
alarm._view._views[0].labels.status.configure.reset_mock()
alarm._flash_tick()
assert alarm._progress.flash_remaining == 1
alarm._view.status_label.configure.assert_called()
alarm._view._views[0].labels.status.configure.assert_called()
alarm._flash_tick()
assert alarm._progress.flash_remaining == 0
@ -246,7 +246,7 @@ class TestFlashChallenge:
# _code_label and _status_label share the same mock; inspect all calls.
all_texts = [
c.kwargs.get("text", "")
for c in alarm._view.code_label.configure.call_args_list
for c in alarm._view._views[0].labels.code.configure.call_args_list
]
assert any("?" in t for t in all_texts)
alarm._stop_beep.set()
@ -259,11 +259,11 @@ class TestFlashChallenge:
alarm = WakeAlarm(demo_mode=True)
alarm._active = False
alarm._progress.flash_remaining = 3
alarm._view.status_label.configure.reset_mock()
alarm._view._views[0].labels.status.configure.reset_mock()
alarm._flash_tick()
alarm._view.status_label.configure.assert_not_called()
alarm._view._views[0].labels.status.configure.assert_not_called()
alarm._stop_beep.set()
def test_wrong_flash_answer_reshows_code(
@ -280,15 +280,14 @@ class TestFlashChallenge:
answer="TESTCODE",
hint="Memorise",
)
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = "WRONGCODE"
alarm._view.code_label.configure.reset_mock()
alarm._view.entry_var.get.return_value = "WRONGCODE"
alarm._view._views[0].labels.code.configure.reset_mock()
alarm._on_submit()
assert alarm.dismissed is False
# Code label should be reconfigured (code shown again + countdown restarted).
alarm._view.code_label.configure.assert_called()
alarm._view._views[0].labels.code.configure.assert_called()
alarm._stop_beep.set()
def test_next_round_flash_starts_countdown(
@ -305,8 +304,7 @@ class TestFlashChallenge:
next_flash = _Challenge(
kind="flash", display="ABCDEFGH", answer="ABCDEFGH", hint="Memorise"
)
mock_entry = mock_tk_module.Entry.return_value
mock_entry.get.return_value = "4"
alarm._view.entry_var.get.return_value = "4"
with patch("wake_alarm._alarm._make_challenge", return_value=next_flash):
alarm._on_submit()
@ -327,7 +325,7 @@ class TestDismissWithoutSkip:
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
mock_widget = MagicMock()
alarm._view.container.winfo_children.return_value = [mock_widget]
alarm._view._views[0].container.winfo_children.return_value = [mock_widget]
with patch(
"wake_alarm._alarm.save_wake_state",
@ -354,7 +352,7 @@ class TestSkipWindowExpiredMessage:
alarm._on_skip_window_expired()
alarm._view.status_label.configure.assert_called_with(
alarm._view._views[0].labels.status.configure.assert_called_with(
text="No workout skip today.",
)
alarm._stop_beep.set()

134
wake_alarm/tests/test_alarm_part4.py
View File

@ -12,7 +12,9 @@ if TYPE_CHECKING:
from collections.abc import Generator
from wake_alarm._alarm import WakeAlarm
from wake_alarm._alarm_view import _EntryProxy
from wake_alarm._sinks import _AudioState
from wake_alarm.tests.conftest import TWO_OUTPUTS
# ---------------------------------------------------------------------------
# Helpers (duplicated from part 1 so this file is self-contained)
@ -26,7 +28,10 @@ def _make_mock_tk() -> MagicMock:
mock_root.winfo_screenwidth.return_value = 1920
mock_root.winfo_screenheight.return_value = 1080
mock.Tk.return_value = mock_root
mock.Frame.return_value = MagicMock()
# Distinct frames and buttons per call: a shared return_value would make
# "every monitor got one" indistinguishable from "one monitor got two".
mock.Frame.side_effect = lambda *_args, **_kwargs: MagicMock()
mock.Button.side_effect = lambda *_args, **_kwargs: MagicMock()
mock.Label.return_value = MagicMock()
mock.Entry.return_value = MagicMock()
mock.TclError = tk.TclError
@ -93,9 +98,9 @@ class TestGatelockHooks:
"""on_focus_ready forces focus onto the dismiss-code entry."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._view.entry.focus_force.reset_mock()
alarm.on_focus_ready()
alarm._view.entry.focus_force.assert_called_once()
alarm._view._views[0].entry.focus_force.reset_mock()
alarm.on_focus_ready(alarm._view._views[0])
alarm._view._views[0].entry.focus_force.assert_called_once()
alarm._stop_beep.set()
def test_on_callback_error_surfaces_and_refocuses(
@ -105,12 +110,12 @@ class TestGatelockHooks:
"""on_callback_error shows a message and refocuses the entry."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._view.entry.focus_force.reset_mock()
alarm._view._views[0].entry.focus_force.reset_mock()
alarm.on_callback_error()
alarm._view.status_label.configure.assert_called_with(
alarm._view._views[0].labels.status.configure.assert_called_with(
text="Something went wrong — try again.",
)
alarm._view.entry.focus_force.assert_called_once()
alarm._view._views[0].entry.focus_force.assert_called_once()
alarm._stop_beep.set()
@ -161,3 +166,118 @@ class TestClose:
alarm.on_close()
mock_restore.assert_called_once_with(state)
alarm._stop_beep.set()
class TestMultiMonitorSurfaces:
"""Per-output behaviour added in the gatelock v0.2.0 migration."""
def test_on_focus_ready_with_no_live_output(
self, mock_tk_module: MagicMock
) -> None:
"""A dark machine arms the alarm with nothing to focus."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._view._views[0].entry.focus_force.reset_mock()
alarm.on_focus_ready(None)
alarm._view._views[0].entry.focus_force.assert_not_called()
alarm._stop_beep.set()
def test_focus_force_skips_destroyed_entries(
self, mock_tk_module: MagicMock
) -> None:
"""A surface whose window already died is stepped over."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
dead, alive = MagicMock(), MagicMock()
dead.focus_force.side_effect = tk.TclError("gone")
proxy = _EntryProxy([dead, alive], MagicMock())
proxy.focus_force()
alive.focus_force.assert_called_once_with()
alarm._stop_beep.set()
def test_title_label_fans_out(self, mock_tk_module: MagicMock) -> None:
"""Every named label proxy reaches its per-monitor copies."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
alarm._view.title_label.configure(text="hi")
alarm._view._views[0].labels.title.configure.assert_called_with(text="hi")
alarm._stop_beep.set()
def test_teardown_forgets_a_dark_output(self, mock_tk_module: MagicMock) -> None:
"""An output going dark drops just its own view."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
surface = alarm._view._views[0]
alarm.teardown_surface(surface)
assert alarm._view._views == []
alarm._stop_beep.set()
def test_focus_force_when_every_entry_is_gone(
self, mock_tk_module: MagicMock
) -> None:
"""All surfaces destroyed: nothing to focus, and no exception."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
dead = MagicMock()
dead.focus_force.side_effect = tk.TclError("gone")
_EntryProxy([dead], MagicMock()).focus_force() # must not raise
alarm._stop_beep.set()
@pytest.mark.usefixtures("dual_output")
class TestFanOutAcrossMonitors:
"""The three remaining v0.2.0 loops, on a desk that actually has two.
Under the single-output default every one of these assertions holds even
if the loop body only ever ran for ``containers[0]`` -- which is exactly
the regression per-output locking exists to prevent.
"""
def test_dismiss_clears_and_repaints_every_monitor(
self, mock_tk_module: MagicMock
) -> None:
"""The dismissal message replaces the challenge on both screens."""
alarm = WakeAlarm(demo_mode=True)
containers = alarm._view.containers
assert len(containers) == len(TWO_OUTPUTS)
leftovers = {id(c): [MagicMock(), MagicMock()] for c in containers}
for container in containers:
container.winfo_children.return_value = leftovers[id(container)]
mock_tk_module.Label.reset_mock()
with patch("wake_alarm._alarm.save_wake_state"):
alarm._dismiss_alarm(earned_skip=True)
for widgets in leftovers.values():
for widget in widgets:
widget.destroy.assert_called_once_with()
parents = [call.args[0] for call in mock_tk_module.Label.call_args_list]
assert {id(parent) for parent in parents} == {id(c) for c in containers}
alarm._stop_beep.set()
def test_flash_recolours_every_monitor(self, mock_tk_module: MagicMock) -> None:
"""A flash tick reaches both surfaces, not just the primary."""
del mock_tk_module
alarm = WakeAlarm(demo_mode=True)
containers = alarm._view.containers
# __init__ already ran one flash tick; only the next one should count.
for container in containers:
container.configure.reset_mock()
container.winfo_toplevel.return_value.configure.reset_mock()
alarm._flash_step()
for container in containers:
container.configure.assert_called_once()
container.winfo_toplevel.return_value.configure.assert_called_once()
alarm._stop_beep.set()

367
wake_alarm/tests/test_escape_flow.py Normal file
View File

@ -0,0 +1,367 @@
"""Tests for the alarm's escape hatch.
The alarm was the only locker with no way out but the challenge itself. These
tests pin the two properties that make the new hatch safe: it is never the
*easy* first move, and it cannot be leaned on repeatedly.
"""
from __future__ import annotations
import tkinter as tk
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from gatelock import EscapePolicy, EscapeTracker
import pytest
from wake_alarm._alarm import WakeAlarm
from wake_alarm._constants import ESCAPE_OFFER_AFTER_SECONDS
from wake_alarm._escape_flow import (
ALARM_ESCAPE_POLICY,
EscapeCallbacks,
EscapeDialog,
build_tracker,
is_offerable,
)
from wake_alarm.tests.conftest import TWO_OUTPUTS
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
TODAY = "2026-07-25"
@pytest.fixture
def tracker(tmp_path: Path) -> EscapeTracker:
"""A tracker over a throwaway history and key."""
key = tmp_path / "hmac.key"
key.write_bytes(b"0" * 32)
return EscapeTracker(
EscapePolicy(name="cant-wake", label="Can't-wake"),
tmp_path / "escape.json",
key_file=key,
)
# ---------------------------------------------------------------------------
# Tk stand-ins (duplicated from the part files so this one is self-contained;
# they are autouse there, and hoisting them into conftest would silently
# neuter the audio and sync suites)
# ---------------------------------------------------------------------------
def _make_mock_tk() -> MagicMock:
"""Build a MagicMock that stands in for the tkinter module.
``Button`` gets a ``side_effect`` rather than a single ``return_value``:
every surface must end up with its *own* button, otherwise
``escape_buttons`` is N references to one mock and "packed on every
monitor" cannot be told apart from "packed once".
"""
mock = MagicMock()
mock_root = MagicMock()
mock_root.winfo_screenwidth.return_value = 1920
mock_root.winfo_screenheight.return_value = 1080
mock.Tk.return_value = mock_root
mock.Frame.side_effect = lambda *_args, **_kwargs: MagicMock()
mock.Label.return_value = MagicMock()
mock.Entry.return_value = MagicMock()
mock.Button.side_effect = lambda *_args, **_kwargs: MagicMock()
mock.TclError = tk.TclError
mock.END = tk.END
return mock
@pytest.fixture(autouse=True)
def _block_real_tk() -> Generator[MagicMock]:
"""Prevent any real Tk windows in tests."""
mock = _make_mock_tk()
with (
patch("wake_alarm._alarm.tk", mock),
patch("wake_alarm._alarm_view.tk", mock),
patch("wake_alarm._alarm.GateRoot", return_value=mock.Tk.return_value),
):
yield mock
@pytest.fixture(autouse=True)
def _block_extra_devices() -> Generator[MagicMock]:
"""Prevent real subprocess.Popen calls for extra ALSA devices."""
with (
patch("wake_alarm._alarm._play_on_all_sinks") as mock,
patch("wake_alarm._alarm._max_fans", return_value=False),
patch("wake_alarm._alarm._restore_fans"),
patch("wake_alarm._alarm._set_max_brightness"),
patch("wake_alarm._alarm._wake_display"),
patch("wake_alarm._alarm._restore_display"),
patch("wake_alarm._alarm._warn_if_no_real_sink"),
patch("wake_alarm._alarm._activate_all_outputs", return_value=None),
patch("wake_alarm._alarm._restore_audio_state"),
patch("wake_alarm._alarm.turn_on_plug"),
patch("wake_alarm._alarm.turn_off_plug"),
):
yield mock
class TestPolicy:
"""The alarm's hatch is exactly as hard as sick mode."""
def test_matches_sick_mode_defaults(self) -> None:
"""No softer budgets, no shorter wait, no shorter justification."""
assert ALARM_ESCAPE_POLICY.budget_per_7_days == 1
assert ALARM_ESCAPE_POLICY.budget_per_30_days == 3
assert ALARM_ESCAPE_POLICY.budget_per_90_days == 10
assert ALARM_ESCAPE_POLICY.lockout_seconds == 120
assert ALARM_ESCAPE_POLICY.justification_min_chars == 120
class TestIsOfferable:
"""Two independent gates."""
def test_hidden_early(self, tracker: EscapeTracker) -> None:
"""Reaching for it immediately is not allowed."""
assert is_offerable(tracker, 5) is False
def test_shown_after_the_threshold(self, tracker: EscapeTracker) -> None:
"""Once the alarm has been up a while, the hatch appears."""
assert is_offerable(tracker, ESCAPE_OFFER_AFTER_SECONDS + 1) is True
def test_hidden_when_budget_spent(self, tracker: EscapeTracker) -> None:
"""A spent budget removes it entirely, however long you wait."""
tracker.history.used_days = [TODAY]
with patch("gatelock._escape._today_iso", return_value=TODAY):
assert is_offerable(tracker, 10_000) is False
def test_build_tracker_loads(self, tmp_path: Path) -> None:
"""The real builder points at the config dir and loads cleanly."""
with patch("wake_alarm._escape_flow.ESCAPE_HISTORY_FILE", tmp_path / "h.json"):
assert build_tracker().history.used_days == []
class TestEscapeDialog:
"""The justification form."""
def _dialog(
self, tracker: EscapeTracker
) -> tuple[EscapeDialog, MagicMock, MagicMock]:
granted, cancelled = MagicMock(), MagicMock()
with patch("wake_alarm._escape_flow.tk") as fake_tk:
# Distinct entries, or all three fields are one mock and a test
# setting reason/onset/severity separately would only be setting
# the last of them three times over.
fake_tk.Entry.side_effect = lambda *_a, **_k: MagicMock()
fake_tk.Text.return_value.get.return_value = "x" * 130
dialog = EscapeDialog(
MagicMock(),
MagicMock(),
tracker,
EscapeCallbacks(on_granted=granted, on_cancel=cancelled),
)
return dialog, granted, cancelled
def test_rejects_a_lazy_justification(self, tracker: EscapeTracker) -> None:
"""A short description is refused and nothing is recorded."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._fields.description.get.return_value = "too short"
dialog._fields.reason.get.return_value = "tired"
dialog._fields.onset.get.return_value = "now"
dialog._fields.severity.get.return_value = "5"
dialog.submit()
granted.assert_not_called()
assert tracker.history.used_days == []
def test_non_numeric_severity_is_refused(self, tracker: EscapeTracker) -> None:
"""A non-numeric severity becomes 0, which fails validation."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._fields.description.get.return_value = "x" * 130
dialog._fields.reason.get.return_value = "flu"
dialog._fields.onset.get.return_value = "03:00"
dialog._fields.severity.get.return_value = "very"
dialog.submit()
granted.assert_not_called()
def test_accepts_and_starts_the_wait(self, tracker: EscapeTracker) -> None:
"""A proper justification is recorded and starts the countdown."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._fields.description.get.return_value = "x" * 130
dialog._fields.reason.get.return_value = "migraine"
dialog._fields.onset.get.return_value = "03:00"
dialog._fields.severity.get.return_value = "8"
dialog.submit()
assert len(tracker.history.used_days) == 1
assert dialog._remaining >= 0
granted.assert_not_called() # only after the wait
def test_refuses_when_budget_spent_mid_form(self, tracker: EscapeTracker) -> None:
"""The budget is re-checked at submit, not just when the form opened."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._fields.description.get.return_value = "x" * 130
dialog._fields.reason.get.return_value = "flu"
dialog._fields.onset.get.return_value = "03:00"
dialog._fields.severity.get.return_value = "8"
tracker.history.used_days = [TODAY]
with patch("gatelock._escape._today_iso", return_value=TODAY):
dialog.submit()
granted.assert_not_called()
def test_tick_counts_down_then_grants(self, tracker: EscapeTracker) -> None:
"""The wait elapses exactly once, then the alarm is released."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._remaining = 0
dialog._tick()
granted.assert_called_once_with()
def test_tick_reschedules_while_waiting(self, tracker: EscapeTracker) -> None:
"""A remaining wait re-arms itself instead of granting."""
dialog, granted, _cancel = self._dialog(tracker)
dialog._remaining = 3
dialog._tick()
granted.assert_not_called()
dialog.frame.after.assert_called_once()
@pytest.mark.usefixtures("dual_output")
class TestAlarmIntegration:
"""The hatch as the alarm actually uses it, on a two-monitor desk."""
def test_every_output_gets_its_own_button(self) -> None:
"""The premise the rest of this class rests on: two distinct buttons."""
alarm = WakeAlarm(demo_mode=True)
buttons = alarm._view.escape_buttons
assert len(buttons) == len(TWO_OUTPUTS)
assert len({id(button) for button in buttons}) == len(TWO_OUTPUTS)
alarm._stop_beep.set()
def test_button_hidden_until_earned(self) -> None:
"""Straight after the alarm starts, the hatch is not offered."""
alarm = WakeAlarm(demo_mode=True)
for button in alarm._view.escape_buttons:
button.pack.reset_mock()
alarm._update_timer()
for button in alarm._view.escape_buttons:
button.pack.assert_not_called()
alarm._stop_beep.set()
def test_button_revealed_once_earned(self) -> None:
"""After the threshold the button appears on every monitor.
Driven through ``_update_timer`` rather than the hatch directly, so
the per-second tick is on the hook for ever offering the hatch at all.
"""
alarm = WakeAlarm(demo_mode=True)
alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10
for button in alarm._view.escape_buttons:
button.pack.reset_mock()
alarm._update_timer()
for button in alarm._view.escape_buttons:
button.pack.assert_called()
alarm._stop_beep.set()
def test_not_reoffered_while_open(self) -> None:
"""With the form already up, the button is not re-packed."""
alarm = WakeAlarm(demo_mode=True)
alarm._progress.started_at -= ESCAPE_OFFER_AFTER_SECONDS + 10
alarm._escape.is_open = True
for button in alarm._view.escape_buttons:
button.pack.reset_mock()
alarm._update_timer()
for button in alarm._view.escape_buttons:
button.pack.assert_not_called()
alarm._stop_beep.set()
def test_open_swaps_every_surface(self) -> None:
"""Opening the form hides the challenge and builds a form per monitor."""
alarm = WakeAlarm(demo_mode=True)
surfaces = alarm._view.surfaces
assert len(surfaces) == len(TWO_OUTPUTS)
with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls:
alarm._escape.open()
assert alarm._escape.is_open is True
assert dialog_cls.call_count == len(surfaces)
# One dialog *per* container, not two against the same parent.
parents = [call.args[0] for call in dialog_cls.call_args_list]
assert {id(parent) for parent in parents} == {
id(view.container) for view in surfaces
}
for view in surfaces:
view.challenge_frame.pack_forget.assert_called_once_with()
alarm._stop_beep.set()
def test_a_dead_surface_does_not_block_the_others(self) -> None:
"""A monitor that has gone away raises TclError; the rest still swap."""
alarm = WakeAlarm(demo_mode=True)
dead, live = alarm._view.surfaces
dead.challenge_frame.pack_forget.side_effect = tk.TclError("bad path name")
with patch("wake_alarm._escape_flow.EscapeDialog") as dialog_cls:
alarm._escape.open()
assert dialog_cls.call_count == 1
assert dialog_cls.call_args.args[0] is live.container
alarm._stop_beep.set()
def test_cancel_puts_the_challenge_back(self) -> None:
"""Backing out destroys every form and re-packs every challenge.
The bool alone is not the property that matters: ``open`` took the
challenge off the screen, so a cancel that only flips a flag leaves
the alarm showing a dead form on both monitors.
"""
alarm = WakeAlarm(demo_mode=True)
surfaces = alarm._view.surfaces
dialogs: list[MagicMock] = []
def make_dialog(*_args: object, **_kwargs: object) -> MagicMock:
dialog = MagicMock()
dialogs.append(dialog)
return dialog
with patch("wake_alarm._escape_flow.EscapeDialog", side_effect=make_dialog):
alarm._escape.open()
# The frame was packed once when the view was built; only the re-pack
# that close() owes us should count.
for view in surfaces:
view.challenge_frame.pack.reset_mock()
alarm._escape.close()
assert alarm._escape.is_open is False
assert alarm.dismissed is False
assert len(dialogs) == len(TWO_OUTPUTS)
for dialog in dialogs:
dialog.frame.destroy.assert_called_once_with()
for view in surfaces:
view.challenge_frame.pack.assert_called_once_with()
alarm._stop_beep.set()
def test_granted_dismisses_without_a_skip(self) -> None:
"""Using the hatch silences the alarm but earns no workout skip."""
alarm = WakeAlarm(demo_mode=True)
with patch.object(alarm, "_dismiss_alarm") as dismiss:
alarm._on_escape_granted()
dismiss.assert_called_once_with(earned_skip=False)
alarm._stop_beep.set()