mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 17:47:24 +01:00
gatelock 0.12.0: corner strip, 180 s wait
Add gatelock.corner_strip, the top-right always-on-top status strip for a gate that is stood down but not done. Raise the morning-session wait from 30 s to 180 s: on 2026-11-07 a boot-time DNS race let the 30 s wait expire and both lockers armed mid morning-session; wake-alarm's refresher now retries 150 s in-process. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WzxdWZd3smuxDGmEgXkYP8
This commit is contained in:
parent
7af6afe51a
commit
97d043d424
@ -69,8 +69,14 @@ unchanged so existing signed history keeps verifying.
|
||||
signed `morning_session.json` (the wake-early carrot: no lock until the signed
|
||||
`exempt_until`, 11:00 for a morning completed in time). screen-locker,
|
||||
leetcode-guard and book-guard all call it; the policy itself stays in
|
||||
wake-alarm. Path and retry budget are arguments so each consumer's tests keep
|
||||
redirecting their own constant.
|
||||
wake-alarm. The arming run waits up to 180 s for a freshly booted PC's
|
||||
refresher (status paths never wait). Path and retry budget are arguments so
|
||||
each consumer's tests keep redirecting their own constant.
|
||||
|
||||
`gatelock.corner_strip` is the top-right always-on-top status strip for a gate
|
||||
that is stood down but not done (`build_corner_strip`, `CornerStrip`,
|
||||
`StripContent`, `corner_rect`, `corner_geometry`). It never grabs input; its
|
||||
tick ends when the owner's `supply` returns `None`.
|
||||
|
||||
`gatelock.night_lockdown.evening_lockdown_today` answers "has tonight's
|
||||
evening night-lockdown already happened?" from night-lockdown's
|
||||
|
||||
195
gatelock/gatelock/corner_strip.py
Normal file
195
gatelock/gatelock/corner_strip.py
Normal file
@ -0,0 +1,195 @@
|
||||
# Copyright (c) 2026 Krzysztof Rudnicki
|
||||
"""A small always-on-top strip in the top-right corner of the primary output.
|
||||
|
||||
For the time a gate is *stood down but not done*: the lock is not up, yet the
|
||||
user still owes something (a morning session to finish, solves to submit).
|
||||
Without it the screen just clears and the lock returns out of nowhere.
|
||||
|
||||
Top-right because the centre is where the user's actual work sits and the
|
||||
bottom-right is where desktop notifications land. Lifted from leetcode-guard's
|
||||
study strip so every gate shows the same thing in the same place.
|
||||
|
||||
The strip never grabs input and never re-asserts a grab; its tick only
|
||||
refreshes the text and re-lifts the window. ``supply`` returning ``None`` is
|
||||
the one way it ends on its own, so the owner decides when the debt is gone.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable, Sequence
|
||||
|
||||
from gatelock._config import LockConfig
|
||||
from gatelock._output_types import OutputRect, OutputScan
|
||||
|
||||
_logger: Final = logging.getLogger(__name__)
|
||||
|
||||
_TOPMOST_ON: Final = 1
|
||||
CORNER_STRIP_WIDTH_PX: Final = 420
|
||||
CORNER_STRIP_TICK_MS: Final = 1_000
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StripContent:
|
||||
"""What the strip says: one bold title and a few plain lines."""
|
||||
|
||||
title: str
|
||||
lines: tuple[str, ...]
|
||||
|
||||
|
||||
def corner_rect(scan: OutputScan) -> OutputRect | None:
|
||||
"""The rectangle to sit on: the primary live output, else the first live.
|
||||
|
||||
The whole X screen spans every monitor on a multi-head desk and would park
|
||||
the strip on the far one, so it is always one output's rectangle.
|
||||
"""
|
||||
live = scan.live
|
||||
for output in live:
|
||||
if output.primary:
|
||||
return output.rect
|
||||
return live[0].rect if live else None
|
||||
|
||||
|
||||
def corner_geometry(rect: OutputRect, *, width: int, height: int, inset: int) -> str:
|
||||
"""Tk geometry for a ``width`` x ``height`` window in ``rect``'s top-right."""
|
||||
x = rect.x + rect.width - width - inset
|
||||
y = rect.y + inset
|
||||
return f"{width}x{height}+{x}+{y}"
|
||||
|
||||
|
||||
class CornerStrip:
|
||||
"""Handle on a built strip: refresh it, tick it, tear it down."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
window: tk.Toplevel,
|
||||
title_label: tk.Label,
|
||||
line_labels: Sequence[tk.Label],
|
||||
*,
|
||||
tick_ms: int = CORNER_STRIP_TICK_MS,
|
||||
) -> None:
|
||||
"""Wrap an already-built strip window."""
|
||||
self.window = window
|
||||
self._title = title_label
|
||||
self._lines = tuple(line_labels)
|
||||
self._tick_ms = tick_ms
|
||||
self._job: str | None = None
|
||||
self._destroyed = False
|
||||
|
||||
@property
|
||||
def closed(self) -> bool:
|
||||
"""Whether the strip is gone (destroyed, or broke on a refresh)."""
|
||||
return self._destroyed
|
||||
|
||||
def show(self, content: StripContent) -> None:
|
||||
"""Write ``content`` and re-lift; lines past the content are blanked."""
|
||||
if self._destroyed:
|
||||
return
|
||||
try:
|
||||
self._title.configure(text=content.title)
|
||||
for index, label in enumerate(self._lines):
|
||||
text = content.lines[index] if index < len(content.lines) else ""
|
||||
label.configure(text=text)
|
||||
self.window.lift()
|
||||
except tk.TclError:
|
||||
# A fullscreen window over us is cosmetic; an exception escaping a
|
||||
# Tk callback is not. Stop ticking and say so.
|
||||
_logger.warning("the corner strip could not refresh", exc_info=True)
|
||||
self._destroyed = True
|
||||
|
||||
def schedule(
|
||||
self,
|
||||
supply: Callable[[], StripContent | None],
|
||||
*,
|
||||
on_close: Callable[[], None],
|
||||
) -> None:
|
||||
"""Tick: show ``supply()`` until it returns None, then close."""
|
||||
if self._destroyed:
|
||||
return
|
||||
|
||||
def run() -> None:
|
||||
content = supply()
|
||||
if content is None:
|
||||
self.destroy()
|
||||
on_close()
|
||||
return
|
||||
self.show(content)
|
||||
if not self._destroyed:
|
||||
self._job = self.window.after(self._tick_ms, run)
|
||||
|
||||
self._job = self.window.after(self._tick_ms, run)
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""Cancel the tick and tear the window down. Idempotent."""
|
||||
if self._destroyed:
|
||||
return
|
||||
self._destroyed = True
|
||||
try:
|
||||
if self._job is not None:
|
||||
self.window.after_cancel(self._job)
|
||||
self.window.destroy()
|
||||
except tk.TclError:
|
||||
_logger.warning("the corner strip could not be destroyed", exc_info=True)
|
||||
|
||||
|
||||
def build_corner_strip(
|
||||
root: tk.Misc,
|
||||
config: LockConfig,
|
||||
content: StripContent,
|
||||
*,
|
||||
rect: OutputRect,
|
||||
max_lines: int = 3,
|
||||
) -> CornerStrip:
|
||||
"""Create the strip in ``rect``'s top-right corner, showing ``content``.
|
||||
|
||||
``max_lines`` labels are built up front so later content can grow into
|
||||
them without re-layout; the height is whatever those labels ask for.
|
||||
"""
|
||||
width = CORNER_STRIP_WIDTH_PX
|
||||
window = tk.Toplevel(root)
|
||||
# Override-redirect must be set while withdrawn or the geometry is ignored.
|
||||
window.withdraw()
|
||||
window.overrideredirect(boolean=True)
|
||||
window.configure(bg=config.palette.field_bg)
|
||||
wrap = width - 2 * config.space("md")
|
||||
|
||||
title = tk.Label(
|
||||
window,
|
||||
text=content.title,
|
||||
font=config.font("label", bold=True),
|
||||
fg=config.palette.fg,
|
||||
bg=config.palette.field_bg,
|
||||
wraplength=wrap,
|
||||
)
|
||||
title.pack(pady=(config.space("sm"), 0), padx=config.space("md"))
|
||||
lines = []
|
||||
for index in range(max_lines):
|
||||
label = tk.Label(
|
||||
window,
|
||||
text=content.lines[index] if index < len(content.lines) else "",
|
||||
font=config.font("body" if index == 0 else "caption"),
|
||||
fg=config.palette.accent if index == 0 else config.palette.muted,
|
||||
bg=config.palette.field_bg,
|
||||
wraplength=wrap,
|
||||
justify="left",
|
||||
)
|
||||
label.pack(padx=config.space("md"))
|
||||
lines.append(label)
|
||||
if lines:
|
||||
lines[-1].pack_configure(pady=(0, config.space("sm")))
|
||||
|
||||
window.update_idletasks()
|
||||
height = int(window.winfo_reqheight())
|
||||
window.geometry(
|
||||
corner_geometry(rect, width=width, height=height, inset=config.space("md"))
|
||||
)
|
||||
window.deiconify()
|
||||
window.lift()
|
||||
# A hint most window managers honour even for override-redirect windows.
|
||||
window.attributes("-topmost", _TOPMOST_ON)
|
||||
return CornerStrip(window, title, lines)
|
||||
@ -14,7 +14,8 @@ cannot mint an unlocked day.
|
||||
|
||||
The one wrinkle is a PC booted mid-morning: the file is missing or yesterday's
|
||||
until wake-alarm's ``Persistent=true`` timer catches up, which needs the
|
||||
network. The arming run waits for that, bounded; status paths never do.
|
||||
network. The arming run waits for that, bounded (180 s, covering wake-alarm's
|
||||
150 s in-process retry); status paths never do.
|
||||
|
||||
The path and the retry budget are parameters, not module state, so each
|
||||
consumer's test suite keeps redirecting its own constant.
|
||||
@ -45,7 +46,10 @@ MORNING_SESSION_FILE: Final = (
|
||||
# Only inside this local window is a non-today file worth waiting for: the
|
||||
# refresher runs 05:00-11:00, so outside it "not today's" is simply the truth.
|
||||
MORNING_WINDOW: Final = ((5, 0), (11, 0))
|
||||
MORNING_RETRY_SECONDS: Final = 30.0
|
||||
# Covers wake-alarm's own in-process retry (150 s) plus a margin: on
|
||||
# 2026-11-07 DNS came up seconds after boot, the refresher's one try had
|
||||
# already failed, and a 30 s wait locked a morning that was being earned.
|
||||
MORNING_RETRY_SECONDS: Final = 180.0
|
||||
MORNING_POLL_SECONDS: Final = 5.0
|
||||
|
||||
|
||||
|
||||
211
gatelock/gatelock/tests/test_corner_strip.py
Normal file
211
gatelock/gatelock/tests/test_corner_strip.py
Normal file
@ -0,0 +1,211 @@
|
||||
# Copyright (c) 2026 Krzysztof Rudnicki
|
||||
"""Tests for the top-right status strip shown while a gate stands down."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from gatelock import corner_strip
|
||||
from gatelock._config import LockConfig
|
||||
from gatelock._outputs import Output, OutputRect, OutputScan
|
||||
from gatelock.corner_strip import (
|
||||
CornerStrip,
|
||||
StripContent,
|
||||
build_corner_strip,
|
||||
corner_geometry,
|
||||
corner_rect,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
import pytest
|
||||
|
||||
_LEFT = Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160))
|
||||
_RIGHT = Output(
|
||||
"HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440), primary=True
|
||||
)
|
||||
_DARK = Output("DP-1", connected=False, rect=None)
|
||||
|
||||
|
||||
def test_corner_rect_prefers_the_primary_output() -> None:
|
||||
scan = OutputScan((_LEFT, _RIGHT), source="randr", ok=True)
|
||||
assert corner_rect(scan) == _RIGHT.rect
|
||||
|
||||
|
||||
def test_corner_rect_falls_back_to_the_first_live_output() -> None:
|
||||
left = Output("DP-0", connected=True, rect=OutputRect(0, 0, 100, 100))
|
||||
scan = OutputScan((_DARK, left), source="randr", ok=True)
|
||||
assert corner_rect(scan) == left.rect
|
||||
|
||||
|
||||
def test_corner_rect_is_none_when_nothing_is_live() -> None:
|
||||
assert corner_rect(OutputScan((_DARK,), source="randr", ok=True)) is None
|
||||
|
||||
|
||||
def test_geometry_hugs_the_top_right_corner_of_the_rect() -> None:
|
||||
rect = OutputRect(3840, 0, 2560, 1440)
|
||||
assert corner_geometry(rect, width=400, height=120, inset=16) == ("400x120+5984+16")
|
||||
|
||||
|
||||
def _after_into(jobs: list[Callable[[], None]]) -> Callable[..., str]:
|
||||
"""A ``window.after`` stand-in that queues the callback instead of timing it."""
|
||||
|
||||
def after(_ms: int, fn: Callable[[], None]) -> str:
|
||||
jobs.append(fn)
|
||||
return f"job{len(jobs)}"
|
||||
|
||||
return after
|
||||
|
||||
|
||||
def _strip() -> tuple[CornerStrip, MagicMock, MagicMock, list[MagicMock]]:
|
||||
window, title = MagicMock(), MagicMock()
|
||||
lines = [MagicMock(), MagicMock()]
|
||||
return CornerStrip(window, title, lines, tick_ms=50), window, title, lines
|
||||
|
||||
|
||||
def test_show_writes_title_and_lines_and_relifts() -> None:
|
||||
strip, window, title, lines = _strip()
|
||||
strip.show(StripContent("Morning session", ("Next: shower", "Locks at 09:13")))
|
||||
title.configure.assert_called_with(text="Morning session")
|
||||
lines[0].configure.assert_called_with(text="Next: shower")
|
||||
lines[1].configure.assert_called_with(text="Locks at 09:13")
|
||||
window.lift.assert_called()
|
||||
|
||||
|
||||
def test_show_blanks_lines_the_content_does_not_fill() -> None:
|
||||
strip, _window, _title, lines = _strip()
|
||||
strip.show(StripContent("t", ("only one",)))
|
||||
lines[1].configure.assert_called_with(text="")
|
||||
|
||||
|
||||
def test_schedule_ticks_until_the_supply_says_none_then_closes() -> None:
|
||||
strip, window, _title, _lines = _strip()
|
||||
jobs: list[Callable[[], None]] = []
|
||||
window.after.side_effect = _after_into(jobs)
|
||||
supplies = iter([StripContent("a", ()), None])
|
||||
closed: list[bool] = []
|
||||
strip.schedule(lambda: next(supplies), on_close=lambda: closed.append(True))
|
||||
jobs.pop(0)() # first tick: content -> shown, rescheduled
|
||||
assert jobs, "the tick must reschedule itself while there is content"
|
||||
jobs.pop(0)() # second tick: None -> closed, not rescheduled
|
||||
assert not jobs
|
||||
assert closed == [True]
|
||||
window.destroy.assert_called_once()
|
||||
|
||||
|
||||
def test_destroy_is_idempotent_and_cancels_the_tick() -> None:
|
||||
strip, window, _title, _lines = _strip()
|
||||
window.after.return_value = "job1"
|
||||
strip.schedule(lambda: StripContent("a", ()), on_close=lambda: None)
|
||||
strip.destroy()
|
||||
strip.destroy()
|
||||
window.after_cancel.assert_called_once_with("job1")
|
||||
window.destroy.assert_called_once()
|
||||
|
||||
|
||||
def test_a_tcl_error_on_refresh_stops_the_strip_loudly(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
import tkinter as tk
|
||||
|
||||
strip, window, title, _lines = _strip()
|
||||
title.configure.side_effect = tk.TclError("gone")
|
||||
strip.show(StripContent("a", ()))
|
||||
assert "could not refresh" in caplog.text
|
||||
strip.destroy()
|
||||
window.destroy.assert_not_called()
|
||||
|
||||
|
||||
def test_build_places_an_override_redirect_topmost_window() -> None:
|
||||
window = MagicMock()
|
||||
window.winfo_reqheight.return_value = 90
|
||||
with (
|
||||
patch.object(corner_strip.tk, "Toplevel", return_value=window),
|
||||
patch.object(corner_strip.tk, "Label", side_effect=lambda *a, **k: MagicMock()),
|
||||
):
|
||||
strip = build_corner_strip(
|
||||
MagicMock(),
|
||||
LockConfig(),
|
||||
StripContent("Morning session", ("Next: shower",)),
|
||||
rect=OutputRect(0, 0, 1920, 1080),
|
||||
max_lines=2,
|
||||
)
|
||||
window.overrideredirect.assert_called_once_with(boolean=True)
|
||||
window.attributes.assert_called_with("-topmost", 1)
|
||||
spec = window.geometry.call_args.args[0]
|
||||
assert spec.startswith("420x90+")
|
||||
assert isinstance(strip, CornerStrip)
|
||||
|
||||
|
||||
def test_closed_reports_destroyed_state() -> None:
|
||||
strip, _window, _title, _lines = _strip()
|
||||
assert strip.closed is False
|
||||
strip.destroy()
|
||||
assert strip.closed is True
|
||||
|
||||
|
||||
def test_show_after_destroy_does_nothing() -> None:
|
||||
strip, window, title, _lines = _strip()
|
||||
strip.destroy()
|
||||
strip.show(StripContent("late", ()))
|
||||
title.configure.assert_not_called()
|
||||
window.lift.assert_not_called()
|
||||
|
||||
|
||||
def test_schedule_after_destroy_does_not_tick() -> None:
|
||||
strip, window, _title, _lines = _strip()
|
||||
strip.destroy()
|
||||
strip.schedule(lambda: StripContent("a", ()), on_close=lambda: None)
|
||||
window.after.assert_not_called()
|
||||
|
||||
|
||||
def test_a_tick_that_breaks_the_strip_does_not_reschedule() -> None:
|
||||
import tkinter as tk
|
||||
|
||||
strip, window, title, _lines = _strip()
|
||||
jobs: list[Callable[[], None]] = []
|
||||
window.after.side_effect = _after_into(jobs)
|
||||
title.configure.side_effect = tk.TclError("gone")
|
||||
strip.schedule(lambda: StripContent("a", ()), on_close=lambda: None)
|
||||
jobs.pop(0)()
|
||||
assert strip.closed
|
||||
assert not jobs
|
||||
|
||||
|
||||
def test_destroy_without_a_tick_skips_the_cancel() -> None:
|
||||
strip, window, _title, _lines = _strip()
|
||||
strip.destroy()
|
||||
window.after_cancel.assert_not_called()
|
||||
window.destroy.assert_called_once()
|
||||
|
||||
|
||||
def test_destroy_logs_a_tcl_error_instead_of_raising(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
import tkinter as tk
|
||||
|
||||
strip, window, _title, _lines = _strip()
|
||||
window.destroy.side_effect = tk.TclError("already gone")
|
||||
strip.destroy()
|
||||
assert "could not be destroyed" in caplog.text
|
||||
assert strip.closed
|
||||
|
||||
|
||||
def test_build_with_zero_lines_builds_only_the_title() -> None:
|
||||
window = MagicMock()
|
||||
window.winfo_reqheight.return_value = 40
|
||||
with (
|
||||
patch.object(corner_strip.tk, "Toplevel", return_value=window),
|
||||
patch.object(corner_strip.tk, "Label", side_effect=lambda *a, **k: MagicMock()),
|
||||
):
|
||||
strip = build_corner_strip(
|
||||
MagicMock(),
|
||||
LockConfig(),
|
||||
StripContent("Just a title", ()),
|
||||
rect=OutputRect(0, 0, 1920, 1080),
|
||||
max_lines=0,
|
||||
)
|
||||
assert isinstance(strip, CornerStrip)
|
||||
window.deiconify.assert_called_once()
|
||||
@ -153,3 +153,12 @@ class TestBootRetry:
|
||||
is None
|
||||
)
|
||||
assert naps == []
|
||||
|
||||
|
||||
def test_the_arming_wait_outlasts_a_slow_boot_network() -> None:
|
||||
"""2026-11-07: DNS came up ~5 s after boot and the refresher's next try was
|
||||
the 2-minute tick, so a 30 s wait locked kuhy mid-session. The budget must
|
||||
cover the refresher's own 150 s in-process retry."""
|
||||
from gatelock import morning_session
|
||||
|
||||
assert morning_session.MORNING_RETRY_SECONDS >= 180
|
||||
|
||||
@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "gatelock"
|
||||
version = "0.11.0"
|
||||
version = "0.12.0"
|
||||
description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps"
|
||||
requires-python = ">=3.14"
|
||||
dependencies = [] # pure stdlib — tkinter is bundled with Python
|
||||
|
||||
Loading…
Reference in New Issue
Block a user