diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index 8efb61b..27dfaf7 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -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) diff --git a/wake_alarm/_alarm_view.py b/wake_alarm/_alarm_view.py index c64efca..bd0a4fd 100644 --- a/wake_alarm/_alarm_view.py +++ b/wake_alarm/_alarm_view.py @@ -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("", 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, ) diff --git a/wake_alarm/_constants.py b/wake_alarm/_constants.py index 7898f9f..77cb06f 100644 --- a/wake_alarm/_constants.py +++ b/wake_alarm/_constants.py @@ -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. diff --git a/wake_alarm/_escape_flow.py b/wake_alarm/_escape_flow.py new file mode 100644 index 0000000..b90a82e --- /dev/null +++ b/wake_alarm/_escape_flow.py @@ -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 diff --git a/wake_alarm/tests/conftest.py b/wake_alarm/tests/conftest.py new file mode 100644 index 0000000..2bf6512 --- /dev/null +++ b/wake_alarm/tests/conftest.py @@ -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 diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index bea6d8c..27d5046 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -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 diff --git a/wake_alarm/tests/test_alarm_part3.py b/wake_alarm/tests/test_alarm_part3.py index 77cc035..15cecad 100644 --- a/wake_alarm/tests/test_alarm_part3.py +++ b/wake_alarm/tests/test_alarm_part3.py @@ -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() diff --git a/wake_alarm/tests/test_alarm_part4.py b/wake_alarm/tests/test_alarm_part4.py index cce2435..b5036c8 100644 --- a/wake_alarm/tests/test_alarm_part4.py +++ b/wake_alarm/tests/test_alarm_part4.py @@ -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() diff --git a/wake_alarm/tests/test_escape_flow.py b/wake_alarm/tests/test_escape_flow.py new file mode 100644 index 0000000..bfd72a6 --- /dev/null +++ b/wake_alarm/tests/test_escape_flow.py @@ -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()