mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 19:07:17 +01:00
gatelock: split _recovery.py and _surfaces.py, and fix two fail-open guards
Some checks failed
gatelock pre-commit / pre-commit (push) Waiting to run
gatelock tests / test (3.10) (push) Waiting to run
gatelock tests / test (3.11) (push) Waiting to run
gatelock tests / test (3.12) (push) Waiting to run
palette drift / palette (push) Waiting to run
sync_settings_ui tests / test (push) Waiting to run
github_device_auth tests / test (push) Has been cancelled
Some checks failed
gatelock pre-commit / pre-commit (push) Waiting to run
gatelock tests / test (3.10) (push) Waiting to run
gatelock tests / test (3.11) (push) Waiting to run
gatelock tests / test (3.12) (push) Waiting to run
palette drift / palette (push) Waiting to run
sync_settings_ui tests / test (push) Waiting to run
github_device_auth tests / test (push) Has been cancelled
_recovery.py 321 -> 242 + _recovery_types 47 + _grabwatch 84
_surfaces.py 380 -> 250 + _placement 136 + _surface_types 54 + _textmirror 70
Two guards had to be repaired first, because both would have gone quiet
rather than red:
1. palette-drift CI was ALREADY BROKEN by the earlier _window.py split.
palette_check.py reads LockConfig's colour defaults from a hardcoded
gatelock/_window.py, and LockConfig had moved to _config.py — so it parsed
0 tk tokens and reported "10 failing check(s)" where HEAD reports 0. Fixed
the path, and made parse_tk raise instead of returning {} when LockConfig
is not found: the empty dict is what let every tk comparison pass
vacuously while the drift guard was off for the whole stack.
2. test_recovery_invariants' symbols() read one hardcoded filename, so
splitting _recovery.py would have narrowed the "no weakening call"
invariant — the one that exists because of the 2026-07-25 unlock — to
whichever half kept the name, with the suite still green. It now globs the
whole _recovery*/_surfaces* family and fails if the glob matches nothing.
Verified it discriminates: a throwaway _recovery_probe.py calling
restore_vt_switching made the test fail, and removing it restored green.
GrabWatch is a small object rather than a free function because holds_grab
and its once-per-cause warning share state; _placement holds the geometry
questions and the re-assertion pass, which can only ever add visibility.
_surfaces keeps an explicit __all__: the autofixer pruned the mirror_text_widgets
re-export once and `import gatelock` broke, so the intent is now stated.
389 passed at 100% branch coverage; gatelock.__all__ is the same 58 names as
HEAD; palette_check back to 0 failing checks.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG
This commit is contained in:
parent
085d08b5c9
commit
02659e0abf
84
gatelock/gatelock/_grabwatch.py
Normal file
84
gatelock/gatelock/_grabwatch.py
Normal file
@ -0,0 +1,84 @@
|
||||
"""Deciding whether this application still holds the input grab.
|
||||
|
||||
Split from :mod:`gatelock._recovery`, which re-asserts the grab when this
|
||||
says it is gone. The question is subtle enough -- and its history costly
|
||||
enough -- to be worth reading on its own, and the log-throttling state it
|
||||
needs is the reason it is a small object rather than a function.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import tkinter as tk
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GrabWatch:
|
||||
"""Answers "is the grab still ours?", warning once per distinct cause."""
|
||||
|
||||
def __init__(self, root: tk.Misc) -> None:
|
||||
"""Watch the grab on `root`'s display.
|
||||
|
||||
Args:
|
||||
root: The Tk widget whose display the grab is read from.
|
||||
"""
|
||||
self._root = root
|
||||
self._last_warning: str | None = None
|
||||
|
||||
def holds_grab(self) -> bool:
|
||||
"""Whether the grab is held by a window this application owns.
|
||||
|
||||
Deliberately *not* ``grab_current() is self._root``. Our own transient
|
||||
children take the grab for themselves: a posted Tk menu grabs the menu
|
||||
window, and the old identity test read that as "the grab was lost" and
|
||||
called ``grab_set_global()`` a second later, stealing it back off our
|
||||
own widget mid-interaction. That is what made the screen-locker sport
|
||||
selector unusable while locked on 2026-07-26. Any window we own still
|
||||
means the lock has the pointer and keyboard.
|
||||
|
||||
``grab_current`` resolves the name Tcl gives it against our own widget
|
||||
map, so anything that does not resolve -- a Tcl-created popdown such as
|
||||
a ``ttk.Combobox``'s, or a stale name -- raises ``KeyError`` and is NOT
|
||||
provably ours. That counts as lost and the grab is re-taken: fail
|
||||
closed, because this module exists to keep the lock, and losing a
|
||||
popdown beats believing we hold a grab we do not.
|
||||
|
||||
Known limitation: a second ``tk.Tk()`` in the same process reports its
|
||||
root as ``"."`` too, which is indistinguishable from ours. The old
|
||||
identity test had exactly the same blind spot. ``_heat_skip``'s
|
||||
throwaway root is the only other one in screen-locker, and it is
|
||||
destroyed before the lock window exists.
|
||||
|
||||
Returns:
|
||||
True while a window we own holds the grab.
|
||||
"""
|
||||
try:
|
||||
held = self._root.grab_current() is not None
|
||||
except KeyError as exc:
|
||||
self.warn_once(
|
||||
f"the grab is held by {exc.args[0]!r}, not one of our windows"
|
||||
)
|
||||
return False
|
||||
except tk.TclError:
|
||||
self.warn_once("could not read the current grab")
|
||||
return False
|
||||
if held:
|
||||
self._last_warning = None
|
||||
return held
|
||||
|
||||
def warn_once(self, reason: str) -> None:
|
||||
"""Warn about a lost grab, but only when the reason changes.
|
||||
|
||||
These states can persist for the whole lock (a dark display, a popdown
|
||||
that keeps re-grabbing), and this runs once a second -- warning every
|
||||
tick would push a line per second into the journal for as long as it
|
||||
lasts. Silence is not the alternative: the first occurrence and every
|
||||
change of cause are still logged at WARNING.
|
||||
|
||||
Args:
|
||||
reason: Why the grab is not provably ours.
|
||||
"""
|
||||
if reason != self._last_warning:
|
||||
_logger.warning("%s; treating it as lost and re-taking it", reason)
|
||||
self._last_warning = reason
|
||||
136
gatelock/gatelock/_placement.py
Normal file
136
gatelock/gatelock/_placement.py
Normal file
@ -0,0 +1,136 @@
|
||||
"""Where a lock surface sits, and putting it back when it drifts.
|
||||
|
||||
Split from :mod:`gatelock._surfaces`, which owns creating and destroying the
|
||||
windows. This half owns the geometry questions -- is the window mapped, is it
|
||||
still exactly on its output -- and the re-assertion pass that answers "no" by
|
||||
correcting it.
|
||||
|
||||
Everything here only ever *adds* visibility. Nothing in this module can hide
|
||||
or unmap a surface; that asymmetry is what makes the recovery loop monotonic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Mapping
|
||||
|
||||
from gatelock._config import LockConfig
|
||||
|
||||
from gatelock._outputs import OutputRect
|
||||
from gatelock._surfaces import SurfaceInfo
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Surface:
|
||||
"""One Toplevel and the output it is pinned to."""
|
||||
|
||||
info: SurfaceInfo
|
||||
window: tk.Toplevel
|
||||
built: bool = field(default=False)
|
||||
|
||||
|
||||
def set_geometry(window: tk.Toplevel, rect: OutputRect) -> None:
|
||||
"""Pin a window to an output's rectangle."""
|
||||
window.geometry(rect.geometry())
|
||||
|
||||
|
||||
def is_mapped(surface: _Surface) -> bool:
|
||||
"""Whether a surface's window is currently mapped."""
|
||||
return bool(surface.window.winfo_ismapped())
|
||||
|
||||
|
||||
def geometry_matches(surface: _Surface) -> bool:
|
||||
"""Whether a surface still sits exactly on its output."""
|
||||
rect = surface.info.rect
|
||||
window = surface.window
|
||||
return (
|
||||
window.winfo_rootx() == rect.x
|
||||
and window.winfo_rooty() == rect.y
|
||||
and window.winfo_width() == rect.width
|
||||
and window.winfo_height() == rect.height
|
||||
)
|
||||
|
||||
|
||||
def enforce_one(surface: _Surface, *, overrideredirect: bool) -> bool:
|
||||
"""Re-assert one surface's placement and stacking.
|
||||
|
||||
Args:
|
||||
surface: The surface to correct.
|
||||
overrideredirect: Whether the lock runs WM-unmanaged, in which case
|
||||
the flag is re-set -- it can be dropped across an unmap/map cycle,
|
||||
and a managed lock window would be moved by the WM.
|
||||
|
||||
Returns:
|
||||
True if anything needed correcting.
|
||||
"""
|
||||
window = surface.window
|
||||
changed = False
|
||||
if not is_mapped(surface):
|
||||
window.deiconify()
|
||||
changed = True
|
||||
if not geometry_matches(surface):
|
||||
set_geometry(window, surface.info.rect)
|
||||
changed = True
|
||||
if overrideredirect:
|
||||
window.overrideredirect(boolean=True)
|
||||
window.lift()
|
||||
return changed
|
||||
|
||||
|
||||
def blind_outputs(
|
||||
surfaces: Mapping[str, _Surface], names: frozenset[str]
|
||||
) -> tuple[str, ...]:
|
||||
"""Live outputs that are NOT currently showing a lock.
|
||||
|
||||
This is the alarm for the original defect: an output the user can see,
|
||||
with no lock on it, means the escape UI is somewhere they cannot reach.
|
||||
|
||||
Args:
|
||||
surfaces: The live surfaces, keyed by output name.
|
||||
names: The outputs that should be covered.
|
||||
|
||||
Returns:
|
||||
The uncovered output names, sorted.
|
||||
"""
|
||||
blind: list[str] = []
|
||||
for name in sorted(names):
|
||||
surface = surfaces.get(name)
|
||||
if surface is None or not is_mapped(surface):
|
||||
blind.append(name)
|
||||
return tuple(blind)
|
||||
|
||||
|
||||
def enforce_all(
|
||||
surfaces: Mapping[str, _Surface], *, overrideredirect: bool
|
||||
) -> tuple[str, ...]:
|
||||
"""Re-assert every surface's placement and stacking.
|
||||
|
||||
Args:
|
||||
surfaces: The live surfaces, keyed by output name.
|
||||
overrideredirect: Whether the lock runs WM-unmanaged.
|
||||
|
||||
Returns:
|
||||
The names of the outputs whose surfaces needed correcting, sorted.
|
||||
"""
|
||||
corrected = [
|
||||
surface.info.output_name
|
||||
for surface in surfaces.values()
|
||||
if enforce_one(surface, overrideredirect=overrideredirect)
|
||||
]
|
||||
return tuple(sorted(corrected))
|
||||
|
||||
|
||||
def needs_backdrop_root(config: LockConfig) -> bool:
|
||||
"""Whether the root must stay mapped as a backdrop and grab holder.
|
||||
|
||||
Deliberately *not* simplified to "iff override-redirect".
|
||||
``LockConfig(mode="soft", grab="global")`` is a reachable combination, and
|
||||
a global grab requires a viewable window -- X refuses to grab for a
|
||||
withdrawn one. Collapsing this predicate would make that config fail at
|
||||
runtime only.
|
||||
"""
|
||||
return config.resolved_overrideredirect() or config.resolved_grab() != "none"
|
||||
@ -36,54 +36,19 @@ keeps the grab. Pulling a cable is therefore not an escape.
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from gatelock._grabwatch import GrabWatch
|
||||
from gatelock._recovery_types import RecoveryCollaborators, RecoveryReport
|
||||
from gatelock._vt import disable_vt_switching
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gatelock._detect import OutputChangeDetector
|
||||
from gatelock._outputs import OutputEnumerator
|
||||
from gatelock._surfaces import SurfaceDelta, SurfaceSet
|
||||
from gatelock._window import LockConfig, LockWindowHooks
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
_VT_REASSERT_EVERY = 30
|
||||
"""Re-disable VT switching every N verify ticks, in case something reset it."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RecoveryReport:
|
||||
"""What one tick observed and corrected."""
|
||||
|
||||
scan_ok: bool
|
||||
source: str
|
||||
live_outputs: tuple[str, ...] = ()
|
||||
delta: SurfaceDelta | None = None
|
||||
corrected: tuple[str, ...] = ()
|
||||
grab_reasserted: bool = False
|
||||
vt_reasserted: bool = False
|
||||
blind: tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RecoveryCollaborators:
|
||||
"""The pieces one tick re-asserts over.
|
||||
|
||||
Bundled to keep the loop's constructor at one object instead of a
|
||||
growing, unbounded arg list.
|
||||
"""
|
||||
|
||||
config: LockConfig
|
||||
surfaces: SurfaceSet
|
||||
enumerator: OutputEnumerator
|
||||
detector: OutputChangeDetector
|
||||
hooks: LockWindowHooks
|
||||
|
||||
|
||||
class RecoveryLoop:
|
||||
"""Re-asserts the lock's coverage on a timer and on change events."""
|
||||
|
||||
@ -99,7 +64,7 @@ class RecoveryLoop:
|
||||
self._ticks = 0
|
||||
self._drain_job: str | None = None
|
||||
self._verify_job: str | None = None
|
||||
self._last_grab_warning: str | None = None
|
||||
self._grab = GrabWatch(root)
|
||||
|
||||
@property
|
||||
def ticks(self) -> int:
|
||||
@ -238,43 +203,12 @@ class RecoveryLoop:
|
||||
def holds_grab(self) -> bool:
|
||||
"""Whether the grab is held by a window this application owns.
|
||||
|
||||
Deliberately *not* ``grab_current() is self._root``. Our own transient
|
||||
children take the grab for themselves: a posted Tk menu grabs the menu
|
||||
window, and the old identity test read that as "the grab was lost" and
|
||||
called ``grab_set_global()`` a second later, stealing it back off our
|
||||
own widget mid-interaction. That is what made the screen-locker sport
|
||||
selector unusable while locked on 2026-07-26. Any window we own still
|
||||
means the lock has the pointer and keyboard.
|
||||
|
||||
``grab_current`` resolves the name Tcl gives it against our own widget
|
||||
map, so anything that does not resolve -- a Tcl-created popdown such as
|
||||
a ``ttk.Combobox``'s, or a stale name -- raises ``KeyError`` and is NOT
|
||||
provably ours. That counts as lost and the grab is re-taken: fail
|
||||
closed, because this module exists to keep the lock, and losing a
|
||||
popdown beats believing we hold a grab we do not.
|
||||
|
||||
Known limitation: a second ``tk.Tk()`` in the same process reports its
|
||||
root as ``"."`` too, which is indistinguishable from ours. The old
|
||||
identity test had exactly the same blind spot. ``_heat_skip``'s
|
||||
throwaway root is the only other one in screen-locker, and it is
|
||||
destroyed before the lock window exists.
|
||||
|
||||
Public so an embedder (or a verification harness) can ask the same
|
||||
question the loop asks, rather than re-implementing it and drifting.
|
||||
See :class:`gatelock._grabwatch.GrabWatch` for why the test is not a
|
||||
simple identity check.
|
||||
"""
|
||||
try:
|
||||
held = self._root.grab_current() is not None
|
||||
except KeyError as exc:
|
||||
self._warn_once(
|
||||
f"the grab is held by {exc.args[0]!r}, not one of our windows"
|
||||
)
|
||||
return False
|
||||
except tk.TclError:
|
||||
self._warn_once("could not read the current grab")
|
||||
return False
|
||||
if held:
|
||||
self._last_grab_warning = None
|
||||
return held
|
||||
return self._grab.holds_grab()
|
||||
|
||||
def _reassert_focus(self) -> None:
|
||||
"""Re-focus the preferred surface after any output came back live.
|
||||
@ -294,19 +228,6 @@ class RecoveryLoop:
|
||||
index = self._surfaces.preferred_focus_index()
|
||||
self._hooks.on_focus_ready(self._surfaces.focus_surface(index))
|
||||
|
||||
def _warn_once(self, reason: str) -> None:
|
||||
"""Warn about a lost grab, but only when the reason changes.
|
||||
|
||||
These states can persist for the whole lock (a dark display, a popdown
|
||||
that keeps re-grabbing), and this runs once a second -- warning every
|
||||
tick would push a line per second into the journal for as long as it
|
||||
lasts. Silence is not the alternative: the first occurrence and every
|
||||
change of cause are still logged at WARNING.
|
||||
"""
|
||||
if reason != self._last_grab_warning:
|
||||
_logger.warning("%s; treating it as lost and re-taking it", reason)
|
||||
self._last_grab_warning = reason
|
||||
|
||||
def _reassert_vt(self) -> bool:
|
||||
"""Periodically re-disable VT switching. Never re-enables it.
|
||||
|
||||
|
||||
47
gatelock/gatelock/_recovery_types.py
Normal file
47
gatelock/gatelock/_recovery_types.py
Normal file
@ -0,0 +1,47 @@
|
||||
"""The value types one recovery tick produces and consumes.
|
||||
|
||||
Split from :mod:`gatelock._recovery` for the 250-line cap. Data only -- no
|
||||
behaviour, and deliberately no import of the loop, so a caller can type
|
||||
against a report without pulling in the machinery that builds one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gatelock._config import LockConfig
|
||||
from gatelock._detect import OutputChangeDetector
|
||||
from gatelock._hooks import LockWindowHooks
|
||||
from gatelock._outputs import OutputEnumerator
|
||||
from gatelock._surfaces import SurfaceDelta, SurfaceSet
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RecoveryReport:
|
||||
"""What one tick observed and corrected."""
|
||||
|
||||
scan_ok: bool
|
||||
source: str
|
||||
live_outputs: tuple[str, ...] = ()
|
||||
delta: SurfaceDelta | None = None
|
||||
corrected: tuple[str, ...] = ()
|
||||
grab_reasserted: bool = False
|
||||
vt_reasserted: bool = False
|
||||
blind: tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RecoveryCollaborators:
|
||||
"""The pieces one tick re-asserts over.
|
||||
|
||||
Bundled to keep the loop's constructor at one object instead of a
|
||||
growing, unbounded arg list.
|
||||
"""
|
||||
|
||||
config: LockConfig
|
||||
surfaces: SurfaceSet
|
||||
enumerator: OutputEnumerator
|
||||
detector: OutputChangeDetector
|
||||
hooks: LockWindowHooks
|
||||
54
gatelock/gatelock/_surface_types.py
Normal file
54
gatelock/gatelock/_surface_types.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""The value types describing a lock surface and a change to the set of them.
|
||||
|
||||
Split from :mod:`gatelock._surfaces` for the 250-line cap. Data and one
|
||||
Protocol, no behaviour, so an app can type against a surface without
|
||||
importing the machinery that builds one.
|
||||
|
||||
Re-exported from :mod:`gatelock._surfaces`, so existing imports of these
|
||||
names keep working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Protocol
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import tkinter as tk
|
||||
|
||||
from gatelock._outputs import OutputRect
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SurfaceInfo:
|
||||
"""Identifies one lock surface and the output it covers."""
|
||||
|
||||
output_name: str
|
||||
rect: OutputRect
|
||||
index: int
|
||||
is_primary: bool
|
||||
|
||||
|
||||
class SurfaceBuilder(Protocol):
|
||||
"""What a consuming app must implement to paint its UI on each output."""
|
||||
|
||||
def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None:
|
||||
"""Build this app's widgets inside ``parent`` for one output."""
|
||||
|
||||
def teardown_surface(self, surface: SurfaceInfo) -> None:
|
||||
"""Forget any per-surface state; the window is about to be destroyed."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SurfaceDelta:
|
||||
"""What one :meth:`SurfaceSet.apply` changed."""
|
||||
|
||||
created: tuple[str, ...] = ()
|
||||
moved: tuple[str, ...] = ()
|
||||
removed: tuple[str, ...] = ()
|
||||
unverified: tuple[str, ...] = ()
|
||||
|
||||
@property
|
||||
def changed(self) -> bool:
|
||||
"""Whether anything structural happened."""
|
||||
return bool(self.created or self.moved or self.removed)
|
||||
@ -34,15 +34,39 @@ recording, because each rules out an approach that looks reasonable:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
import logging
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING, Protocol
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from gatelock._placement import (
|
||||
_Surface,
|
||||
blind_outputs,
|
||||
needs_backdrop_root,
|
||||
enforce_all,
|
||||
set_geometry,
|
||||
)
|
||||
from gatelock._surface_types import (
|
||||
SurfaceBuilder,
|
||||
SurfaceDelta,
|
||||
SurfaceInfo,
|
||||
)
|
||||
from gatelock._textmirror import TextMirror, mirror_text_widgets
|
||||
|
||||
# Names this module re-exports after the 250-line split. Listed explicitly so
|
||||
# the autofixer cannot prune an import that exists *for* its re-export -- it
|
||||
# stripped mirror_text_widgets once, and `gatelock` itself stopped importing.
|
||||
__all__ = [
|
||||
"SurfaceBuilder",
|
||||
"SurfaceDelta",
|
||||
"SurfaceInfo",
|
||||
"SurfaceSet",
|
||||
"TextMirror",
|
||||
"mirror_text_widgets",
|
||||
"needs_backdrop_root",
|
||||
]
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
|
||||
from gatelock._outputs import OutputRect, OutputScan
|
||||
from gatelock._outputs import OutputScan
|
||||
from gatelock._window import LockConfig
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
@ -52,117 +76,6 @@ _logger = logging.getLogger(__name__)
|
||||
# and the type stubs, which have no keyword overload.
|
||||
_TOPMOST_ON = True
|
||||
|
||||
_TEXT_START = "1.0"
|
||||
_TEXT_END = "end-1c"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SurfaceInfo:
|
||||
"""Identifies one lock surface and the output it covers."""
|
||||
|
||||
output_name: str
|
||||
rect: OutputRect
|
||||
index: int
|
||||
is_primary: bool
|
||||
|
||||
|
||||
class SurfaceBuilder(Protocol):
|
||||
"""What a consuming app must implement to paint its UI on each output."""
|
||||
|
||||
def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None:
|
||||
"""Build this app's widgets inside ``parent`` for one output."""
|
||||
|
||||
def teardown_surface(self, surface: SurfaceInfo) -> None:
|
||||
"""Forget any per-surface state; the window is about to be destroyed."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SurfaceDelta:
|
||||
"""What one :meth:`SurfaceSet.apply` changed."""
|
||||
|
||||
created: tuple[str, ...] = ()
|
||||
moved: tuple[str, ...] = ()
|
||||
removed: tuple[str, ...] = ()
|
||||
unverified: tuple[str, ...] = ()
|
||||
|
||||
@property
|
||||
def changed(self) -> bool:
|
||||
"""Whether anything structural happened."""
|
||||
return bool(self.created or self.moved or self.removed)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Surface:
|
||||
"""One Toplevel and the output it is pinned to."""
|
||||
|
||||
info: SurfaceInfo
|
||||
window: tk.Toplevel
|
||||
built: bool = field(default=False)
|
||||
|
||||
|
||||
def needs_backdrop_root(config: LockConfig) -> bool:
|
||||
"""Whether the root must stay mapped as a backdrop and grab holder.
|
||||
|
||||
Deliberately *not* simplified to "iff override-redirect".
|
||||
``LockConfig(mode="soft", grab="global")`` is a reachable combination, and
|
||||
a global grab requires a viewable window -- X refuses to grab for a
|
||||
withdrawn one. Collapsing this predicate would make that config fail at
|
||||
runtime only.
|
||||
"""
|
||||
return config.resolved_overrideredirect() or config.resolved_grab() != "none"
|
||||
|
||||
|
||||
class TextMirror:
|
||||
"""Keeps N ``tk.Text`` widgets showing the same value.
|
||||
|
||||
``tk.Text`` has no ``textvariable``, so it is the one widget that cannot
|
||||
ride the shared-``StringVar`` mechanism the rest of the mirroring uses.
|
||||
"""
|
||||
|
||||
def __init__(self, widgets: Sequence[tk.Text], var: tk.StringVar) -> None:
|
||||
"""Bind ``widgets`` to each other through ``var``."""
|
||||
self._widgets = list(widgets)
|
||||
self._var = var
|
||||
self._syncing = False
|
||||
for widget in self._widgets:
|
||||
widget.bind("<KeyRelease>", self._on_key_release, add="+")
|
||||
self._var.trace_add("write", self._on_var_write)
|
||||
|
||||
def _on_key_release(self, event: tk.Event[tk.Text]) -> None:
|
||||
"""Push the edited widget's content into the shared variable."""
|
||||
if self._syncing:
|
||||
return
|
||||
self._syncing = True
|
||||
try:
|
||||
self._var.set(event.widget.get(_TEXT_START, _TEXT_END))
|
||||
finally:
|
||||
self._syncing = False
|
||||
self._fan_out(source=event.widget)
|
||||
|
||||
def _on_var_write(self, *_args: str) -> None:
|
||||
"""Push the shared variable's value into every widget."""
|
||||
if self._syncing:
|
||||
return
|
||||
self._fan_out(source=None)
|
||||
|
||||
def _fan_out(self, *, source: tk.Text | None) -> None:
|
||||
"""Write the variable's value into every widget but ``source``."""
|
||||
value = self._var.get()
|
||||
self._syncing = True
|
||||
try:
|
||||
for widget in self._widgets:
|
||||
if widget is source or widget.get(_TEXT_START, _TEXT_END) == value:
|
||||
continue
|
||||
widget.delete(_TEXT_START, "end")
|
||||
widget.insert(_TEXT_START, value)
|
||||
finally:
|
||||
self._syncing = False
|
||||
|
||||
|
||||
def mirror_text_widgets(widgets: Sequence[tk.Text], var: tk.StringVar) -> TextMirror:
|
||||
"""Keep several ``tk.Text`` mirrors in sync through one variable."""
|
||||
return TextMirror(widgets, var)
|
||||
|
||||
|
||||
class SurfaceSet:
|
||||
"""The full set of lock windows, kept in step with the live outputs."""
|
||||
@ -252,12 +165,7 @@ class SurfaceSet:
|
||||
with no lock on it, means the escape UI is somewhere they cannot reach.
|
||||
"""
|
||||
names = live_names if live_names is not None else self.names()
|
||||
blind: list[str] = []
|
||||
for name in sorted(names):
|
||||
surface = self._surfaces.get(name)
|
||||
if surface is None or not self._is_mapped(surface):
|
||||
blind.append(name)
|
||||
return tuple(blind)
|
||||
return blind_outputs(self._surfaces, names)
|
||||
|
||||
def enforce(self) -> tuple[str, ...]:
|
||||
"""Re-assert every surface's placement and stacking.
|
||||
@ -265,29 +173,10 @@ class SurfaceSet:
|
||||
Only ever *adds* visibility: remaps what became unmapped, corrects
|
||||
drifted geometry, and raises. Nothing here can hide a surface.
|
||||
"""
|
||||
corrected = [
|
||||
surface.info.output_name
|
||||
for surface in self._surfaces.values()
|
||||
if self._enforce_one(surface)
|
||||
]
|
||||
return tuple(sorted(corrected))
|
||||
|
||||
def _enforce_one(self, surface: _Surface) -> bool:
|
||||
"""Re-assert one surface. True if anything needed correcting."""
|
||||
window = surface.window
|
||||
changed = False
|
||||
if not self._is_mapped(surface):
|
||||
window.deiconify()
|
||||
changed = True
|
||||
if not self._geometry_matches(surface):
|
||||
self._set_geometry(window, surface.info.rect)
|
||||
changed = True
|
||||
# Re-assert override-redirect: it can be dropped across an unmap/map
|
||||
# cycle, and a managed lock window would be moved by the WM.
|
||||
if self._config.resolved_overrideredirect():
|
||||
window.overrideredirect(boolean=True)
|
||||
window.lift()
|
||||
return changed
|
||||
return enforce_all(
|
||||
self._surfaces,
|
||||
overrideredirect=self._config.resolved_overrideredirect(),
|
||||
)
|
||||
|
||||
def raise_all(self) -> None:
|
||||
"""Lift every surface above whatever else is on screen."""
|
||||
@ -321,7 +210,7 @@ class SurfaceSet:
|
||||
if self._config.resolved_overrideredirect():
|
||||
window.overrideredirect(boolean=True)
|
||||
window.configure(bg=self._config.bg, cursor="arrow")
|
||||
self._set_geometry(window, info.rect)
|
||||
set_geometry(window, info.rect)
|
||||
if not self._config.resolved_overrideredirect():
|
||||
window.attributes("-topmost", _TOPMOST_ON)
|
||||
window.deiconify()
|
||||
@ -342,7 +231,7 @@ class SurfaceSet:
|
||||
not discard half-typed input.
|
||||
"""
|
||||
surface.info = info
|
||||
self._set_geometry(surface.window, info.rect)
|
||||
set_geometry(surface.window, info.rect)
|
||||
surface.window.lift()
|
||||
_logger.info(
|
||||
"lock surface on %s moved to %s", info.output_name, info.rect.geometry()
|
||||
@ -359,22 +248,3 @@ class SurfaceSet:
|
||||
"""Tear down every surface. Used only when the whole lock is closing."""
|
||||
for name in list(self._surfaces):
|
||||
self._destroy(name)
|
||||
|
||||
def _set_geometry(self, window: tk.Toplevel, rect: OutputRect) -> None:
|
||||
"""Pin a window to an output's rectangle."""
|
||||
window.geometry(rect.geometry())
|
||||
|
||||
def _is_mapped(self, surface: _Surface) -> bool:
|
||||
"""Whether a surface's window is currently mapped."""
|
||||
return bool(surface.window.winfo_ismapped())
|
||||
|
||||
def _geometry_matches(self, surface: _Surface) -> bool:
|
||||
"""Whether a surface still sits exactly on its output."""
|
||||
rect = surface.info.rect
|
||||
window = surface.window
|
||||
return (
|
||||
window.winfo_rootx() == rect.x
|
||||
and window.winfo_rooty() == rect.y
|
||||
and window.winfo_width() == rect.width
|
||||
and window.winfo_height() == rect.height
|
||||
)
|
||||
|
||||
75
gatelock/gatelock/_textmirror.py
Normal file
75
gatelock/gatelock/_textmirror.py
Normal file
@ -0,0 +1,75 @@
|
||||
"""Keeping several Text widgets showing the same string.
|
||||
|
||||
Split from :mod:`gatelock._surfaces` for the 250-line cap. A lock renders one
|
||||
surface per output, so a multi-line field exists once per monitor and all the
|
||||
copies have to agree; ``tk.StringVar`` does this for Entry and Spinbox but not
|
||||
for ``tk.Text``, which has no textvariable.
|
||||
|
||||
Re-exported from :mod:`gatelock._surfaces` and from :mod:`gatelock`, so
|
||||
existing imports of either name keep working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
|
||||
# Tk text indices: the first character, and the last one before the newline
|
||||
# Text always appends.
|
||||
_TEXT_START = "1.0"
|
||||
_TEXT_END = "end-1c"
|
||||
|
||||
|
||||
class TextMirror:
|
||||
"""Keeps N ``tk.Text`` widgets showing the same value.
|
||||
|
||||
``tk.Text`` has no ``textvariable``, so it is the one widget that cannot
|
||||
ride the shared-``StringVar`` mechanism the rest of the mirroring uses.
|
||||
"""
|
||||
|
||||
def __init__(self, widgets: Sequence[tk.Text], var: tk.StringVar) -> None:
|
||||
"""Bind ``widgets`` to each other through ``var``."""
|
||||
self._widgets = list(widgets)
|
||||
self._var = var
|
||||
self._syncing = False
|
||||
for widget in self._widgets:
|
||||
widget.bind("<KeyRelease>", self._on_key_release, add="+")
|
||||
self._var.trace_add("write", self._on_var_write)
|
||||
|
||||
def _on_key_release(self, event: tk.Event[tk.Text]) -> None:
|
||||
"""Push the edited widget's content into the shared variable."""
|
||||
if self._syncing:
|
||||
return
|
||||
self._syncing = True
|
||||
try:
|
||||
self._var.set(event.widget.get(_TEXT_START, _TEXT_END))
|
||||
finally:
|
||||
self._syncing = False
|
||||
self._fan_out(source=event.widget)
|
||||
|
||||
def _on_var_write(self, *_args: str) -> None:
|
||||
"""Push the shared variable's value into every widget."""
|
||||
if self._syncing:
|
||||
return
|
||||
self._fan_out(source=None)
|
||||
|
||||
def _fan_out(self, *, source: tk.Text | None) -> None:
|
||||
"""Write the variable's value into every widget but ``source``."""
|
||||
value = self._var.get()
|
||||
self._syncing = True
|
||||
try:
|
||||
for widget in self._widgets:
|
||||
if widget is source or widget.get(_TEXT_START, _TEXT_END) == value:
|
||||
continue
|
||||
widget.delete(_TEXT_START, "end")
|
||||
widget.insert(_TEXT_START, value)
|
||||
finally:
|
||||
self._syncing = False
|
||||
|
||||
|
||||
def mirror_text_widgets(widgets: Sequence[tk.Text], var: tk.StringVar) -> TextMirror:
|
||||
"""Keep several ``tk.Text`` mirrors in sync through one variable."""
|
||||
return TextMirror(widgets, var)
|
||||
@ -4,7 +4,6 @@ Split from ``test_escape.py`` (250-line cap), which keeps policy defaults,
|
||||
timestamp parsing, loading, and the integrity guarantee.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
@ -5,8 +5,6 @@ keeps the python-xlib backend itself; this covers the enumerator that drives
|
||||
it and the OutputScan it produces.
|
||||
"""
|
||||
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@ -4,7 +4,6 @@ Split from ``test_outputs.py`` (250-line cap), which keeps rect geometry,
|
||||
the live predicate, and xrandr-query parsing.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@ -18,11 +18,24 @@ SOURCE_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def symbols(module: str) -> set[str]:
|
||||
"""Every attribute and name referenced in a module's source."""
|
||||
tree = ast.parse((SOURCE_ROOT / module).read_text(encoding="utf-8"))
|
||||
return {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)} | {
|
||||
n.id for n in ast.walk(tree) if isinstance(n, ast.Name)
|
||||
}
|
||||
"""Every attribute and name referenced across the `module` family.
|
||||
|
||||
Globbed rather than read as one exact filename, so that splitting a module
|
||||
for the 250-line cap cannot narrow the invariant to whichever half kept the
|
||||
original name. A new `_recovery_*.py` sibling is covered the moment it
|
||||
exists; nobody has to remember to extend a list here.
|
||||
"""
|
||||
stem = module.removesuffix(".py")
|
||||
paths = sorted(SOURCE_ROOT.glob(f"{stem}*.py"))
|
||||
if not paths:
|
||||
msg = f"no module matching {stem}*.py -- renamed? the guard is off"
|
||||
raise AssertionError(msg)
|
||||
found: set[str] = set()
|
||||
for path in paths:
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
found |= {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)}
|
||||
found |= {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}
|
||||
return found
|
||||
|
||||
|
||||
class TestStaticInvariant:
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
Split from ``test_surfaces.py`` to hold the shared 250-line cap.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
Split from ``test_window.py`` to hold the shared 250-line cap.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tkinter as tk
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
Split from ``test_window_surfaces.py`` to hold the shared 250-line cap.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
|
||||
@ -4,7 +4,6 @@ Split from ``test_window_surfaces.py`` to hold the shared 250-line cap;
|
||||
that file keeps arming without a display and the grab-blocked logging.
|
||||
"""
|
||||
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import signal
|
||||
|
||||
@ -37,7 +37,9 @@ ROOT = Path(__file__).resolve().parents[2]
|
||||
TOKENS_MD = ROOT / "unified-design-system" / "tokens.md"
|
||||
TOKENS_CSS = ROOT / "web_ui" / "src" / "tokens.css"
|
||||
TOKENS_DART = ROOT / "design_system" / "lib" / "src" / "tokens.dart"
|
||||
LOCKCONFIG_PY = ROOT / "gatelock" / "gatelock" / "_window.py"
|
||||
# LockConfig moved out of _window.py when that file was split for the
|
||||
# 250-line cap; the palette lives with the dataclass, not the window.
|
||||
LOCKCONFIG_PY = ROOT / "gatelock" / "gatelock" / "_config.py"
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -121,7 +123,16 @@ def parse_tk(text: str) -> dict[str, str]:
|
||||
found: dict[str, str] = {}
|
||||
start = text.find("class LockConfig")
|
||||
if start == -1:
|
||||
return found
|
||||
# Fail closed. Returning {} here made every tk comparison vacuously
|
||||
# pass when LockConfig moved to _config.py during the 250-line split:
|
||||
# "0 parsed token(s)" read as success, and the drift guard was off
|
||||
# for the whole gatelock stack without anything going red.
|
||||
msg = (
|
||||
"LockConfig not found in the file palette_check reads. It has "
|
||||
"moved; update LOCKCONFIG_PY rather than letting the tk stack go "
|
||||
"unchecked."
|
||||
)
|
||||
raise SystemExit(msg)
|
||||
body = text[start:]
|
||||
decl = re.compile(r'^\s{4}(\w+):\s*str\s*=\s*"(#[0-9A-Fa-f]{6})"', re.MULTILINE)
|
||||
for name, value in decl.findall(body):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user