kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 19:07:17 +01:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity

gatelock: scope pylint to source on 4.0.8 and clear the residue it grades

Browse Source 
Same config move as crdt-sync (ignore-paths, init-hook, the three
implicit-booleaness disables the reference repos carry). The gate stays
at 8.0 for now: LockConfig's 35-field record and the two fail-open
guards in the recovery loop are still open.

Everything else pylint had left on source is gone. The optional
python-xlib dependency is loaded through one importlib helper instead
of two function-scope import statements, keeping the "not installed"
arm reachable for branch coverage; RecoveryLoop and LockWindow hold
their collaborator bundles rather than unpacking them into a second
set of attributes; the claim types are imported from _claims, where
they are defined, so _arbiter no longer needs an export list that
mirrored the package's; runtime imports sit above TYPE_CHECKING
blocks; the RandR Protocol stubs say which Xlib call they mirror.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHCbkJWXHjtTUaqvnHRAqA
This commit is contained in:
Krzysztof Rudnicki 2026-09-12 19:15:39 +02:00
parent 8274a3e918
commit 1723df3885
20 changed files with 144 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gatelock/.pre-commit-config.yaml
View File

@ -101,7 +101,7 @@ repos:
- --disable-error-code=list-item
- repo: https://github.com/pylint-dev/pylint
rev: v3.3.2
rev: v4.0.8
hooks:
- id: pylint
args:

5
gatelock/gatelock/__init__.py
View File

@ -13,11 +13,8 @@ from gatelock._arbiter import (
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
ArbiterVerdict,
Claim,
default_runtime_dir,
grab_strength,
)
from gatelock._claims import ArbiterVerdict, Claim, default_runtime_dir, grab_strength
from gatelock._detect import OutputChangeDetector
from gatelock._escape import (
EscapeDraft,

15
gatelock/gatelock/_arbiter.py
View File

@ -42,7 +42,6 @@ from gatelock._claims import (
Claim,
_try_lock,
default_runtime_dir,
grab_strength,
live_claims,
read_claim_if_held,
)
@ -52,20 +51,6 @@ if TYPE_CHECKING:
_logger = logging.getLogger(__name__)
# Re-exported after the 250-line split: `from gatelock._arbiter import
# Claim` is used by _window, the tests and gatelock's own __init__.
__all__ = [
"RANK_DIET_GUARD",
"RANK_SCREEN_LOCKER",
"RANK_WAKE_ALARM",
"Arbiter",
"ArbiterVerdict",
"Claim",
"default_runtime_dir",
"grab_strength",
]
RANK_WAKE_ALARM: Final = 300
RANK_SCREEN_LOCKER: Final = 200
RANK_DIET_GUARD: Final = 100

9
gatelock/gatelock/_detect.py
View File

@ -32,6 +32,8 @@ import threading
import tkinter as tk
from typing import Any
from gatelock._xlib import load_xlib
_logger = logging.getLogger(__name__)
_SELECT_TIMEOUT_S = 0.5
@ -64,15 +66,14 @@ class _RandrEventSource:
def _connect(self) -> bool:
"""Open the connection and subscribe to layout changes."""
try:
from Xlib import display as xdisplay
from Xlib.ext import randr
except ImportError:
modules = load_xlib("Xlib.display", "Xlib.ext.randr")
if modules is None:
_logger.info(
"python-xlib is not installed; falling back to Tk <Configure> "
"plus xrandr polling for output-change detection"
)
return False
xdisplay, randr = modules
try:
self._display = xdisplay.Display()
root = self._display.screen().root

9
gatelock/gatelock/_escape.py
View File

@ -31,11 +31,6 @@ import json
import logging
from typing import TYPE_CHECKING, Any
from gatelock.log_integrity import compute_entry_hmac, verify_entry_hmac
if TYPE_CHECKING:
from pathlib import Path
from gatelock._escape_types import (
EscapeDraft,
EscapeHistory,
@ -43,6 +38,10 @@ from gatelock._escape_types import (
_parse_iso,
_today_iso,
)
from gatelock.log_integrity import compute_entry_hmac, verify_entry_hmac
if TYPE_CHECKING:
from pathlib import Path
_logger = logging.getLogger(__name__)

16
gatelock/gatelock/_outputs.py
View File

@ -17,11 +17,11 @@ Two backends compute the same predicate:
* **``xrandr --query``** text -- a re-derivation of the same predicate for
systems without ``python-xlib``.
``python-xlib`` is an optional dependency, so its import lives inside a
function rather than at module scope. That is deliberate and load-bearing: a
module-scope ``try/except ImportError`` is resolved at import time, which makes
the ``except`` arm permanently unreachable for branch coverage -- and this is
precisely the degradation path that must stay tested.
``python-xlib`` is an optional dependency, loaded on demand through
:mod:`gatelock._xlib` rather than at module scope. That is deliberate and
load-bearing: a module-scope ``try/except ImportError`` is resolved at import
time, which makes the ``except`` arm permanently unreachable for branch
coverage -- and this is precisely the degradation path that must stay tested.
"""
from __future__ import annotations
@ -30,13 +30,13 @@ import logging
import re
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import tkinter as tk
from gatelock._output_types import Output, OutputRect, OutputScan
from gatelock._randr import RandrBackend
from gatelock._xrandr import parse_xrandr_query, scan_xrandr
if TYPE_CHECKING:
import tkinter as tk
# Re-exported after the 250-line split, so `from gatelock._outputs import
# Output` -- which the tests and sibling modules do -- keeps resolving.
__all__ = [

3
gatelock/gatelock/_preempt.py
View File

@ -19,7 +19,8 @@ import signal
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from gatelock._arbiter import Arbiter, Claim
from gatelock._arbiter import Arbiter
from gatelock._claims import Claim
from gatelock._config import LockConfig
_logger = logging.getLogger(__name__)

3
gatelock/gatelock/_queue.py
View File

@ -40,7 +40,8 @@ QUEUE_HEARTBEAT_SECONDS: Final = 300.0
if TYPE_CHECKING:
from collections.abc import Callable
from gatelock._arbiter import Arbiter, Claim
from gatelock._arbiter import Arbiter
from gatelock._claims import Claim
@dataclass(frozen=True)

28
gatelock/gatelock/_randr.py
View File

@ -13,6 +13,7 @@ import logging
from typing import TYPE_CHECKING, Protocol
from gatelock._output_types import Output, OutputRect
from gatelock._xlib import load_xlib
if TYPE_CHECKING:
from collections.abc import Sequence
@ -41,23 +42,27 @@ if TYPE_CHECKING:
output: int
class _XRoot(Protocol):
def xrandr_get_screen_resources(self) -> _ScreenResources: ...
def xrandr_get_screen_resources(self) -> _ScreenResources:
"""Mirror ``root.xrandr_get_screen_resources()``."""
def xrandr_get_output_primary(self) -> _PrimaryReply: ...
def xrandr_get_output_primary(self) -> _PrimaryReply:
"""Mirror ``root.xrandr_get_output_primary()``."""
class _XScreen(Protocol):
root: _XRoot
class _XDisplay(Protocol):
def screen(self) -> _XScreen: ...
def screen(self) -> _XScreen:
"""Mirror ``Display.screen()``."""
def xrandr_get_output_info(
self, output: int, timestamp: int
) -> _OutputInfo: ...
def xrandr_get_output_info(self, output: int, timestamp: int) -> _OutputInfo:
"""Mirror ``Display.xrandr_get_output_info()``."""
def xrandr_get_crtc_info(self, crtc: int, timestamp: int) -> _CrtcInfo: ...
def xrandr_get_crtc_info(self, crtc: int, timestamp: int) -> _CrtcInfo:
"""Mirror ``Display.xrandr_get_crtc_info()``."""
def close(self) -> None: ...
def close(self) -> None:
"""Mirror ``Display.close()``."""
_logger = logging.getLogger(__name__)
@ -88,16 +93,15 @@ class RandrBackend:
@classmethod
def create(cls) -> RandrBackend | None:
"""Open an X connection for RandR queries, or None if unavailable."""
try:
from Xlib import display as xdisplay
except ImportError:
modules = load_xlib("Xlib.display")
if modules is None:
_logger.debug(
"python-xlib is not installed; RandR events and RandR "
"enumeration are unavailable, falling back to xrandr"
)
return None
try:
return cls(xdisplay.Display())
return cls(modules[0].Display())
except (OSError, ValueError) as exc:
_logger.warning("could not open an X display for RandR: %s", exc)
return None

32
gatelock/gatelock/_recovery.py
View File

@ -56,11 +56,7 @@ class RecoveryLoop:
def __init__(self, root: tk.Misc, collaborators: RecoveryCollaborators) -> None:
"""Wire the loop to the pieces it re-asserts over."""
self._root = root
self._config = collaborators.config
self._surfaces = collaborators.surfaces
self._enumerator = collaborators.enumerator
self._detector = collaborators.detector
self._hooks = collaborators.hooks
self._parts = collaborators
self._running = False
self._ticks = 0
self._drain_job: str | None = None
@ -92,18 +88,22 @@ class RecoveryLoop:
"""Queue the next cheap drain tick."""
if not self._running:
return
self._drain_job = self._root.after(self._config.detect_drain_ms, self._drain)
self._drain_job = self._root.after(
self._parts.config.detect_drain_ms, self._drain
)
def _schedule_verify(self) -> None:
"""Queue the next full verify tick."""
if not self._running:
return
self._verify_job = self._root.after(self._config.recovery_tick_ms, self._verify)
self._verify_job = self._root.after(
self._parts.config.recovery_tick_ms, self._verify
)
def _drain(self) -> None:
"""Run a full tick only if a push signal arrived. Cheap otherwise."""
try:
if self._detector.take_pending():
if self._parts.detector.take_pending():
_logger.debug("output-change signal received; re-asserting the lock")
self.tick()
# Same fail-open reasoning as _verify.
@ -136,7 +136,7 @@ class RecoveryLoop:
What this pass saw and corrected.
"""
self._ticks += 1
scan = self._enumerator.scan()
scan = self._parts.enumerator.scan()
if not scan.ok:
# No information is not a reason to change anything. Keep every
@ -146,9 +146,9 @@ class RecoveryLoop:
)
return RecoveryReport(scan_ok=False, source=scan.source)
self._surfaces.update_backdrop()
delta = self._surfaces.apply(scan)
corrected = self._surfaces.enforce()
self._parts.surfaces.update_backdrop()
delta = self._parts.surfaces.apply(scan)
corrected = self._parts.surfaces.enforce()
grab_reasserted = self._reassert_grab()
vt_reasserted = self._reassert_vt()
if delta.created:
@ -191,7 +191,7 @@ class RecoveryLoop:
Returns:
True if the grab had to be re-taken.
"""
if self._config.resolved_grab() != "global":
if self._parts.config.resolved_grab() != "global":
return False
if self.holds_grab():
return False
@ -226,8 +226,8 @@ class RecoveryLoop:
that widget the actual Tk focus target.
"""
with contextlib.suppress(tk.TclError):
index = self._surfaces.preferred_focus_index()
self._hooks.on_focus_ready(self._surfaces.focus_surface(index))
index = self._parts.surfaces.preferred_focus_index()
self._parts.hooks.on_focus_ready(self._parts.surfaces.focus_surface(index))
def _reassert_vt(self) -> bool:
"""Periodically re-disable VT switching. Never re-enables it.
@ -235,7 +235,7 @@ class RecoveryLoop:
Returns:
True on ticks where the re-assert ran.
"""
if not self._config.resolved_disable_vt():
if not self._parts.config.resolved_disable_vt():
return False
if self._ticks % _VT_REASSERT_EVERY:
return False

6
gatelock/gatelock/_scrollfit.py
View File

@ -150,9 +150,9 @@ def request_content_size(target: FitTarget) -> None:
"""
wanted_w = target.content.winfo_reqwidth()
wanted_h = target.content.winfo_reqheight()
bar = target.scrollbar.winfo_reqwidth() if target.scrollbar else 0
bar_w = target.scrollbar.winfo_reqwidth() if target.scrollbar else 0
target.canvas.configure(
width=max(1, min(wanted_w, target.canvas.winfo_screenwidth() - bar)),
width=max(1, min(wanted_w, target.canvas.winfo_screenwidth() - bar_w)),
height=max(1, min(wanted_h, target.canvas.winfo_screenheight())),
)
@ -173,7 +173,7 @@ def report_overflow(target: FitTarget) -> None:
# A viewport Tk has not laid out yet reports 1px, which is not an
# overflow -- reporting it would cry wolf on every startup and teach
# the reader to ignore the one message that matters.
if view_h > _UNMAPPED_VIEWPORT_PX and wanted_h > view_h:
if _UNMAPPED_VIEWPORT_PX < view_h < wanted_h:
_logger.warning(
"lock content overflows its surface: %dpx of content in a "
"%dpx viewport (%dpx off-screen); it is reachable only by "

38
gatelock/gatelock/_window.py
View File

@ -103,25 +103,27 @@ class LockWindow:
self._closed = False
self._focus_notified = False
self._preempted_pids: set[int] = set()
self._surfaces = SurfaceSet(root, config, hooks)
self._enumerator = OutputEnumerator(root)
self._detector = OutputChangeDetector(root)
self._recovery = RecoveryLoop(
surfaces = SurfaceSet(root, config, hooks)
enumerator = OutputEnumerator(root)
detector = OutputChangeDetector(root)
recovery = RecoveryLoop(
root,
RecoveryCollaborators(
config=config,
surfaces=self._surfaces,
enumerator=self._enumerator,
detector=self._detector,
surfaces=surfaces,
enumerator=enumerator,
detector=detector,
hooks=hooks,
),
)
# The arming bundle is also where the window reaches its
# collaborators afterwards; it is the one owner of those four.
self._arming = _arming.ArmingCollaborators(
config=config,
surfaces=self._surfaces,
enumerator=self._enumerator,
detector=self._detector,
recovery=self._recovery,
surfaces=surfaces,
enumerator=enumerator,
detector=detector,
recovery=recovery,
notify_focus_ready=self._notify_focus_ready,
log_grab_blocked=self._log_grab_blocked,
)
@ -129,7 +131,7 @@ class LockWindow:
@property
def surfaces(self) -> SurfaceSet:
"""The live surface set, for apps that need to fan work out over it."""
return self._surfaces
return self._arming.surfaces
# -- window mechanics -----------------------------------------------------
@ -179,8 +181,8 @@ class LockWindow:
return
self._focus_notified = True
with contextlib.suppress(tk.TclError):
index = self._surfaces.preferred_focus_index()
self._hooks.on_focus_ready(self._surfaces.focus_surface(index))
index = self._arming.surfaces.preferred_focus_index()
self._hooks.on_focus_ready(self._arming.surfaces.focus_surface(index))
# -- lifecycle --------------------------------------------------------
@ -222,15 +224,15 @@ class LockWindow:
if self._closed:
return
self._closed = True
self._recovery.stop()
self._detector.stop()
self._arming.recovery.stop()
self._arming.detector.stop()
self._hooks.on_close()
self._restore_vt()
if self._arbiter is not None:
self._arbiter.release()
self._enumerator.close()
self._arming.enumerator.close()
with contextlib.suppress(tk.TclError):
self._surfaces.destroy_all()
self._arming.surfaces.destroy_all()
with contextlib.suppress(tk.TclError):
self.root.destroy()

30
gatelock/gatelock/_xlib.py Normal file
View File

@ -0,0 +1,30 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Loading the optional ``python-xlib`` dependency, in one place.
``python-xlib`` is optional: without it, output detection falls back to Tk
``<Configure>`` events plus ``xrandr`` polling. The fallback is the path that
must stay tested, and a module-scope ``try/except ImportError`` would resolve
it once at import time, leaving the ``except`` arm permanently unreachable for
branch coverage. So the modules are loaded on demand, through :mod:`importlib`,
by the two call sites that need them.
"""
from __future__ import annotations
import importlib
from types import ModuleType
def load_xlib(*names: str) -> tuple[ModuleType, ...] | None:
"""Import the named ``Xlib`` modules, or None if python-xlib is unavailable.
Args:
names: Fully-qualified module names, e.g. ``"Xlib.display"``.
Returns:
The modules in the order asked for, or None on the first ImportError.
"""
try:
return tuple(importlib.import_module(name) for name in names)
except ImportError:
return None

5
gatelock/gatelock/tests/test_arbiter.py
View File

@ -18,11 +18,14 @@ from gatelock._arbiter import (
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
)
from gatelock._claims import (
Claim,
_same_file,
_try_lock,
default_runtime_dir,
grab_strength,
)
from gatelock._claims import _same_file, _try_lock
if TYPE_CHECKING:
from collections.abc import Iterator

2
gatelock/gatelock/tests/test_arbiter_lifecycle.py
View File

@ -18,8 +18,8 @@ from gatelock._arbiter import (
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
Claim,
)
from gatelock._claims import Claim
if TYPE_CHECKING:
from collections.abc import Iterator

2
gatelock/gatelock/tests/test_arbiter_release.py
View File

@ -18,8 +18,8 @@ from gatelock._arbiter import (
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
Claim,
)
from gatelock._claims import Claim
if TYPE_CHECKING:
from collections.abc import Iterator

6
gatelock/gatelock/tests/test_recovery.py
View File

@ -122,7 +122,7 @@ class TestScheduling:
"""No change signal means no full pass."""
recovery, enumerator, _surfaces = loop
enumerator.scan.return_value = BOTH
recovery._detector.take_pending.return_value = False
recovery._parts.detector.take_pending.return_value = False
recovery._running = True
recovery._drain()
enumerator.scan.assert_not_called()
@ -133,7 +133,7 @@ class TestScheduling:
"""A pending change triggers a full pass."""
recovery, enumerator, _surfaces = loop
enumerator.scan.return_value = BOTH
recovery._detector.take_pending.return_value = True
recovery._parts.detector.take_pending.return_value = True
recovery._running = True
recovery._drain()
enumerator.scan.assert_called_once_with()
@ -172,7 +172,7 @@ class TestScheduling:
"""Same for the cheap cadence."""
recovery, enumerator, _surfaces = loop
enumerator.scan.side_effect = RuntimeError("boom")
recovery._detector.take_pending.return_value = True
recovery._parts.detector.take_pending.return_value = True
recovery._running = True
with caplog.at_level(logging.ERROR):
recovery._drain()

2
gatelock/gatelock/tests/test_window_surfaces.py
View File

@ -8,8 +8,8 @@ from unittest.mock import MagicMock, patch
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
Claim,
)
from gatelock._claims import Claim
from gatelock._detect import _RandrEventSource
from gatelock._outputs import Output, OutputRect
from gatelock._window import LockConfig

2
gatelock/gatelock/tests/test_window_surfaces_preempt.py
View File

@ -15,8 +15,8 @@ from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Claim,
)
from gatelock._claims import Claim
from gatelock._detect import _RandrEventSource
from gatelock._outputs import Output, OutputRect
from gatelock._window import LockConfig

26
gatelock/pyproject.toml
View File

@ -95,16 +95,40 @@ color_output = true
exclude = [".venv/"]
[tool.pylint.main]
# pre-commit runs from the monorepo root (~/src/utils), so the package dir
# goes on the path for `import gatelock` to resolve the same way it does
# under pytest -- otherwise pylint reports import-error for the package's
# own modules.
init-hook = "import sys; sys.path.insert(0, 'gatelock')"
analyse-fallback-blocks = true
persistent = true
jobs = 0
py-version = "3.14"
ignore = [".venv", "__pycache__"]
# Test suites intentionally use patterns (protected-access, fixture-arg
# shadowing) that don't apply to source code; ruff's per-file-ignores carry
# the equivalent exemptions, so tests stay linted -- by ruff. This is the same
# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has
# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the
# basename of what pylint is *given*, and pre-commit hands it explicit file
# paths, so a directory name in `ignore` never fires under the hook.
ignore-paths = ["^(.*/)?tests/.*$"]
ignore-patterns = [".*\\.pyi$"]
[tool.pylint.messages_control]
enable = "all"
disable = []
# Globally disabled checks, matching wake-alarm / diet-guard: each is a
# stylistic preference that conflicts with deliberate, clearer code.
# Everything else stays at max strictness.
disable = [
# use-implicit-booleaness family (C1803/C1804/C1805): pylint wants
# `not x` / `not s` instead of `x == 0` / `s == ""`. Explicit comparisons
# against 0 and "" state numeric/string intent more clearly than truthiness
# (and are not equivalent when the value may be None), so we keep them.
"use-implicit-booleaness-not-comparison",
"use-implicit-booleaness-not-comparison-to-string",
"use-implicit-booleaness-not-comparison-to-zero",
]
[tool.pylint.design]
min-public-methods = 0