Add morning-session corner strip

Top-right always-on-top strip while the signed morning session is
in_progress: title, "Next: <step>", "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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZd3smuxDGmEgXkYP8
This commit is contained in:
Krzysztof Rudnicki 2026-11-07 08:19:12 +01:00
parent 537a3a79e4
commit f182453567
12 changed files with 730 additions and 8 deletions

View File

@ -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

View File

@ -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

View File

@ -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.

View File

@ -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

View File

@ -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() {

14
wake-alarm-strip.path Normal file
View File

@ -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

14
wake-alarm-strip.service Normal file
View File

@ -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

View File

@ -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())

View File

@ -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",

View File

@ -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

View File

@ -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

View File

@ -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))