From f18245356701a72b74a7ec8a4445b7e862889f20 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 7 Nov 2026 08:19:12 +0100 Subject: [PATCH] Add morning-session corner strip Top-right always-on-top strip while the signed morning session is in_progress: title, "Next: ", "Locks return at HH:MM". wake-alarm-strip.path starts it on every file rewrite; it re-reads the file each second, asks for a refresh every 30 s, exits 0 when the session ends. No display/output: warning, exit 1. Bump gatelock pin to 0.12.0 (corner_strip). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WzxdWZd3smuxDGmEgXkYP8 --- .pre-commit-config.yaml | 2 +- install.sh | 9 +- pyproject.toml | 2 +- requirements.txt | 6 +- uninstall.sh | 6 +- wake-alarm-strip.path | 14 ++ wake-alarm-strip.service | 14 ++ wake_alarm/_session_strip.py | 247 +++++++++++++++++++++ wake_alarm/tests/conftest.py | 11 + wake_alarm/tests/test_session_strip.py | 128 +++++++++++ wake_alarm/tests/test_session_strip_run.py | 196 ++++++++++++++++ wake_alarm/tests/test_strip_support.py | 103 +++++++++ 12 files changed, 730 insertions(+), 8 deletions(-) create mode 100644 wake-alarm-strip.path create mode 100644 wake-alarm-strip.service create mode 100644 wake_alarm/_session_strip.py create mode 100644 wake_alarm/tests/test_session_strip.py create mode 100644 wake_alarm/tests/test_session_strip_run.py create mode 100644 wake_alarm/tests/test_strip_support.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 2c5e880..407b5759 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -106,7 +106,7 @@ repos: # direct URLs for one distribution, so any pin that drifts from # the manifest becomes an unresolvable hook environment rather # than merely an old one. - - gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.10.0#subdirectory=gatelock + - gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.12.0#subdirectory=gatelock - crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync - freedays @ git+https://github.com/kuhyx/utils@freedays-v0.4.0#subdirectory=freedays - sickdays @ git+https://github.com/kuhyx/utils@sickdays-v0.3.0#subdirectory=sickdays diff --git a/install.sh b/install.sh index 5d56fb6..ad735bd 100755 --- a/install.sh +++ b/install.sh @@ -11,7 +11,7 @@ # What it does: # 1. Installs wake_alarm + dependencies for /usr/bin/python # 2. Installs the systemd user units and enables their timers (trigger, -# session, fallback, rtc, audit) +# session, strip, fallback, rtc, audit) # 3. Installs the root pieces of the fallback alarm: the ring WAVs, the # ALSA ring script + system unit, the sudoers lines that let the user # start/stop that unit, and (pc_wake.json "hibernate" mode only) the @@ -47,6 +47,8 @@ readonly TRIGGER_SERVICE_SRC="$SCRIPT_DIR/wake-alarm-trigger.service" readonly TRIGGER_TIMER_SRC="$SCRIPT_DIR/wake-alarm-trigger.timer" readonly SESSION_SERVICE_SRC="$SCRIPT_DIR/wake-alarm-session.service" readonly SESSION_TIMER_SRC="$SCRIPT_DIR/wake-alarm-session.timer" +readonly STRIP_SERVICE_SRC="$SCRIPT_DIR/wake-alarm-strip.service" +readonly STRIP_PATH_SRC="$SCRIPT_DIR/wake-alarm-strip.path" readonly SYSTEMD_USER_DIR="$HOME/.config/systemd/user" # Enforcement v2 user units (service + timer each). readonly V2_UNITS=(wake-alarm-fallback wake-alarm-rtc wake-alarm-audit) @@ -134,7 +136,8 @@ echo " wake_alarm and gatelock import under /usr/bin/python." echo "[2/4] Installing systemd user units..." mkdir -p "$SYSTEMD_USER_DIR" cp "$TRIGGER_SERVICE_SRC" "$TRIGGER_TIMER_SRC" \ - "$SESSION_SERVICE_SRC" "$SESSION_TIMER_SRC" "$SYSTEMD_USER_DIR/" + "$SESSION_SERVICE_SRC" "$SESSION_TIMER_SRC" \ + "$STRIP_SERVICE_SRC" "$STRIP_PATH_SRC" "$SYSTEMD_USER_DIR/" for unit in "${V2_UNITS[@]}"; do cp "$SCRIPT_DIR/$unit.service" "$SCRIPT_DIR/$unit.timer" "$SYSTEMD_USER_DIR/" done @@ -143,6 +146,8 @@ systemctl --user enable --now wake-alarm-trigger.timer echo " Trigger timer enabled and started (ticks once a minute)." systemctl --user enable --now wake-alarm-session.timer echo " Session timer enabled (07:00-11:00, every 2 min)." +systemctl --user enable --now wake-alarm-strip.path +echo " Strip path enabled (corner reminder while the morning is in progress)." for unit in "${V2_UNITS[@]}"; do systemctl --user enable --now "$unit.timer" done diff --git a/pyproject.toml b/pyproject.toml index 5736674..5d66059 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,7 @@ version = "1.0.0" description = "Trigger that starts the morning routine at the phone's alarm time" requires-python = ">=3.14" dependencies = [ - "gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.11.0#subdirectory=gatelock", + "gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.12.0#subdirectory=gatelock", # Firebase transport for the alarm read. _sync.py imports it lazily and # falls back to its stdlib GitHub path if the import fails, so a broken # install degrades the read instead of killing the per-minute trigger. diff --git a/requirements.txt b/requirements.txt index 29aee86..88f8fc0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,10 +4,10 @@ # it is missing — but it is required here for mypy/pylint/tests. bandit==1.9.4 codespell==2.4.3 -coverage==7.17.0 +coverage==7.17.2 crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync freedays @ git+https://github.com/kuhyx/utils@freedays-v0.4.0#subdirectory=freedays -gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.11.0#subdirectory=gatelock +gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.12.0#subdirectory=gatelock mypy==2.4.1 pre-commit==4.7.0 pylint==4.1.3 @@ -16,5 +16,5 @@ pytest-cov==7.1.0 pytest-randomly==5.0.0 pytest-sugar==1.1.1 pytest-xdist==3.8.0 -ruff==0.17.3 +ruff==0.17.4 sickdays @ git+https://github.com/kuhyx/utils@sickdays-v0.4.0#subdirectory=sickdays diff --git a/uninstall.sh b/uninstall.sh index 183bc6c..c4786ce 100755 --- a/uninstall.sh +++ b/uninstall.sh @@ -86,10 +86,14 @@ remove_trigger() { echo " Disabling the trigger timer" systemctl --user disable --now wake-alarm-trigger.timer >/dev/null 2>&1 || true systemctl --user disable --now wake-alarm-session.timer >/dev/null 2>&1 || true + systemctl --user disable --now wake-alarm-strip.path >/dev/null 2>&1 || true + systemctl --user stop wake-alarm-strip.service >/dev/null 2>&1 || true rm -f "$SYSTEMD_USER_DIR/wake-alarm-trigger.service" \ "$SYSTEMD_USER_DIR/wake-alarm-trigger.timer" \ "$SYSTEMD_USER_DIR/wake-alarm-session.service" \ - "$SYSTEMD_USER_DIR/wake-alarm-session.timer" + "$SYSTEMD_USER_DIR/wake-alarm-session.timer" \ + "$SYSTEMD_USER_DIR/wake-alarm-strip.service" \ + "$SYSTEMD_USER_DIR/wake-alarm-strip.path" } remove_enforcement() { diff --git a/wake-alarm-strip.path b/wake-alarm-strip.path new file mode 100644 index 0000000..3dea399 --- /dev/null +++ b/wake-alarm-strip.path @@ -0,0 +1,14 @@ +[Unit] +Description=Show the morning-session strip whenever the signed session file is rewritten +Documentation=https://github.com/kuhyx/wake-alarm + +[Path] +# wake-alarm-session.timer rewrites this every 2 min through the morning (and +# the strip itself asks for a refresh every 30 s while it is up). Starting a +# service that is already running is a no-op, so the rewrites cost nothing; +# once the session is over the strip exits 0 at once. +PathChanged=%h/.local/state/wake_alarm/morning_session.json +Unit=wake-alarm-strip.service + +[Install] +WantedBy=paths.target diff --git a/wake-alarm-strip.service b/wake-alarm-strip.service new file mode 100644 index 0000000..cfd32f0 --- /dev/null +++ b/wake-alarm-strip.service @@ -0,0 +1,14 @@ +[Unit] +Description=Wake Alarm morning-session strip (top-right corner reminder) +Documentation=https://github.com/kuhyx/wake-alarm + +[Service] +Type=simple +# Same display handling as leetcode-guard.service, the other Tk user unit: +# the session's own X server, no env fallbacks invented here. +Environment=DISPLAY=:0 +Environment=XAUTHORITY=%h/.Xauthority +# Exits 0 by itself when the morning session is not in progress (the normal +# case for most path triggers) and again the moment it ends. Exit 1 means no +# display or no live output; the next file rewrite tries again. +ExecStart=/usr/bin/python -m wake_alarm._session_strip diff --git a/wake_alarm/_session_strip.py b/wake_alarm/_session_strip.py new file mode 100644 index 0000000..0f9e8bc --- /dev/null +++ b/wake_alarm/_session_strip.py @@ -0,0 +1,247 @@ +"""A corner strip while the morning session is started but not finished. + +wake-alarm's signed ``morning_session.json`` makes the lockers stand down +while the morning is live. The screen simply clears, though, and kuhy still +owes the rest of the routine: the locks return at ``exempt_until`` out of +nowhere. This strip is the reminder: a small always-on-top box in the +top-right corner of the primary output (``gatelock.corner_strip``), saying the +session is not finished, what the phone's next step is, and when locks return. + + python -m wake_alarm._session_strip + +``wake-alarm-strip.path`` starts it whenever the file is rewritten (every two +minutes through the morning); starting a running service is a no-op. It shows +only while the entry is *signed, dated today, in_progress, and now < +exempt_until*, re-reading the file every second, and exits 0 the moment that +stops being true. While visible it also asks systemd every 30 s to run +``wake-alarm-session.service`` so ``next_step`` follows the phone instead of +the two-minute timer. + +Fail modes, out loud: nothing to show is exit 0 (normal, logged at info); no +display or no live output is a warning and exit 1, so the unit shows failed; +a missing, unreadable or tampered file shows nothing (the lockers do not +honour it either) and warns once per distinct problem, not once a second. +""" + +from __future__ import annotations + +from datetime import datetime +import json +import logging +import subprocess +import sys +import time +import tkinter as tk +from typing import TYPE_CHECKING + +from gatelock import LockConfig, enumerate_outputs +from gatelock.corner_strip import StripContent, build_corner_strip, corner_rect +from gatelock.log_integrity import verify_entry_hmac + +from wake_alarm._constants import MORNING_SESSION_FILE +from wake_alarm._logging_setup import configure_logging + +if TYPE_CHECKING: + from subprocess import Popen + +_logger = logging.getLogger(__name__) + +TITLE = "Morning session not finished" +REFRESH_EVERY_SECONDS = 30.0 +REFRESH_COMMAND = [ + "systemctl", + "--user", + "start", + "--no-block", + "wake-alarm-session.service", +] + + +class _Once: + """Warns about a problem once, until a clean read clears it. + + The strip reads the file every second; a file that stays broken for an + hour must be one journal line, not 3600. + """ + + def __init__(self) -> None: + self.last: str | None = None + + def warn(self, message: str) -> None: + """Log ``message`` at warning level unless it was just logged.""" + if message != self.last: + _logger.warning(message) + self.last = message + + def clear(self) -> None: + """A clean read happened: the next problem is news again.""" + self.last = None + + +_once = _Once() + + +def _local_now() -> datetime: + """The current time as an aware local datetime (a test seam).""" + return datetime.now().astimezone() + + +def _monotonic() -> float: + """Seconds on the monotonic clock (a test seam).""" + return time.monotonic() + + +def _popen(cmd: list[str]) -> Popen[bytes]: + """Start ``cmd`` without waiting for it (a test seam).""" + return subprocess.Popen(cmd) + + +def _make_root() -> tk.Tk: + """A withdrawn Tk root; only the strip's Toplevel is ever visible.""" + root = tk.Tk() + root.withdraw() + return root + + +def read_entry() -> dict[str, object] | None: + """The file's parsed JSON object, or None. Verification is the caller's.""" + try: + text = MORNING_SESSION_FILE.read_text() + except FileNotFoundError: + return None + except OSError as exc: + _once.warn(f"Cannot read {MORNING_SESSION_FILE}: {exc}") + return None + try: + entry = json.loads(text) + except json.JSONDecodeError as exc: + _once.warn(f"Cannot parse {MORNING_SESSION_FILE}: {exc}") + return None + if not isinstance(entry, dict): + _once.warn(f"{MORNING_SESSION_FILE} is not a JSON object") + return None + _once.clear() + return entry + + +def _why_not_today(entry: dict[str, object], now: datetime) -> str | None: + """Why ``entry`` is not a verified, live, in-progress morning; None if it is.""" + if not verify_entry_hmac(entry): + return "morning session entry is unsigned or tampered" + if entry.get("date") != now.date().isoformat(): + return f"morning session is dated {entry.get('date')}, not today" + if entry.get("outcome") != "in_progress": + return f"morning session outcome is {entry.get('outcome')}" + return None + + +def _exempt_until(entry: dict[str, object]) -> datetime | str: + """The entry's aware ``exempt_until``, or the sentence saying why not.""" + until = entry.get("exempt_until") + if not isinstance(until, str): + return "morning session has no exempt_until" + try: + parsed = datetime.fromisoformat(until) + except ValueError: + return f"morning session exempt_until is unreadable: {until!r}" + if parsed.tzinfo is None: + # Guessing its zone would be inventing a carrot the signer never gave. + return "morning session exempt_until has no timezone" + return parsed + + +def _verdict(entry: dict[str, object] | None, now: datetime) -> StripContent | str: + """The strip's content, or the sentence saying why there is none.""" + if entry is None: + return "no morning session file" + reason = _why_not_today(entry, now) + if reason is not None: + return reason + exempt_until = _exempt_until(entry) + if isinstance(exempt_until, str): + return exempt_until + if now >= exempt_until: + return f"the exemption ended at {exempt_until.isoformat()}" + lines: list[str] = [] + step = entry.get("next_step") + if isinstance(step, str) and step: + lines.append(f"Next: {step}") + lines.append(f"Locks return at {exempt_until.astimezone(now.tzinfo):%H:%M}") + return StripContent(TITLE, tuple(lines)) + + +def strip_content( + entry: dict[str, object] | None, now: datetime +) -> StripContent | None: + """What the strip says for ``entry`` at local ``now``; None shows nothing. + + Something is shown only for a verified entry dated today, still + ``in_progress``, whose ``exempt_until`` has not passed. + """ + verdict = _verdict(entry, now) + return verdict if isinstance(verdict, StripContent) else None + + +class Refresher: + """Starts wake-alarm-session.service every 30 s, never two at once.""" + + def __init__(self, *, start: float) -> None: + self._due = start + REFRESH_EVERY_SECONDS + self._proc: Popen[bytes] | None = None + + def tick(self, now: float) -> None: + """Reap a finished refresh, and start the next one when it is due.""" + if self._proc is not None: + code = self._proc.poll() + if code is None: + return + if code != 0: + _logger.warning("the session refresh start exited %d", code) + self._proc = None + if now < self._due: + return + self._due = now + REFRESH_EVERY_SECONDS + try: + self._proc = _popen(REFRESH_COMMAND) + except OSError as exc: + _logger.warning("could not start the session refresh: %s", exc) + + +def main() -> int: + """Show the strip while the morning is in progress; 0 when it is over.""" + configure_logging(component="session-strip") + verdict = _verdict(read_entry(), _local_now()) + if not isinstance(verdict, StripContent): + _logger.info("Nothing to show: %s", verdict) + return 0 + try: + root = _make_root() + except tk.TclError as exc: + _logger.warning("Cannot open the display for the strip: %s", exc) + return 1 + rect = corner_rect(enumerate_outputs(root)) + if rect is None: + _logger.warning("Cannot place the strip: no live output on this display") + root.destroy() + return 1 + strip = build_corner_strip(root, LockConfig(), verdict, rect=rect) + refresher = Refresher(start=_monotonic()) + ending: list[str] = [] + + def supply() -> StripContent | None: + current = _verdict(read_entry(), _local_now()) + if not isinstance(current, StripContent): + ending.append(current) + return None + refresher.tick(_monotonic()) + return current + + strip.schedule(supply, on_close=root.quit) + root.mainloop() + root.destroy() + _logger.info("Strip closed: %s", ending[0] if ending else "window gone") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/wake_alarm/tests/conftest.py b/wake_alarm/tests/conftest.py index e6aa964..7927f22 100644 --- a/wake_alarm/tests/conftest.py +++ b/wake_alarm/tests/conftest.py @@ -38,6 +38,8 @@ import freedays import pytest import sickdays +from wake_alarm import _session_strip + if TYPE_CHECKING: from collections.abc import Generator from pathlib import Path @@ -94,6 +96,15 @@ def _hermetic(tmp_path: Path) -> Generator[None]: "wake_alarm._session.MORNING_SESSION_FILE", tmp_path / "morning_session.json", ), + # The strip imported the path by value too; its tick re-reads it every + # second, its refresh spawns systemctl, and its warn-once memory must + # not leak from one test into the next. + patch( + "wake_alarm._session_strip.MORNING_SESSION_FILE", + tmp_path / "morning_session.json", + ), + patch("wake_alarm._session_strip._popen", _no_process), + patch("wake_alarm._session_strip._once", _session_strip._Once()), patch( "wake_alarm._workday_penalty.WORKDAY_PENALTY_FILE", tmp_path / "workday_penalty.json", diff --git a/wake_alarm/tests/test_session_strip.py b/wake_alarm/tests/test_session_strip.py new file mode 100644 index 0000000..ef1ba97 --- /dev/null +++ b/wake_alarm/tests/test_session_strip.py @@ -0,0 +1,128 @@ +"""The corner strip's content rules, file reader and thin seams.""" + +from __future__ import annotations + +from datetime import UTC, timedelta +import json +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +from gatelock.corner_strip import StripContent +import pytest + +from wake_alarm import _session_strip +from wake_alarm.tests.test_strip_support import NOW, STEP, signed + +if TYPE_CHECKING: + from pathlib import Path + +# conftest replaces _popen for every test; the real one is captured at import. +REAL_POPEN = _session_strip._popen + + +class TestStripContent: + def test_in_progress_says_next_step_and_when_locks_return(self) -> None: + assert _session_strip.strip_content(signed(), NOW) == StripContent( + "Morning session not finished", + (f"Next: {STEP}", "Locks return at 08:45"), + ) + + def test_next_step_null_drops_the_next_line(self) -> None: + content = _session_strip.strip_content(signed(next_step=None), NOW) + assert content == StripContent( + "Morning session not finished", ("Locks return at 08:45",) + ) + + def test_exempt_until_is_shown_in_local_time(self) -> None: + utc = (NOW + timedelta(minutes=45)).astimezone(UTC).isoformat() + content = _session_strip.strip_content(signed(exempt_until=utc), NOW) + assert content is not None + assert content.lines[-1] == "Locks return at 08:45" + + @pytest.mark.parametrize("outcome", ["completed", "failed", "none", "weird"]) + def test_other_outcomes_show_nothing(self, outcome: str) -> None: + assert _session_strip.strip_content(signed(outcome=outcome), NOW) is None + + def test_yesterday_shows_nothing(self) -> None: + assert _session_strip.strip_content(signed(date="2026-11-07"), NOW) is None + + def test_expired_shows_nothing(self) -> None: + later = NOW + timedelta(minutes=45) + assert _session_strip.strip_content(signed(), later) is None + + def test_tampered_shows_nothing(self) -> None: + entry = signed() + entry["next_step"] = "Something else" + assert _session_strip.strip_content(entry, NOW) is None + + def test_unsigned_shows_nothing(self) -> None: + entry = signed() + del entry["hmac"] + assert _session_strip.strip_content(entry, NOW) is None + + def test_missing_entry_shows_nothing(self) -> None: + assert _session_strip.strip_content(None, NOW) is None + + @pytest.mark.parametrize("until", [None, 5, "garbage", "2026-11-08T09:00:00"]) + def test_unusable_exempt_until_shows_nothing(self, until: object) -> None: + assert _session_strip.strip_content(signed(exempt_until=until), NOW) is None + + @pytest.mark.parametrize("step", [5, ""]) + def test_unusable_next_step_is_left_out(self, step: object) -> None: + content = _session_strip.strip_content(signed(next_step=step), NOW) + assert content is not None + assert content.lines == ("Locks return at 08:45",) + + +class TestReadEntry: + def test_missing_file_is_none_and_quiet( + self, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.DEBUG): + assert _session_strip.read_entry() is None + assert not [r for r in caplog.records if r.levelno >= logging.WARNING] + + def test_reads_the_file(self) -> None: + entry = signed() + _session_strip.MORNING_SESSION_FILE.write_text(json.dumps(entry)) + assert _session_strip.read_entry() == entry + + @pytest.mark.parametrize("text", ["{nope", "[1, 2]"]) + def test_unreadable_file_warns_once_per_state( + self, text: str, caplog: pytest.LogCaptureFixture + ) -> None: + _session_strip.MORNING_SESSION_FILE.write_text(text) + with caplog.at_level(logging.WARNING): + assert _session_strip.read_entry() is None + assert _session_strip.read_entry() is None + assert len([r for r in caplog.records if r.levelno >= logging.WARNING]) == 1 + + def test_directory_in_the_way_warns(self, caplog: pytest.LogCaptureFixture) -> None: + _session_strip.MORNING_SESSION_FILE.mkdir() + with caplog.at_level(logging.WARNING): + assert _session_strip.read_entry() is None + assert "Cannot read" in caplog.text + + +class TestHelpers: + def test_local_now_is_aware(self) -> None: + assert _session_strip._local_now().tzinfo is not None + + def test_monotonic_moves_forward(self) -> None: + first = _session_strip._monotonic() + assert _session_strip._monotonic() >= first + + def test_make_root_builds_a_withdrawn_tk_root(self) -> None: + with patch("wake_alarm._session_strip.tk.Tk") as tk_root: + root = _session_strip._make_root() + root.withdraw.assert_called_once() + assert root is tk_root.return_value + + def test_popen_runs_the_command(self) -> None: + with patch("wake_alarm._session_strip.subprocess.Popen") as popen: + REAL_POPEN(["true"]) + popen.assert_called_once_with(["true"]) + + def test_log_state_path_is_the_conftest_scratch(self, tmp_path: Path) -> None: + assert _session_strip.MORNING_SESSION_FILE.parent == tmp_path diff --git a/wake_alarm/tests/test_session_strip_run.py b/wake_alarm/tests/test_session_strip_run.py new file mode 100644 index 0000000..bd4cea8 --- /dev/null +++ b/wake_alarm/tests/test_session_strip_run.py @@ -0,0 +1,196 @@ +"""The corner strip's refresher and its main loop, on a fake Tk.""" + +from __future__ import annotations + +from contextlib import ExitStack +from datetime import timedelta +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +from gatelock.corner_strip import StripContent +import pytest + +from wake_alarm import _session_strip +from wake_alarm.tests.test_strip_support import ( + NOW, + STEP, + Display, + Procs, + signed, + write, +) + +if TYPE_CHECKING: + from collections.abc import Generator + + +class TestRefresher: + @pytest.fixture + def procs(self) -> Procs: + fake = Procs() + with patch("wake_alarm._session_strip._popen", fake): + yield fake + + def test_first_refresh_is_thirty_seconds_in(self, procs: Procs) -> None: + refresher = _session_strip.Refresher(start=100.0) + refresher.tick(129.0) + assert procs.started == [] + refresher.tick(130.0) + assert procs.started == [ + ["systemctl", "--user", "start", "--no-block", "wake-alarm-session.service"] + ] + + def test_one_start_per_interval(self, procs: Procs) -> None: + refresher = _session_strip.Refresher(start=0.0) + for second in range(30, 60): + refresher.tick(float(second)) + procs.procs[-1].poll.return_value = 0 + assert len(procs.started) == 1 + refresher.tick(60.0) + assert len(procs.started) == 2 + + def test_no_second_spawn_while_the_first_runs(self, procs: Procs) -> None: + refresher = _session_strip.Refresher(start=0.0) + refresher.tick(30.0) + refresher.tick(60.0) + refresher.tick(95.0) + assert len(procs.started) == 1 + procs.procs[0].poll.return_value = 0 + refresher.tick(96.0) + assert len(procs.started) == 2 + + def test_failed_exit_warns( + self, procs: Procs, caplog: pytest.LogCaptureFixture + ) -> None: + refresher = _session_strip.Refresher(start=0.0) + refresher.tick(30.0) + procs.procs[0].poll.return_value = 5 + with caplog.at_level(logging.WARNING): + refresher.tick(31.0) + assert "exited 5" in caplog.text + + def test_spawn_failure_warns_and_waits_for_the_next_interval( + self, procs: Procs, caplog: pytest.LogCaptureFixture + ) -> None: + procs.error = FileNotFoundError("systemctl") + refresher = _session_strip.Refresher(start=0.0) + with caplog.at_level(logging.WARNING): + refresher.tick(30.0) + refresher.tick(31.0) + assert caplog.text.count("could not start") == 1 + procs.error = None + refresher.tick(60.0) + assert len(procs.started) == 1 + + +@pytest.fixture +def display(monkeypatch: pytest.MonkeyPatch) -> Generator[Display]: + """A fake Tk world with a pinned clock and a recording ``_popen``.""" + fake = Display() + monkeypatch.setattr("wake_alarm._session_strip._local_now", lambda: NOW) + monkeypatch.setattr("wake_alarm._session_strip._monotonic", lambda: 0.0) + monkeypatch.setattr("wake_alarm._session_strip._popen", Procs()) + with ExitStack() as stack: + for item in fake.patches(): + stack.enter_context(item) + yield fake + + +class TestMain: + def test_nothing_to_show_exits_zero_without_a_window( + self, display: Display, caplog: pytest.LogCaptureFixture + ) -> None: + write(signed(outcome="completed")) + with caplog.at_level(logging.INFO): + assert _session_strip.main() == 0 + display.root.mainloop.assert_not_called() + assert display.built == [] + assert "completed" in caplog.text + + def test_missing_file_exits_zero(self, display: Display) -> None: + assert _session_strip.main() == 0 + assert display.built == [] + + def test_no_output_is_a_warning_and_a_failure( + self, display: Display, caplog: pytest.LogCaptureFixture + ) -> None: + write(signed()) + display.rect = None + with caplog.at_level(logging.WARNING): + assert _session_strip.main() == 1 + assert "no live output" in caplog.text + display.root.destroy.assert_called_once() + + def test_shows_then_closes_when_the_session_completes( + self, display: Display + ) -> None: + write(signed()) + + def flip(tick: int) -> None: + if tick == 3: + write(signed(outcome="completed", next_step=None)) + + display.on_tick = flip + assert _session_strip.main() == 0 + assert display.built == [ + StripContent( + "Morning session not finished", + (f"Next: {STEP}", "Locks return at 08:45"), + ) + ] + assert display.ticks == 3 + display.window.destroy.assert_called_once() + display.root.quit.assert_called_once() + display.root.destroy.assert_called_once() + + def test_closes_when_the_exemption_ends( + self, display: Display, monkeypatch: pytest.MonkeyPatch + ) -> None: + write(signed()) + + def expire(tick: int) -> None: + if tick == 2: + monkeypatch.setattr( + "wake_alarm._session_strip._local_now", + lambda: NOW + timedelta(minutes=45), + ) + + display.on_tick = expire + assert _session_strip.main() == 0 + assert display.ticks == 2 + + def test_refresh_starts_while_visible_and_not_after_close( + self, display: Display, monkeypatch: pytest.MonkeyPatch + ) -> None: + procs = Procs() + clock = {"t": 0.0} + monkeypatch.setattr("wake_alarm._session_strip._popen", procs) + monkeypatch.setattr("wake_alarm._session_strip._monotonic", lambda: clock["t"]) + write(signed()) + + def advance(tick: int) -> None: + clock["t"] = tick * 10.0 + if tick == 4: + write(signed(outcome="completed")) + + display.on_tick = advance + assert _session_strip.main() == 0 + # Ticks run at t=10, 20, 30 (refresh due), then 40 closes the strip. + assert len(procs.started) == 1 + + def test_unopenable_display_is_logged_and_fails( + self, display: Display, caplog: pytest.LogCaptureFixture + ) -> None: + import tkinter as tk + + write(signed()) + with ( + patch( + "wake_alarm._session_strip._make_root", + side_effect=tk.TclError("no display"), + ), + caplog.at_level(logging.WARNING), + ): + assert _session_strip.main() == 1 + assert "no display" in caplog.text diff --git a/wake_alarm/tests/test_strip_support.py b/wake_alarm/tests/test_strip_support.py new file mode 100644 index 0000000..f284f01 --- /dev/null +++ b/wake_alarm/tests/test_strip_support.py @@ -0,0 +1,103 @@ +"""Shared fakes for the corner-strip tests: signed entries, Tk and Popen.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import json +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from gatelock.corner_strip import CornerStrip, StripContent +from gatelock.log_integrity import compute_entry_hmac + +from wake_alarm import _session_strip + +if TYPE_CHECKING: + from collections.abc import Callable + +TZ = timezone(timedelta(hours=1)) +NOW = datetime(2026, 11, 8, 8, 0, tzinfo=TZ) +STEP = "Shower, then tap the bathroom tag again" + + +def signed(**overrides: object) -> dict[str, object]: + """A verified in-progress entry for NOW's day, with fields overridden.""" + entry: dict[str, object] = { + "date": "2026-11-08", + "outcome": "in_progress", + "exempt_until": (NOW + timedelta(minutes=45)).isoformat(), + "next_step": STEP, + } + entry.update(overrides) + entry["hmac"] = compute_entry_hmac(entry) + return entry + + +class Procs: + """A fake ``_popen`` recording starts; each proc finishes when told.""" + + def __init__(self) -> None: + self.started: list[list[str]] = [] + self.procs: list[MagicMock] = [] + self.error: OSError | None = None + + def __call__(self, cmd: list[str]) -> MagicMock: + if self.error is not None: + raise self.error + self.started.append(cmd) + proc = MagicMock() + proc.poll.return_value = None + self.procs.append(proc) + return proc + + +class Display: + """A fake Tk world: records the strip's tick callbacks and drives them.""" + + def __init__(self) -> None: + self.root = MagicMock(name="root") + self.window = MagicMock(name="window") + self.callbacks: list[Callable[[], None]] = [] + self.window.after.side_effect = self._after + self.root.quit.side_effect = self.callbacks.clear + self.root.mainloop.side_effect = self._mainloop + self.ticks = 0 + self.on_tick: Callable[[int], None] = lambda _n: None + self.rect: object = MagicMock(name="rect") + self.built: list[StripContent] = [] + + def _after(self, _ms: int, callback: Callable[[], None]) -> str: + self.callbacks.append(callback) + return "job" + + def _mainloop(self) -> None: + while self.callbacks: + self.ticks += 1 + self.on_tick(self.ticks) + self.callbacks.pop(0)() + if self.ticks > 50: # runaway-loop guard + msg = "strip never closed" + raise AssertionError(msg) + + def build( + self, _root: object, _config: object, content: StripContent, **kwargs: object + ) -> object: + + assert kwargs["rect"] is self.rect + self.built.append(content) + return CornerStrip(self.window, MagicMock(), [MagicMock(), MagicMock()]) + + def patches(self) -> list[object]: + return [ + patch("wake_alarm._session_strip._make_root", return_value=self.root), + patch("wake_alarm._session_strip.enumerate_outputs"), + patch( + "wake_alarm._session_strip.corner_rect", + side_effect=lambda _scan: self.rect, + ), + patch("wake_alarm._session_strip.build_corner_strip", self.build), + ] + + +def write(entry: dict[str, object]) -> None: + _session_strip.MORNING_SESSION_FILE.write_text(json.dumps(entry))