mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 21:27:14 +01:00
gatelock: split _arbiter.py and _escape.py — package now fully under the cap
_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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG
This commit is contained in:
parent
40f94833c1
commit
f5f9db21dc
@ -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.
|
||||
|
||||
237
gatelock/gatelock/_claims.py
Normal file
237
gatelock/gatelock/_claims.py
Normal file
@ -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
|
||||
@ -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."""
|
||||
|
||||
|
||||
88
gatelock/gatelock/_escape_types.py
Normal file
88
gatelock/gatelock/_escape_types.py
Normal file
@ -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
|
||||
@ -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)
|
||||
|
||||
@ -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."""
|
||||
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
@ -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()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user