From f5f9db21dce0a95f520030ec6a54945b22819c66 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 22 Aug 2026 15:21:46 +0200 Subject: [PATCH] =?UTF-8?q?gatelock:=20split=20=5Farbiter.py=20and=20=5Fes?= =?UTF-8?q?cape.py=20=E2=80=94=20package=20now=20fully=20under=20the=20cap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _arbiter.py 397 -> 246 + _claims 241 _escape.py 300 -> 230 + _escape_types 82 _claims holds the claim document and everything that decides whether one is still owned: taking the advisory lock IS the liveness proof, so reading a claim and reaping a dead one are the same operation and belong together. _arbiter keeps the policy — who outranks whom, and whether this app may arm. An AST pass over every module in the package (names loaded vs. names defined or imported) caught two things a green suite would not have: - _try_lock and fcntl moved with the locking code, so the tests' patch targets had to move too; - OutputSource was left behind by the earlier _outputs split and only resolved because of `from __future__ import annotations` — a latent NameError the moment anything evaluated that annotation. gatelock is now at 0 files over 250 lines: 389 passed, 100% branch coverage, the same 58 public names, and palette_check still at 0 failing checks. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG --- gatelock/gatelock/_arbiter.py | 209 +++------------ gatelock/gatelock/_claims.py | 237 ++++++++++++++++++ gatelock/gatelock/_escape.py | 91 ++----- gatelock/gatelock/_escape_types.py | 88 +++++++ gatelock/gatelock/_output_types.py | 3 + gatelock/gatelock/_outputs.py | 3 +- gatelock/gatelock/tests/test_arbiter.py | 5 +- .../gatelock/tests/test_arbiter_lifecycle.py | 4 +- .../tests/test_window_surfaces_lifecycle.py | 2 +- 9 files changed, 381 insertions(+), 261 deletions(-) create mode 100644 gatelock/gatelock/_claims.py create mode 100644 gatelock/gatelock/_escape_types.py diff --git a/gatelock/gatelock/_arbiter.py b/gatelock/gatelock/_arbiter.py index 73cb817..8915668 100644 --- a/gatelock/gatelock/_arbiter.py +++ b/gatelock/gatelock/_arbiter.py @@ -30,33 +30,48 @@ arms anyway. See :meth:`Arbiter.evaluate`. from __future__ import annotations -from dataclasses import dataclass from datetime import datetime, timezone -import errno -import fcntl -import json import logging import os from pathlib import Path -import tempfile -from typing import IO, Final, Literal +from typing import IO, Final import uuid +from gatelock._claims import ( + ArbiterVerdict, + Claim, + _try_lock, + default_runtime_dir, + grab_strength, + live_claims, + read_claim_if_held, +) + _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 -_RUNTIME_DIR_ENV: Final = "GATELOCK_RUNTIME_DIR" _CLAIMS_DIRNAME: Final = "claims" _HOLDER_FILENAME: Final = "holder.lock" # Ordering for the strength check. A stand-down is only safe towards an # incumbent whose grab is at least this strong. -_GRAB_STRENGTH: Final[dict[str, int]] = {"none": 0, "local": 1, "global": 2} - -VerdictReason = Literal["clear", "outranked", "weaker-incumbent-armed-anyway"] def _utc_now_iso() -> str: @@ -64,133 +79,6 @@ def _utc_now_iso() -> str: return datetime.now(tz=timezone.utc).isoformat() -def grab_strength(grab: str) -> int: - """Return the comparable strength of a grab kind. - - Unknown values sort as the weakest possible, so a future grab kind can - never be silently treated as strong enough to stand down for. - """ - return _GRAB_STRENGTH.get(grab, 0) - - -@dataclass(frozen=True) -class Claim: - """One app's declared intent to hold the screen.""" - - app: str - rank: int - pid: int - started: str - grab: str - disable_vt: bool - instance_id: str - """Identifies the *arbiter*, not the process. - - A pid is not unique enough: one process may legitimately host two - arbiters, and the whole test suite does exactly that to exercise real - ``flock`` contention without spawning subprocesses. - """ - - def to_json(self) -> str: - """Serialise this claim for its on-disk file.""" - return json.dumps( - { - "app": self.app, - "rank": self.rank, - "pid": self.pid, - "started": self.started, - "grab": self.grab, - "disable_vt": self.disable_vt, - "instance_id": self.instance_id, - } - ) - - @classmethod - def from_json(cls, text: str) -> Claim | None: - """Parse a claim file's contents, or None if it is unusable. - - A malformed claim is treated as absent rather than fatal: a half-written - file must never be able to stop a locker from arming. - """ - try: - raw = json.loads(text) - except json.JSONDecodeError: - return None - if not isinstance(raw, dict): - return None - try: - return cls( - app=str(raw["app"]), - rank=int(raw["rank"]), - pid=int(raw["pid"]), - started=str(raw["started"]), - grab=str(raw["grab"]), - disable_vt=bool(raw["disable_vt"]), - instance_id=str(raw["instance_id"]), - ) - except (KeyError, TypeError, ValueError): - return None - - def at_least_as_strong_as(self, other: Claim) -> bool: - """Whether this claim locks at least as firmly as ``other``.""" - return ( - grab_strength(self.grab) >= grab_strength(other.grab) - and self.disable_vt >= other.disable_vt - ) - - def sort_key(self) -> tuple[int, str, int]: - """Rank descending, then earliest start, then lowest pid.""" - return (-self.rank, self.started, self.pid) - - -@dataclass(frozen=True) -class ArbiterVerdict: - """Whether this app may arm, and why.""" - - may_arm: bool - blocked_by: Claim | None - reason: VerdictReason - - -def default_runtime_dir() -> Path: - """Return the directory holding claims and the holder lock. - - Prefers ``$XDG_RUNTIME_DIR`` (tmpfs, so it self-clears on reboot) and falls - back to a per-uid temp directory when that is unset or unwritable. - """ - override = os.environ.get(_RUNTIME_DIR_ENV) - if override: - return Path(override) - xdg = os.environ.get("XDG_RUNTIME_DIR") - if xdg: - return Path(xdg) / "gatelock" - return Path(tempfile.gettempdir()) / f"gatelock-{os.getuid()}" - - -def _try_lock(handle: IO[str]) -> bool: - """Try to take an exclusive non-blocking flock. True if acquired.""" - try: - fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - except OSError as exc: - if exc.errno in {errno.EACCES, errno.EAGAIN}: - return False - raise - return True - - -def _same_file(handle: IO[str], path: Path) -> bool: - """Whether ``path`` still names the file behind ``handle``. - - Guards the reap below: between locking a dead app's claim and unlinking it, - a fresh process may have recreated the same path. Comparing inodes stops us - deleting the newcomer's claim. - """ - try: - return os.fstat(handle.fileno()).st_ino == path.stat().st_ino - except OSError: - return False - - class Arbiter: """Publishes this app's claim and reports whether it may arm.""" @@ -262,42 +150,10 @@ class Arbiter: """Return every claim whose owning process is still alive. Dead owners' claims are reaped as a side effect -- proving a claim is - stale and deleting it are the same operation. + stale and deleting it are the same operation. See + :func:`gatelock._claims.live_claims`. """ - if not self._claims_dir.is_dir(): - return () - claims: list[Claim] = [] - for path in sorted(self._claims_dir.glob("*.json")): - claim = self._read_if_live(path) - if claim is not None: - claims.append(claim) - return tuple(sorted(claims, key=Claim.sort_key)) - - def _read_if_live(self, path: Path) -> Claim | None: - """Return the claim at ``path`` if its owner is alive; else reap it.""" - try: - handle = path.open("r+", encoding="utf-8") - except OSError: - return None - with handle: - if _try_lock(handle): - # We got the lock, so nobody holds it, so the owner is gone. - self._reap(handle, path) - return None - try: - return Claim.from_json(handle.read()) - except OSError: - return None - - def _reap(self, handle: IO[str], path: Path) -> None: - """Delete a claim whose owner has died, unless it was recreated.""" - if not _same_file(handle, path): - _logger.debug("claim %s was recreated while reaping; leaving it", path) - return - try: - path.unlink() - except OSError as exc: - _logger.debug("could not reap stale claim %s: %s", path, exc) + return live_claims(self._claims_dir) def evaluate(self) -> ArbiterVerdict: """Decide whether this app may arm. @@ -365,14 +221,7 @@ class Arbiter: return None if self._holder_handle is not None: return self._claim - try: - with path.open("r+", encoding="utf-8") as handle: - if _try_lock(handle): - # Lock was free, so no live app holds the screen. - return None - return Claim.from_json(handle.read()) - except OSError: - return None + return read_claim_if_held(path) def release(self) -> None: """Drop the holder lock and this app's claim. diff --git a/gatelock/gatelock/_claims.py b/gatelock/gatelock/_claims.py new file mode 100644 index 0000000..d568b63 --- /dev/null +++ b/gatelock/gatelock/_claims.py @@ -0,0 +1,237 @@ +"""The claim an app publishes while it holds the screen, and how to read one. + +Split from :mod:`gatelock._arbiter`, which keeps the arbiter itself. A claim +is a small JSON document under the runtime dir, held open with an advisory +lock for as long as its process lives -- so "is this claim live?" is answered +by trying to take that lock, not by trusting the file's contents. + +Re-exported from :mod:`gatelock._arbiter`, so existing imports keep working. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import errno +import fcntl +import json +import logging +import os +from pathlib import Path +import tempfile +from typing import IO, Final, Literal + +_logger = logging.getLogger(__name__) + +_RUNTIME_DIR_ENV: Final = "GATELOCK_RUNTIME_DIR" + +VerdictReason = Literal["clear", "outranked", "weaker-incumbent-armed-anyway"] + + +_GRAB_STRENGTH: Final[dict[str, int]] = {"none": 0, "local": 1, "global": 2} + + +def grab_strength(grab: str) -> int: + """Return the comparable strength of a grab kind. + + Unknown values sort as the weakest possible, so a future grab kind can + never be silently treated as strong enough to stand down for. + """ + return _GRAB_STRENGTH.get(grab, 0) + + +@dataclass(frozen=True) +class Claim: + """One app's declared intent to hold the screen.""" + + app: str + rank: int + pid: int + started: str + grab: str + disable_vt: bool + instance_id: str + """Identifies the *arbiter*, not the process. + + A pid is not unique enough: one process may legitimately host two + arbiters, and the whole test suite does exactly that to exercise real + ``flock`` contention without spawning subprocesses. + """ + + def to_json(self) -> str: + """Serialise this claim for its on-disk file.""" + return json.dumps( + { + "app": self.app, + "rank": self.rank, + "pid": self.pid, + "started": self.started, + "grab": self.grab, + "disable_vt": self.disable_vt, + "instance_id": self.instance_id, + } + ) + + @classmethod + def from_json(cls, text: str) -> Claim | None: + """Parse a claim file's contents, or None if it is unusable. + + A malformed claim is treated as absent rather than fatal: a half-written + file must never be able to stop a locker from arming. + """ + try: + raw = json.loads(text) + except json.JSONDecodeError: + return None + if not isinstance(raw, dict): + return None + try: + return cls( + app=str(raw["app"]), + rank=int(raw["rank"]), + pid=int(raw["pid"]), + started=str(raw["started"]), + grab=str(raw["grab"]), + disable_vt=bool(raw["disable_vt"]), + instance_id=str(raw["instance_id"]), + ) + except (KeyError, TypeError, ValueError): + return None + + def at_least_as_strong_as(self, other: Claim) -> bool: + """Whether this claim locks at least as firmly as ``other``.""" + return ( + grab_strength(self.grab) >= grab_strength(other.grab) + and self.disable_vt >= other.disable_vt + ) + + def sort_key(self) -> tuple[int, str, int]: + """Rank descending, then earliest start, then lowest pid.""" + return (-self.rank, self.started, self.pid) + + +@dataclass(frozen=True) +class ArbiterVerdict: + """Whether this app may arm, and why.""" + + may_arm: bool + blocked_by: Claim | None + reason: VerdictReason + + +def default_runtime_dir() -> Path: + """Return the directory holding claims and the holder lock. + + Prefers ``$XDG_RUNTIME_DIR`` (tmpfs, so it self-clears on reboot) and falls + back to a per-uid temp directory when that is unset or unwritable. + """ + override = os.environ.get(_RUNTIME_DIR_ENV) + if override: + return Path(override) + xdg = os.environ.get("XDG_RUNTIME_DIR") + if xdg: + return Path(xdg) / "gatelock" + return Path(tempfile.gettempdir()) / f"gatelock-{os.getuid()}" + + +def _try_lock(handle: IO[str]) -> bool: + """Try to take an exclusive non-blocking flock. True if acquired.""" + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in {errno.EACCES, errno.EAGAIN}: + return False + raise + return True + + +def _same_file(handle: IO[str], path: Path) -> bool: + """Whether ``path`` still names the file behind ``handle``. + + Guards the reap below: between locking a dead app's claim and unlinking it, + a fresh process may have recreated the same path. Comparing inodes stops us + deleting the newcomer's claim. + """ + try: + return os.fstat(handle.fileno()).st_ino == path.stat().st_ino + except OSError: + return False + + +def _reap(handle: IO[str], path: Path) -> None: + """Delete a claim whose owner has died, unless it was recreated. + + Args: + handle: The open handle whose lock proved the owner is gone. + path: The claim file to remove. + """ + if not _same_file(handle, path): + _logger.debug("claim %s was recreated while reaping; leaving it", path) + return + try: + path.unlink() + except OSError as exc: + _logger.debug("could not reap stale claim %s: %s", path, exc) + + +def _read_if_live(path: Path) -> Claim | None: + """Return the claim at `path` if its owner is alive; else reap it. + + Taking the advisory lock succeeds only when nobody holds it, which means + the owning process is gone -- so a successful lock is the proof of death, + and the file is removed rather than returned. + """ + try: + handle = path.open("r+", encoding="utf-8") + except OSError: + return None + with handle: + if _try_lock(handle): + _reap(handle, path) + return None + try: + return Claim.from_json(handle.read()) + except OSError: + return None + + +def live_claims(claims_dir: Path) -> tuple[Claim, ...]: + """Every claim under `claims_dir` whose owning process is still alive. + + Dead owners' claims are reaped as a side effect -- proving a claim is + stale and deleting it are the same operation. + + Args: + claims_dir: Directory holding the `*.json` claim files. + + Returns: + The live claims, in arbitration order. + """ + if not claims_dir.is_dir(): + return () + claims = [ + claim + for path in sorted(claims_dir.glob("*.json")) + if (claim := _read_if_live(path)) is not None + ] + return tuple(sorted(claims, key=Claim.sort_key)) + + +def read_claim_if_held(path: Path) -> Claim | None: + """The claim at `path`, but only while a live process still holds it. + + Taking the advisory lock succeeds only when nobody holds it, so a + successful lock means the screen is free and there is no holder to name. + + Args: + path: The holder claim file. + + Returns: + The holder's claim, or None if the lock was free or unreadable. + """ + try: + with path.open("r+", encoding="utf-8") as handle: + if _try_lock(handle): + return None + return Claim.from_json(handle.read()) + except OSError: + return None diff --git a/gatelock/gatelock/_escape.py b/gatelock/gatelock/_escape.py index 057ac98..00cd1f1 100644 --- a/gatelock/gatelock/_escape.py +++ b/gatelock/gatelock/_escape.py @@ -25,8 +25,7 @@ workout debt are not general, and are deliberately not modelled here. from __future__ import annotations -from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone +from datetime import timedelta import json import logging from typing import TYPE_CHECKING, Any @@ -36,85 +35,31 @@ 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, + EscapePolicy, + _parse_iso, + _today_iso, +) + _logger = logging.getLogger(__name__) -_DATE_FORMAT = "%Y-%m-%d" +# Re-exported after the 250-line split; gatelock's __init__ and the tests +# both import these from here. +__all__ = [ + "EscapeDraft", + "EscapeHistory", + "EscapePolicy", + "EscapeTracker", +] + # The severity scale the dialog presents. _MIN_SEVERITY = 1 _MAX_SEVERITY = 10 -@dataclass(frozen=True) -class EscapePolicy: - """How hard one app's escape hatch is to use. - - Defaults mirror screen-locker's sick mode, which is the proven shape. - - Attributes: - name: Short identifier used in logs and the summary line. - label: Human wording for one use, e.g. "sick day". - budget_per_7_days: Uses allowed in a rolling week. - budget_per_30_days: Uses allowed in a rolling month. - budget_per_90_days: Uses allowed in a rolling quarter. - lockout_seconds: Base wait before the hatch actually opens. - lockout_multiplier_per_recent: Wait multiplier per use in 30 days. - justification_min_chars: Minimum length of the written reason. - history_review_count: How many past reasons to show back to the user. - """ - - name: str - label: str - budget_per_7_days: int = 1 - budget_per_30_days: int = 3 - budget_per_90_days: int = 10 - lockout_seconds: int = 120 - lockout_multiplier_per_recent: int = 2 - justification_min_chars: int = 120 - history_review_count: int = 10 - - -@dataclass -class EscapeHistory: - """Persistent record of every use of one app's hatch.""" - - used_days: list[str] = field(default_factory=list) - justifications: list[dict[str, Any]] = field(default_factory=list) - - -@dataclass(frozen=True) -class EscapeDraft: - """A proposed use of the hatch, before validation.""" - - reason: str - onset: str - severity: int - description: str - - -def _today_iso() -> str: - """Return today's date as ``YYYY-MM-DD`` (UTC).""" - return datetime.now(tz=timezone.utc).strftime(_DATE_FORMAT) - - -def _parse_iso(date_str: str) -> datetime | None: - """Parse ``YYYY-MM-DD`` into a UTC datetime, or None if unparsable. - - An unparsable entry is *ignored* rather than fatal, but logged loudly: - silently counting it as zero would quietly widen the budget. - """ - try: - return datetime.strptime(date_str, _DATE_FORMAT).replace(tzinfo=timezone.utc) - except ValueError as exc: - _logger.warning( - "escape history holds an unparsable date %r (%s) -- that entry is " - "IGNORED when counting against the budget", - date_str, - exc, - ) - return None - - class EscapeTracker: """Budget, lockout and tamper-evident history for one app's hatch.""" diff --git a/gatelock/gatelock/_escape_types.py b/gatelock/gatelock/_escape_types.py new file mode 100644 index 0000000..51647e9 --- /dev/null +++ b/gatelock/gatelock/_escape_types.py @@ -0,0 +1,88 @@ +"""The policy, history and draft types the escape hatch works with. + +Split from :mod:`gatelock._escape`, which keeps the tracker. Data and two date +helpers, no I/O, so the shapes can be read without the persistence around them. + +Re-exported from :mod:`gatelock._escape`, so existing imports keep working. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +import logging +from typing import Any + +_logger = logging.getLogger(__name__) + +_DATE_FORMAT = "%Y-%m-%d" + + +@dataclass(frozen=True) +class EscapePolicy: + """How hard one app's escape hatch is to use. + + Defaults mirror screen-locker's sick mode, which is the proven shape. + + Attributes: + name: Short identifier used in logs and the summary line. + label: Human wording for one use, e.g. "sick day". + budget_per_7_days: Uses allowed in a rolling week. + budget_per_30_days: Uses allowed in a rolling month. + budget_per_90_days: Uses allowed in a rolling quarter. + lockout_seconds: Base wait before the hatch actually opens. + lockout_multiplier_per_recent: Wait multiplier per use in 30 days. + justification_min_chars: Minimum length of the written reason. + history_review_count: How many past reasons to show back to the user. + """ + + name: str + label: str + budget_per_7_days: int = 1 + budget_per_30_days: int = 3 + budget_per_90_days: int = 10 + lockout_seconds: int = 120 + lockout_multiplier_per_recent: int = 2 + justification_min_chars: int = 120 + history_review_count: int = 10 + + +@dataclass +class EscapeHistory: + """Persistent record of every use of one app's hatch.""" + + used_days: list[str] = field(default_factory=list) + justifications: list[dict[str, Any]] = field(default_factory=list) + + +@dataclass(frozen=True) +class EscapeDraft: + """A proposed use of the hatch, before validation.""" + + reason: str + onset: str + severity: int + description: str + + +def _today_iso() -> str: + """Return today's date as ``YYYY-MM-DD`` (UTC).""" + return datetime.now(tz=timezone.utc).strftime(_DATE_FORMAT) + + +def _parse_iso(date_str: str) -> datetime | None: + """Parse ``YYYY-MM-DD`` into a UTC datetime, or None if unparsable. + + An unparsable entry is *ignored* rather than fatal, but logged loudly: + silently counting it as zero would quietly widen the budget. + """ + try: + return datetime.strptime(date_str, _DATE_FORMAT).replace(tzinfo=timezone.utc) + except ValueError as exc: + _logger.warning( + "escape history holds an unparsable date %r (%s) -- that entry is " + "IGNORED when counting against the budget", + date_str, + exc, + ) + return None diff --git a/gatelock/gatelock/_output_types.py b/gatelock/gatelock/_output_types.py index 40c13d0..0faba13 100644 --- a/gatelock/gatelock/_output_types.py +++ b/gatelock/gatelock/_output_types.py @@ -11,6 +11,9 @@ Re-exported from :mod:`gatelock._outputs`, so existing imports keep working. from __future__ import annotations from dataclasses import dataclass +from typing import Literal + +OutputSource = Literal["randr", "xrandr", "tk", "none"] @dataclass(frozen=True) diff --git a/gatelock/gatelock/_outputs.py b/gatelock/gatelock/_outputs.py index d04b45b..b08e792 100644 --- a/gatelock/gatelock/_outputs.py +++ b/gatelock/gatelock/_outputs.py @@ -35,7 +35,7 @@ if TYPE_CHECKING: from collections.abc import Sequence import tkinter as tk -from gatelock._output_types import Output, OutputRect, OutputScan +from gatelock._output_types import Output, OutputRect, OutputScan, OutputSource from gatelock._randr import RandrBackend from gatelock._xrandr import parse_xrandr_query, scan_xrandr @@ -66,7 +66,6 @@ _OUTPUT_LINE = re.compile( re.MULTILINE, ) -OutputSource = Literal["randr", "xrandr", "tk", "none"] """Which backend produced a scan. Reported so logs can explain a degradation.""" diff --git a/gatelock/gatelock/tests/test_arbiter.py b/gatelock/gatelock/tests/test_arbiter.py index 0ece730..6eac6ce 100644 --- a/gatelock/gatelock/tests/test_arbiter.py +++ b/gatelock/gatelock/tests/test_arbiter.py @@ -13,13 +13,12 @@ from unittest.mock import patch import pytest +from gatelock._claims import _same_file, _try_lock from gatelock._arbiter import ( RANK_SCREEN_LOCKER, RANK_WAKE_ALARM, Arbiter, Claim, - _same_file, - _try_lock, default_runtime_dir, grab_strength, ) @@ -207,7 +206,7 @@ class TestTryLock: exc = OSError() exc.errno = errno.ENOSPC with ( - patch("gatelock._arbiter.fcntl.flock", side_effect=exc), + patch("gatelock._claims.fcntl.flock", side_effect=exc), pytest.raises(OSError, match=r"^$"), ): _try_lock(handle) diff --git a/gatelock/gatelock/tests/test_arbiter_lifecycle.py b/gatelock/gatelock/tests/test_arbiter_lifecycle.py index ee3a8e6..5a41a56 100644 --- a/gatelock/gatelock/tests/test_arbiter_lifecycle.py +++ b/gatelock/gatelock/tests/test_arbiter_lifecycle.py @@ -109,7 +109,7 @@ class TestPublishAndLiveClaims: """A claim recreated between lock and unlink is not deleted.""" dead = tmp_path / "rt" / "claims" / "0300-999-dead.json" dead.write_text(dead_claim().to_json(), "utf-8") - with patch("gatelock._arbiter._same_file", return_value=False): + with patch("gatelock._claims._same_file", return_value=False): arb.live_claims() assert dead.exists() @@ -127,7 +127,7 @@ class TestPublishAndLiveClaims: """A second publish on a locked path continues unpublished.""" arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt") arbiter.publish() - with patch("gatelock._arbiter._try_lock", return_value=False): + with patch("gatelock._claims._try_lock", return_value=False): arbiter.publish() # must not raise arbiter.release() diff --git a/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py b/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py index c8dab83..9a71798 100644 --- a/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py +++ b/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py @@ -69,7 +69,7 @@ class TestRemainingBranches: handle.read.side_effect = OSError("io error") with ( patch("gatelock._arbiter.Path.open", return_value=handle), - patch("gatelock._arbiter._try_lock", return_value=False), + patch("gatelock._claims._try_lock", return_value=False), ): assert arbiter.live_claims() == () arbiter.release()