kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 20:27:15 +01:00
Code Issues Actions 16 Packages Projects Releases Wiki Activity

gatelock: split seven oversized test files by test group

Browse Source 
Each file split at a class or concern boundary, never mechanically:

  test_arbiter.py       432 -> 222 + _lifecycle 183 + _release 201
  test_outputs.py       418 -> 187 + _randr 204 + _enumerator 126
  test_scrollable.py    374 -> 231 + _into_view 197
  test_escape.py        343 -> 175 + _budget 216
  test_window_surfaces  341 -> 110 + _lifecycle 104 + _preempt 182
  test_surfaces.py      282 -> 167 + _focus 154
  test_window.py        269 -> 143 + _grab 139

Test counts verified against a clean HEAD worktree, per file group rather
than in aggregate: arbiter 49/49, outputs+escape 84/84, window+surfaces
72/72, scrollable 18/18. The whole suite is 389 passed at 100% branch
coverage, identical to before.

The one non-mechanical part: _reply/_randr_display are module-level helpers
defined *between* classes, so they did not travel with the header and had to
be moved to the RandR file by hand — six tests failed until they did.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG
This commit is contained in:
Krzysztof Rudnicki 2026-08-22 14:45:08 +02:00
parent 408c7a2585
commit 389832314a
17 changed files with 1692 additions and 1229 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,12 +9,11 @@ from __future__ import annotations
import errno
import os
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
@ -220,213 +219,3 @@ class TestTryLock:
with path.open("r+", encoding="utf-8") as handle:
path.unlink()
assert _same_file(handle, path) is False
class TestPublishAndLiveClaims:
"""Publishing, liveness and reaping."""
def test_own_claim_is_live(self, arb: Arbiter) -> None:
"""A published claim shows up as live."""
assert [c.app for c in arb.live_claims()] == ["screen_locker"]
def test_no_claims_dir(self, tmp_path: Path) -> None:
"""Before anyone publishes, there are no claims."""
assert hard("a", 1, tmp_path / "empty").live_claims() == ()
def test_dead_claim_is_reaped(self, arb: Arbiter, tmp_path: Path) -> None:
"""An unlocked claim file proves its owner died; it is deleted."""
dead = tmp_path / "rt" / "claims" / "0300-999-dead.json"
dead.write_text(dead_claim().to_json(), "utf-8")
assert "ghost" not in [c.app for c in arb.live_claims()]
assert not dead.exists()
def test_unreadable_claim_ignored(self, arb: Arbiter, tmp_path: Path) -> None:
"""A claim that cannot be opened is skipped, not fatal."""
bad = tmp_path / "rt" / "claims" / "0100-1-bad.json"
bad.write_text("{}", encoding="utf-8")
with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")):
assert arb.live_claims() == ()
def test_reap_skips_recreated_file(self, arb: Arbiter, tmp_path: Path) -> None:
"""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):
arb.live_claims()
assert dead.exists()
def test_reap_tolerates_unlink_failure(self, arb: Arbiter, tmp_path: Path) -> None:
"""A failed unlink is logged, not raised."""
dead = tmp_path / "rt" / "claims" / "0300-999-dead.json"
dead.write_text(dead_claim().to_json(), "utf-8")
with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")):
live = arb.live_claims()
# The reap is best-effort: the claim is excluded from the live set even
# when the file itself cannot be removed.
assert "ghost" not in [c.app for c in live]
def test_publish_twice_warns_and_continues(self, tmp_path: Path) -> None:
"""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):
arbiter.publish() # must not raise
arbiter.release()
class TestEvaluate:
"""Who may arm."""
def test_clear_when_alone(self, arb: Arbiter) -> None:
"""Nothing pending means arm."""
verdict = arb.evaluate()
assert verdict.may_arm is True
assert verdict.reason == "clear"
assert verdict.blocked_by is None
def test_lower_rank_does_not_block(self, arb: Arbiter, tmp_path: Path) -> None:
"""A weaker-ranked peer never blocks."""
diet = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
diet.publish()
assert arb.evaluate().may_arm is True
diet.release()
def test_stronger_higher_rank_blocks(self, arb: Arbiter, tmp_path: Path) -> None:
"""A higher-ranked, equally strong app wins."""
alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt")
alarm.publish()
verdict = arb.evaluate()
assert verdict.may_arm is False
assert verdict.reason == "outranked"
assert verdict.blocked_by is not None
assert verdict.blocked_by.app == "wake_alarm"
alarm.release()
def test_weaker_higher_rank_arms_anyway(self, arb: Arbiter, tmp_path: Path) -> None:
"""Rank must never be able to reduce total lock strength."""
soft_alarm = Arbiter(
"wake_alarm",
RANK_WAKE_ALARM,
grab="none",
disable_vt=False,
runtime_dir=tmp_path / "rt",
)
soft_alarm.publish()
verdict = arb.evaluate()
assert verdict.may_arm is True
assert verdict.reason == "weaker-incumbent-armed-anyway"
soft_alarm.release()
class TestHolder:
"""Owning and naming the screen."""
def test_acquire_and_reacquire(self, arb: Arbiter) -> None:
"""Acquiring twice is idempotent."""
assert arb.acquire_holder() is True
assert arb.holds_screen is True
assert arb.acquire_holder() is True
def test_second_app_is_blocked(self, arb: Arbiter, tmp_path: Path) -> None:
"""Only one app holds the screen."""
arb.acquire_holder()
other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
other.publish()
assert other.acquire_holder() is False
other.release()
def test_blocked_app_learns_the_holder(self, arb: Arbiter, tmp_path: Path) -> None:
"""A failed acquire must not erase the incumbent's claim."""
arb.acquire_holder()
other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
other.publish()
other.acquire_holder()
holder = other.describe_holder()
assert holder is not None
assert holder.app == "screen_locker"
other.release()
def test_describe_holder_when_we_hold_it(self, arb: Arbiter) -> None:
"""The holder describes itself without touching the file."""
arb.acquire_holder()
holder = arb.describe_holder()
assert holder is not None
assert holder.app == "screen_locker"
def test_describe_holder_no_file(self, arb: Arbiter) -> None:
"""No holder file means nobody holds the screen."""
assert arb.describe_holder() is None
def test_describe_holder_stale_lock(self, arb: Arbiter, tmp_path: Path) -> None:
"""A holder file nobody locks means the owner died."""
(tmp_path / "rt").mkdir(parents=True, exist_ok=True)
(tmp_path / "rt" / "holder.lock").write_text(
dead_claim().to_json(), encoding="utf-8"
)
assert arb.describe_holder() is None
def test_describe_holder_oserror(self, arb: Arbiter, tmp_path: Path) -> None:
"""An unreadable holder file is reported as no holder."""
(tmp_path / "rt").mkdir(parents=True, exist_ok=True)
(tmp_path / "rt" / "holder.lock").write_text("{}", encoding="utf-8")
with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")):
assert arb.describe_holder() is None
class TestRelease:
"""Clean exit hands the screen over."""
def test_release_lets_the_next_app_in(self, tmp_path: Path) -> None:
"""The alarm -> workout handoff: a clean exit must free the screen."""
alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt")
alarm.publish()
assert alarm.acquire_holder() is True
workout = hard("screen_locker", RANK_SCREEN_LOCKER, tmp_path / "rt")
workout.publish()
assert workout.acquire_holder() is False
assert workout.evaluate().may_arm is False
alarm.release()
assert workout.acquire_holder() is True
assert workout.evaluate().may_arm is True
workout.release()
def test_release_is_idempotent(self, tmp_path: Path) -> None:
"""Releasing twice is safe."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
arbiter.release()
arbiter.release()
assert arbiter.holds_screen is False
def test_release_tolerates_unlink_failure(self, tmp_path: Path) -> None:
"""A claim that cannot be removed does not break teardown."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")):
arbiter.release()
def test_claim_property(self, arb: Arbiter) -> None:
"""The arbiter exposes its own claim."""
assert arb.claim.app == "screen_locker"
assert arb.claim.rank == RANK_SCREEN_LOCKER
class TestReleaseHandleClose:
"""Handle-close failures are non-fatal."""
def test_close_oserror_is_swallowed(self, tmp_path: Path) -> None:
"""A failing close does not stop the release."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
arbiter.acquire_holder()
bad = MagicMock()
bad.close.side_effect = OSError("nope")
real_handle = arbiter._holder_handle
assert real_handle is not None
real_handle.close()
arbiter._holder_handle = bad
arbiter.release()
assert arbiter.holds_screen is False

176
gatelock/gatelock/tests/test_arbiter_lifecycle.py Normal file
View File

@ -0,0 +1,176 @@
"""Tests for an arbiter claim's life after it is taken.
Split from ``test_arbiter.py`` (250-line cap): that file covers grab
strength, claim serialisation, the runtime dir and taking the lock; this one
covers publishing, evaluating rivals, describing the holder, and release.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
Claim,
)
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
DEFAULT_STARTED = "2026-07-25T12:00:00+00:00"
def dead_claim(app: str = "ghost", instance_id: str = "dead") -> Claim:
"""A claim belonging to a process that has since died."""
return Claim(
app=app,
rank=RANK_WAKE_ALARM,
pid=999999,
started=DEFAULT_STARTED,
grab="global",
disable_vt=True,
instance_id=instance_id,
)
def make_claim(
*,
rank: int = RANK_SCREEN_LOCKER,
started: str = DEFAULT_STARTED,
grab: str = "global",
disable_vt: bool = True,
) -> Claim:
"""A claim with sensible defaults."""
return Claim(
app="test_app",
rank=rank,
pid=1234,
started=started,
grab=grab,
disable_vt=disable_vt,
instance_id="tok",
)
@pytest.fixture
def arb(tmp_path: Path) -> Iterator[Arbiter]:
"""A published hard-locking arbiter in an isolated runtime dir."""
arbiter = Arbiter(
"screen_locker",
RANK_SCREEN_LOCKER,
grab="global",
disable_vt=True,
runtime_dir=tmp_path / "rt",
)
arbiter.publish()
yield arbiter
arbiter.release()
def hard(name: str, rank: int, root: Path) -> Arbiter:
"""A hard-locking arbiter."""
return Arbiter(name, rank, grab="global", disable_vt=True, runtime_dir=root)
class TestPublishAndLiveClaims:
"""Publishing, liveness and reaping."""
def test_own_claim_is_live(self, arb: Arbiter) -> None:
"""A published claim shows up as live."""
assert [c.app for c in arb.live_claims()] == ["screen_locker"]
def test_no_claims_dir(self, tmp_path: Path) -> None:
"""Before anyone publishes, there are no claims."""
assert hard("a", 1, tmp_path / "empty").live_claims() == ()
def test_dead_claim_is_reaped(self, arb: Arbiter, tmp_path: Path) -> None:
"""An unlocked claim file proves its owner died; it is deleted."""
dead = tmp_path / "rt" / "claims" / "0300-999-dead.json"
dead.write_text(dead_claim().to_json(), "utf-8")
assert "ghost" not in [c.app for c in arb.live_claims()]
assert not dead.exists()
def test_unreadable_claim_ignored(self, arb: Arbiter, tmp_path: Path) -> None:
"""A claim that cannot be opened is skipped, not fatal."""
bad = tmp_path / "rt" / "claims" / "0100-1-bad.json"
bad.write_text("{}", encoding="utf-8")
with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")):
assert arb.live_claims() == ()
def test_reap_skips_recreated_file(self, arb: Arbiter, tmp_path: Path) -> None:
"""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):
arb.live_claims()
assert dead.exists()
def test_reap_tolerates_unlink_failure(self, arb: Arbiter, tmp_path: Path) -> None:
"""A failed unlink is logged, not raised."""
dead = tmp_path / "rt" / "claims" / "0300-999-dead.json"
dead.write_text(dead_claim().to_json(), "utf-8")
with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")):
live = arb.live_claims()
# The reap is best-effort: the claim is excluded from the live set even
# when the file itself cannot be removed.
assert "ghost" not in [c.app for c in live]
def test_publish_twice_warns_and_continues(self, tmp_path: Path) -> None:
"""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):
arbiter.publish() # must not raise
arbiter.release()
class TestEvaluate:
"""Who may arm."""
def test_clear_when_alone(self, arb: Arbiter) -> None:
"""Nothing pending means arm."""
verdict = arb.evaluate()
assert verdict.may_arm is True
assert verdict.reason == "clear"
assert verdict.blocked_by is None
def test_lower_rank_does_not_block(self, arb: Arbiter, tmp_path: Path) -> None:
"""A weaker-ranked peer never blocks."""
diet = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
diet.publish()
assert arb.evaluate().may_arm is True
diet.release()
def test_stronger_higher_rank_blocks(self, arb: Arbiter, tmp_path: Path) -> None:
"""A higher-ranked, equally strong app wins."""
alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt")
alarm.publish()
verdict = arb.evaluate()
assert verdict.may_arm is False
assert verdict.reason == "outranked"
assert verdict.blocked_by is not None
assert verdict.blocked_by.app == "wake_alarm"
alarm.release()
def test_weaker_higher_rank_arms_anyway(self, arb: Arbiter, tmp_path: Path) -> None:
"""Rank must never be able to reduce total lock strength."""
soft_alarm = Arbiter(
"wake_alarm",
RANK_WAKE_ALARM,
grab="none",
disable_vt=False,
runtime_dir=tmp_path / "rt",
)
soft_alarm.publish()
verdict = arb.evaluate()
assert verdict.may_arm is True
assert verdict.reason == "weaker-incumbent-armed-anyway"
soft_alarm.release()

194
gatelock/gatelock/tests/test_arbiter_release.py Normal file
View File

@ -0,0 +1,194 @@
"""Tests for identifying the current holder and releasing a claim.
Split from ``test_arbiter.py`` (250-line cap). ``test_arbiter_lifecycle.py``
keeps publishing and rival evaluation; this one covers who holds the claim
now and what happens when it is given up.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
import pytest
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Arbiter,
Claim,
)
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
DEFAULT_STARTED = "2026-07-25T12:00:00+00:00"
def dead_claim(app: str = "ghost", instance_id: str = "dead") -> Claim:
"""A claim belonging to a process that has since died."""
return Claim(
app=app,
rank=RANK_WAKE_ALARM,
pid=999999,
started=DEFAULT_STARTED,
grab="global",
disable_vt=True,
instance_id=instance_id,
)
def make_claim(
*,
rank: int = RANK_SCREEN_LOCKER,
started: str = DEFAULT_STARTED,
grab: str = "global",
disable_vt: bool = True,
) -> Claim:
"""A claim with sensible defaults."""
return Claim(
app="test_app",
rank=rank,
pid=1234,
started=started,
grab=grab,
disable_vt=disable_vt,
instance_id="tok",
)
@pytest.fixture
def arb(tmp_path: Path) -> Iterator[Arbiter]:
"""A published hard-locking arbiter in an isolated runtime dir."""
arbiter = Arbiter(
"screen_locker",
RANK_SCREEN_LOCKER,
grab="global",
disable_vt=True,
runtime_dir=tmp_path / "rt",
)
arbiter.publish()
yield arbiter
arbiter.release()
def hard(name: str, rank: int, root: Path) -> Arbiter:
"""A hard-locking arbiter."""
return Arbiter(name, rank, grab="global", disable_vt=True, runtime_dir=root)
class TestHolder:
"""Owning and naming the screen."""
def test_acquire_and_reacquire(self, arb: Arbiter) -> None:
"""Acquiring twice is idempotent."""
assert arb.acquire_holder() is True
assert arb.holds_screen is True
assert arb.acquire_holder() is True
def test_second_app_is_blocked(self, arb: Arbiter, tmp_path: Path) -> None:
"""Only one app holds the screen."""
arb.acquire_holder()
other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
other.publish()
assert other.acquire_holder() is False
other.release()
def test_blocked_app_learns_the_holder(self, arb: Arbiter, tmp_path: Path) -> None:
"""A failed acquire must not erase the incumbent's claim."""
arb.acquire_holder()
other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt")
other.publish()
other.acquire_holder()
holder = other.describe_holder()
assert holder is not None
assert holder.app == "screen_locker"
other.release()
def test_describe_holder_when_we_hold_it(self, arb: Arbiter) -> None:
"""The holder describes itself without touching the file."""
arb.acquire_holder()
holder = arb.describe_holder()
assert holder is not None
assert holder.app == "screen_locker"
def test_describe_holder_no_file(self, arb: Arbiter) -> None:
"""No holder file means nobody holds the screen."""
assert arb.describe_holder() is None
def test_describe_holder_stale_lock(self, arb: Arbiter, tmp_path: Path) -> None:
"""A holder file nobody locks means the owner died."""
(tmp_path / "rt").mkdir(parents=True, exist_ok=True)
(tmp_path / "rt" / "holder.lock").write_text(
dead_claim().to_json(), encoding="utf-8"
)
assert arb.describe_holder() is None
def test_describe_holder_oserror(self, arb: Arbiter, tmp_path: Path) -> None:
"""An unreadable holder file is reported as no holder."""
(tmp_path / "rt").mkdir(parents=True, exist_ok=True)
(tmp_path / "rt" / "holder.lock").write_text("{}", encoding="utf-8")
with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")):
assert arb.describe_holder() is None
class TestRelease:
"""Clean exit hands the screen over."""
def test_release_lets_the_next_app_in(self, tmp_path: Path) -> None:
"""The alarm -> workout handoff: a clean exit must free the screen."""
alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt")
alarm.publish()
assert alarm.acquire_holder() is True
workout = hard("screen_locker", RANK_SCREEN_LOCKER, tmp_path / "rt")
workout.publish()
assert workout.acquire_holder() is False
assert workout.evaluate().may_arm is False
alarm.release()
assert workout.acquire_holder() is True
assert workout.evaluate().may_arm is True
workout.release()
def test_release_is_idempotent(self, tmp_path: Path) -> None:
"""Releasing twice is safe."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
arbiter.release()
arbiter.release()
assert arbiter.holds_screen is False
def test_release_tolerates_unlink_failure(self, tmp_path: Path) -> None:
"""A claim that cannot be removed does not break teardown."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")):
arbiter.release()
def test_claim_property(self, arb: Arbiter) -> None:
"""The arbiter exposes its own claim."""
assert arb.claim.app == "screen_locker"
assert arb.claim.rank == RANK_SCREEN_LOCKER
class TestReleaseHandleClose:
"""Handle-close failures are non-fatal."""
def test_close_oserror_is_swallowed(self, tmp_path: Path) -> None:
"""A failing close does not stop the release."""
arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt")
arbiter.publish()
arbiter.acquire_holder()
bad = MagicMock()
bad.close.side_effect = OSError("nope")
real_handle = arbiter._holder_handle
assert real_handle is not None
real_handle.close()
arbiter._holder_handle = bad
arbiter.release()
assert arbiter.holds_screen is False

168
gatelock/gatelock/tests/test_escape.py
View File

@ -173,171 +173,3 @@ class TestIntegrityNeverWidensTheBudget:
tracker.load()
assert tracker.tampered == 1
assert tracker.is_budget_exhausted(today=TODAY) is True
class TestBudget:
"""Rolling windows."""
def test_counts_only_inside_the_window(self, tracker: EscapeTracker) -> None:
"""An old use falls out of the 7-day window."""
tracker.history.used_days = ["2026-07-24", "2026-06-01"]
assert tracker.count_in_window(7, today=TODAY) == 1
assert tracker.count_in_window(90, today=TODAY) == 2
def test_unparsable_today_counts_nothing(self, tracker: EscapeTracker) -> None:
"""A broken 'today' cannot silently exhaust the budget."""
tracker.history.used_days = [TODAY]
assert tracker.count_in_window(7, today="garbage") == 0
def test_unparsable_entry_is_skipped(self, tracker: EscapeTracker) -> None:
"""A corrupt stored date is ignored, loudly."""
tracker.history.used_days = ["nonsense", TODAY]
assert tracker.count_in_window(7, today=TODAY) == 1
@pytest.mark.parametrize(
("days", "expected"),
[(["2026-07-25"], True), ([], False)],
)
def test_week_window(
self, tracker: EscapeTracker, days: list[str], *, expected: bool
) -> None:
"""One use exhausts the weekly budget of 1."""
tracker.history.used_days = days
assert tracker.is_budget_exhausted(today=TODAY) is expected
def test_month_window(self, tracker: EscapeTracker) -> None:
"""Three uses in 30 days exhausts, even if none is in the last week."""
tracker.history.used_days = ["2026-07-05", "2026-07-06", "2026-07-07"]
assert tracker.is_budget_exhausted(today=TODAY) is True
def test_quarter_window(self, tracker: EscapeTracker) -> None:
"""Ten uses in 90 days exhausts, spread out enough to clear the others."""
tracker.history.used_days = [f"2026-05-{day:02d}" for day in range(1, 11)]
assert tracker.is_budget_exhausted(today=TODAY) is True
def test_nothing_used_is_not_exhausted(self, tracker: EscapeTracker) -> None:
"""A clean history leaves the hatch available."""
assert tracker.is_budget_exhausted(today=TODAY) is False
class TestLockout:
"""The wait grows the more you lean on the hatch."""
@pytest.mark.parametrize(
("recent", "seconds"), [(0, 120), (1, 240), (2, 480), (3, 960)]
)
def test_doubles_per_recent_use(
self, tracker: EscapeTracker, recent: int, seconds: int
) -> None:
"""base * 2 ** uses_in_30_days."""
tracker.history.used_days = [f"2026-07-{10 + i:02d}" for i in range(recent)]
assert tracker.compute_lockout_seconds(today=TODAY) == seconds
class TestValidation:
"""What counts as a real justification."""
def test_accepts_a_proper_one(self, tracker: EscapeTracker) -> None:
"""A complete draft passes."""
assert tracker.validate(good_draft()) is None
@pytest.mark.parametrize(
("draft", "fragment"),
[
(EscapeDraft("", "03:00", 5, "x" * 130), "what the problem is"),
(EscapeDraft("flu", "", 5, "x" * 130), "when it started"),
(EscapeDraft("flu", "03:00", 0, "x" * 130), "between 1 and 10"),
(EscapeDraft("flu", "03:00", 11, "x" * 130), "between 1 and 10"),
(EscapeDraft("flu", "03:00", 5, "short"), "Explain properly"),
],
)
def test_rejections(
self, tracker: EscapeTracker, draft: EscapeDraft, fragment: str
) -> None:
"""Each missing or lazy field is refused with a specific complaint."""
message = tracker.validate(draft)
assert message is not None
assert fragment in message
class TestRecord:
"""Writing a use down."""
def test_records_and_signs(self, tracker: EscapeTracker) -> None:
"""A use is appended, dated and signed."""
assert tracker.record(good_draft(), today=TODAY) is True
entry = tracker.history.justifications[-1]
assert entry["date"] == TODAY
assert entry["policy"] == "sick"
assert "hmac" in entry
def test_unsigned_when_key_unavailable(
self, tracker: EscapeTracker, tmp_path: Path
) -> None:
"""Without a key the entry is still recorded, just unsigned."""
tracker._key_file = tmp_path / "absent.key"
assert tracker.record(good_draft(), today=TODAY) is True
assert "hmac" not in tracker.history.justifications[-1]
def test_same_day_twice_counts_once(self, tracker: EscapeTracker) -> None:
"""Two uses on one day are one day of budget."""
tracker.record(good_draft(), today=TODAY)
tracker.record(good_draft(), today=TODAY)
assert tracker.history.used_days == [TODAY]
assert len(tracker.history.justifications) == 2
def test_uses_today_when_unspecified(self, tracker: EscapeTracker) -> None:
"""Omitting the date records the real today."""
tracker.record(good_draft())
assert tracker.history.used_days == [_today_iso()]
def test_save_failure_is_reported(self, tracker: EscapeTracker) -> None:
"""A read-only disk is surfaced, not swallowed."""
with patch("pathlib.Path.write_text", side_effect=OSError("ro")):
assert tracker.record(good_draft(), today=TODAY) is False
class TestReadBack:
"""Showing past excuses back to the user is half the deterrent."""
def test_empty(self, tracker: EscapeTracker) -> None:
"""A clean history says so."""
assert tracker.format_recent() == "(no previous uses)"
assert tracker.recent_justifications() == []
def test_lists_recent(self, tracker: EscapeTracker) -> None:
"""Past entries are rendered with date, severity and reason."""
tracker.record(good_draft(), today=TODAY)
rendered = tracker.format_recent()
assert TODAY in rendered
assert "8/10" in rendered
assert "migraine" in rendered
def test_respects_explicit_limit(self, tracker: EscapeTracker) -> None:
"""An explicit count overrides the policy default."""
for day in range(1, 6):
tracker.record(good_draft(), today=f"2026-07-{day:02d}")
assert len(tracker.recent_justifications(2)) == 2
def test_handles_entries_missing_fields(
self, tracker: EscapeTracker, tmp_path: Path
) -> None:
"""A partial entry renders without raising."""
(tmp_path / "escape.json").write_text(
json.dumps({"used_days": [], "justifications": [{}]}), encoding="utf-8"
)
tracker.load()
assert "?" in tracker.format_recent()
class TestSummary:
"""The one-line budget readout."""
def test_summary_shows_all_three_windows(self, tracker: EscapeTracker) -> None:
"""Week, month and quarter counts all appear."""
tracker.history.used_days = [TODAY]
assert tracker.budget_summary(today=TODAY) == "Sick: 1/1w · 1/3m · 1/10q"
def test_policy_property(self, tracker: EscapeTracker) -> None:
"""The tracker exposes the policy it enforces."""
assert tracker.policy is POLICY

216
gatelock/gatelock/tests/test_escape_budget.py Normal file
View File

@ -0,0 +1,216 @@
"""Tests for the escape-hatch budget, lockout and recording.
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
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from gatelock._escape import (
EscapeDraft,
EscapePolicy,
EscapeTracker,
_today_iso,
)
if TYPE_CHECKING:
from pathlib import Path
TODAY = "2026-07-25"
POLICY = EscapePolicy(name="sick", label="Sick")
@pytest.fixture
def key_file(tmp_path: Path) -> Path:
"""A throwaway signing key, so tests never touch the real root-owned one."""
path = tmp_path / "hmac.key"
path.write_bytes(b"0" * 32)
return path
@pytest.fixture
def tracker(tmp_path: Path, key_file: Path) -> EscapeTracker:
"""A tracker over an empty history."""
return EscapeTracker(POLICY, tmp_path / "escape.json", key_file=key_file)
def good_draft() -> EscapeDraft:
"""A justification that passes validation."""
return EscapeDraft(
reason="migraine", onset="03:00", severity=8, description="x" * 130
)
class TestBudget:
"""Rolling windows."""
def test_counts_only_inside_the_window(self, tracker: EscapeTracker) -> None:
"""An old use falls out of the 7-day window."""
tracker.history.used_days = ["2026-07-24", "2026-06-01"]
assert tracker.count_in_window(7, today=TODAY) == 1
assert tracker.count_in_window(90, today=TODAY) == 2
def test_unparsable_today_counts_nothing(self, tracker: EscapeTracker) -> None:
"""A broken 'today' cannot silently exhaust the budget."""
tracker.history.used_days = [TODAY]
assert tracker.count_in_window(7, today="garbage") == 0
def test_unparsable_entry_is_skipped(self, tracker: EscapeTracker) -> None:
"""A corrupt stored date is ignored, loudly."""
tracker.history.used_days = ["nonsense", TODAY]
assert tracker.count_in_window(7, today=TODAY) == 1
@pytest.mark.parametrize(
("days", "expected"),
[(["2026-07-25"], True), ([], False)],
)
def test_week_window(
self, tracker: EscapeTracker, days: list[str], *, expected: bool
) -> None:
"""One use exhausts the weekly budget of 1."""
tracker.history.used_days = days
assert tracker.is_budget_exhausted(today=TODAY) is expected
def test_month_window(self, tracker: EscapeTracker) -> None:
"""Three uses in 30 days exhausts, even if none is in the last week."""
tracker.history.used_days = ["2026-07-05", "2026-07-06", "2026-07-07"]
assert tracker.is_budget_exhausted(today=TODAY) is True
def test_quarter_window(self, tracker: EscapeTracker) -> None:
"""Ten uses in 90 days exhausts, spread out enough to clear the others."""
tracker.history.used_days = [f"2026-05-{day:02d}" for day in range(1, 11)]
assert tracker.is_budget_exhausted(today=TODAY) is True
def test_nothing_used_is_not_exhausted(self, tracker: EscapeTracker) -> None:
"""A clean history leaves the hatch available."""
assert tracker.is_budget_exhausted(today=TODAY) is False
class TestLockout:
"""The wait grows the more you lean on the hatch."""
@pytest.mark.parametrize(
("recent", "seconds"), [(0, 120), (1, 240), (2, 480), (3, 960)]
)
def test_doubles_per_recent_use(
self, tracker: EscapeTracker, recent: int, seconds: int
) -> None:
"""base * 2 ** uses_in_30_days."""
tracker.history.used_days = [f"2026-07-{10 + i:02d}" for i in range(recent)]
assert tracker.compute_lockout_seconds(today=TODAY) == seconds
class TestValidation:
"""What counts as a real justification."""
def test_accepts_a_proper_one(self, tracker: EscapeTracker) -> None:
"""A complete draft passes."""
assert tracker.validate(good_draft()) is None
@pytest.mark.parametrize(
("draft", "fragment"),
[
(EscapeDraft("", "03:00", 5, "x" * 130), "what the problem is"),
(EscapeDraft("flu", "", 5, "x" * 130), "when it started"),
(EscapeDraft("flu", "03:00", 0, "x" * 130), "between 1 and 10"),
(EscapeDraft("flu", "03:00", 11, "x" * 130), "between 1 and 10"),
(EscapeDraft("flu", "03:00", 5, "short"), "Explain properly"),
],
)
def test_rejections(
self, tracker: EscapeTracker, draft: EscapeDraft, fragment: str
) -> None:
"""Each missing or lazy field is refused with a specific complaint."""
message = tracker.validate(draft)
assert message is not None
assert fragment in message
class TestRecord:
"""Writing a use down."""
def test_records_and_signs(self, tracker: EscapeTracker) -> None:
"""A use is appended, dated and signed."""
assert tracker.record(good_draft(), today=TODAY) is True
entry = tracker.history.justifications[-1]
assert entry["date"] == TODAY
assert entry["policy"] == "sick"
assert "hmac" in entry
def test_unsigned_when_key_unavailable(
self, tracker: EscapeTracker, tmp_path: Path
) -> None:
"""Without a key the entry is still recorded, just unsigned."""
tracker._key_file = tmp_path / "absent.key"
assert tracker.record(good_draft(), today=TODAY) is True
assert "hmac" not in tracker.history.justifications[-1]
def test_same_day_twice_counts_once(self, tracker: EscapeTracker) -> None:
"""Two uses on one day are one day of budget."""
tracker.record(good_draft(), today=TODAY)
tracker.record(good_draft(), today=TODAY)
assert tracker.history.used_days == [TODAY]
assert len(tracker.history.justifications) == 2
def test_uses_today_when_unspecified(self, tracker: EscapeTracker) -> None:
"""Omitting the date records the real today."""
tracker.record(good_draft())
assert tracker.history.used_days == [_today_iso()]
def test_save_failure_is_reported(self, tracker: EscapeTracker) -> None:
"""A read-only disk is surfaced, not swallowed."""
with patch("pathlib.Path.write_text", side_effect=OSError("ro")):
assert tracker.record(good_draft(), today=TODAY) is False
class TestReadBack:
"""Showing past excuses back to the user is half the deterrent."""
def test_empty(self, tracker: EscapeTracker) -> None:
"""A clean history says so."""
assert tracker.format_recent() == "(no previous uses)"
assert tracker.recent_justifications() == []
def test_lists_recent(self, tracker: EscapeTracker) -> None:
"""Past entries are rendered with date, severity and reason."""
tracker.record(good_draft(), today=TODAY)
rendered = tracker.format_recent()
assert TODAY in rendered
assert "8/10" in rendered
assert "migraine" in rendered
def test_respects_explicit_limit(self, tracker: EscapeTracker) -> None:
"""An explicit count overrides the policy default."""
for day in range(1, 6):
tracker.record(good_draft(), today=f"2026-07-{day:02d}")
assert len(tracker.recent_justifications(2)) == 2
def test_handles_entries_missing_fields(
self, tracker: EscapeTracker, tmp_path: Path
) -> None:
"""A partial entry renders without raising."""
(tmp_path / "escape.json").write_text(
json.dumps({"used_days": [], "justifications": [{}]}), encoding="utf-8"
)
tracker.load()
assert "?" in tracker.format_recent()
class TestSummary:
"""The one-line budget readout."""
def test_summary_shows_all_three_windows(self, tracker: EscapeTracker) -> None:
"""Week, month and quarter counts all appear."""
tracker.history.used_days = [TODAY]
assert tracker.budget_summary(today=TODAY) == "Sick: 1/1w · 1/3m · 1/10q"
def test_policy_property(self, tracker: EscapeTracker) -> None:
"""The tracker exposes the policy it enforces."""
assert tracker.policy is POLICY

231
gatelock/gatelock/tests/test_outputs.py
View File

@ -14,12 +14,9 @@ import pytest
from gatelock._outputs import (
Output,
OutputEnumerator,
OutputRect,
OutputScan,
RandrBackend,
_decode_name,
enumerate_outputs,
parse_xrandr_query,
)
from gatelock._outputs import scan_xrandr as real_scan_xrandr
@ -188,231 +185,3 @@ class TestDecodeName:
def test_decode(self, raw: object, expected: str) -> None:
"""Bytes are decoded, str passes through, bad bytes are replaced."""
assert _decode_name(raw) == expected
def _reply(**fields: object) -> MagicMock:
"""An Xlib reply exposing its fields as attributes, as python-xlib does."""
reply = MagicMock()
for key, value in fields.items():
setattr(reply, key, value)
return reply
def _randr_display(*, outputs: dict[int, dict[str, object]]) -> MagicMock:
"""Build a fake Xlib display exposing ``outputs``."""
display = MagicMock()
root = MagicMock()
root.xrandr_get_screen_resources.return_value = _reply(
config_timestamp=1, outputs=list(outputs)
)
root.xrandr_get_output_primary.return_value = _reply(output=1)
display.screen.return_value.root = root
def output_info(output_id: int, _ts: int) -> MagicMock:
return _reply(**outputs[output_id]["info"])
def crtc_info(crtc: int, _ts: int) -> MagicMock:
return _reply(**outputs[crtc]["crtc"])
display.xrandr_get_output_info.side_effect = output_info
display.xrandr_get_crtc_info.side_effect = crtc_info
return display
class TestRandrBackend:
"""Tests for the RandR backend."""
def test_scan_maps_crtc_to_rect(self) -> None:
"""A connected output with a CRTC and mode becomes a live rect."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 55, "x": 0, "y": 0, "width": 3840, "height": 2160},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.name == "DP-0"
assert output.live is True
assert output.primary is True
assert output.rect == OutputRect(0, 0, 3840, 2160)
def test_no_crtc_is_not_live(self) -> None:
"""crtc == 0 is the dark-monitor case."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}}
)
(output,) = RandrBackend(display).scan() or ()
assert output.connected is True
assert output.rect is None
def test_crtc_without_mode_is_not_live(self) -> None:
"""A CRTC exists but carries no mode: still dark."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 0, "x": 0, "y": 0, "width": 0, "height": 0},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.rect is None
def test_zero_size_crtc_is_not_live(self) -> None:
"""A mode with no pixels is not a screen."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 55, "x": 0, "y": 0, "width": 0, "height": 1080},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.rect is None
def test_disconnected(self) -> None:
"""connection != 0 means disconnected."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-1", "connection": 1, "crtc": 0}}}
)
(output,) = RandrBackend(display).scan() or ()
assert output.connected is False
def test_scan_failure_returns_none(self) -> None:
"""A protocol error degrades to None instead of raising."""
display = MagicMock()
display.screen.side_effect = OSError("connection reset")
assert RandrBackend(display).scan() is None
def test_primary_query_failure_is_tolerated(self) -> None:
"""RandR refusing to name a primary is not fatal."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}}
)
root = display.screen.return_value.root
root.xrandr_get_output_primary.side_effect = OSError("nope")
(output,) = RandrBackend(display).scan() or ()
assert output.primary is False
def test_close_swallows_errors(self) -> None:
"""Teardown never raises."""
display = MagicMock()
display.close.side_effect = OSError("already gone")
RandrBackend(display).close()
def test_close_normal(self) -> None:
"""A clean close closes the display."""
display = MagicMock()
RandrBackend(display).close()
display.close.assert_called_once_with()
class TestCreateWithoutXlib:
"""RandrBackend.create's degradation path, exercised without the patch."""
def test_import_error_returns_none(self) -> None:
"""A missing python-xlib degrades to None, never an exception.
This is the branch the function-scope import exists to keep reachable.
"""
with patch.dict("sys.modules", {"Xlib": None}):
assert REAL_CREATE(RandrBackend) is None
def test_display_error_returns_none(self) -> None:
"""A display that refuses to open returns None."""
fake_display = MagicMock()
fake_display.Display.side_effect = OSError("cannot connect")
# `from Xlib import display` reads the attribute off the parent
# module, so wiring sys.modules alone is not enough.
fake_xlib = MagicMock(display=fake_display)
with patch.dict(
"sys.modules",
{"Xlib": fake_xlib, "Xlib.display": fake_display},
):
assert REAL_CREATE(RandrBackend) is None
def test_success_returns_backend(self) -> None:
"""A working python-xlib yields a usable backend."""
fake_display = MagicMock()
fake_xlib = MagicMock(display=fake_display)
with patch.dict(
"sys.modules",
{"Xlib": fake_xlib, "Xlib.display": fake_display},
):
backend = REAL_CREATE(RandrBackend)
assert isinstance(backend, RandrBackend)
class TestOutputEnumerator:
"""Tests for the backend ladder."""
def test_prefers_randr(self, mock_root: MagicMock) -> None:
"""When RandR answers, xrandr is never consulted."""
backend = MagicMock()
backend.scan.return_value = (
Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)),
)
scan = OutputEnumerator(mock_root, backend=backend).scan()
assert scan.source == "randr"
assert scan.ok is True
def test_falls_back_to_xrandr(self, mock_root: MagicMock) -> None:
"""A failed RandR scan drops to xrandr."""
backend = MagicMock()
backend.scan.return_value = None
with patch(
"gatelock._outputs.scan_xrandr",
return_value=(Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)),),
):
scan = OutputEnumerator(mock_root, backend=backend).scan()
assert scan.source == "xrandr"
def test_falls_back_to_tk(self, mock_root: MagicMock) -> None:
"""With neither backend, Tk's screen size is the floor."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = OutputEnumerator(mock_root, backend=None).scan()
assert scan.source == "tk"
assert scan.live[0].rect == OutputRect(0, 0, 1920, 1080)
def test_total_failure_is_not_ok(self) -> None:
"""With no root and no backend, the scan reports failure."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = OutputEnumerator(None, backend=None).scan()
assert scan.ok is False
assert scan.source == "none"
assert scan.live == ()
def test_backend_property_and_close(self) -> None:
"""close() releases the backend once and is safe to repeat."""
backend = MagicMock()
enumerator = OutputEnumerator(None, backend=backend)
assert enumerator.backend is backend
enumerator.close()
enumerator.close()
backend.close.assert_called_once_with()
def test_enumerate_outputs_helper(self, mock_root: MagicMock) -> None:
"""The one-shot helper works and cleans up after itself."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = enumerate_outputs(mock_root)
assert scan.source == "tk"
class TestOutputScan:
"""Tests for OutputScan's live filter."""
def test_live_filters_dark_outputs(self) -> None:
"""Only connected-with-mode outputs are live."""
scan = OutputScan(
outputs=(
Output("DP-0", connected=True, rect=None),
Output("HDMI-0", connected=True, rect=OutputRect(0, 0, 1, 1)),
Output("DP-1", connected=False, rect=None),
),
source="randr",
ok=True,
)
assert [o.name for o in scan.live] == ["HDMI-0"]

126
gatelock/gatelock/tests/test_outputs_enumerator.py Normal file
View File

@ -0,0 +1,126 @@
"""Tests for OutputEnumerator and the scan it returns.
Split from ``test_outputs.py`` (250-line cap). ``test_outputs_randr.py``
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
from gatelock._outputs import (
Output,
OutputEnumerator,
OutputRect,
OutputScan,
RandrBackend,
enumerate_outputs,
)
# Bound at import, so the autouse hermetic patch (which replaces these
# attributes) does not hide the real implementations from their own tests.
REAL_CREATE = RandrBackend.create.__func__
XRANDR_REAL = """\
Screen 0: minimum 8 x 8, current 6400 x 2160, maximum 32767 x 32767
HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 597mm x 336mm
2560x1440 59.95*+ 143.98 120.00
1920x1080 60.00 59.94
DP-0 connected primary 3840x2160+0+0 (normal left inverted right x axis) 621mm x 341mm
3840x2160 60.00*+ 144.00
DP-1 disconnected (normal left inverted right x axis y axis)
"""
XRANDR_INCIDENT = """\
Screen 0: minimum 8 x 8, current 6400 x 1440, maximum 32767 x 32767
HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 600mm x 330mm
2560x1440 59.95*+
DP-0 connected primary (normal left inverted right x axis y axis) 621mm x 341mm
3840x2160 60.00 +
"""
XRANDR_ALL_DARK = """\
HDMI-0 connected (normal left inverted right x axis)
DP-0 connected primary (normal left inverted right x axis y axis)
"""
XRANDR_ROTATED = "DP-0 connected primary 1080x1920+0+0 left (normal left inverted)\n"
XRANDR_NEGATIVE = "DP-0 connected 1920x1080+-1920+0 (normal left inverted)\n"
XRANDR_ZERO_SIZE = "DP-0 connected 0x0+0+0 (normal left inverted)\n"
XRANDR_MODE_LINES_ONLY = " 2560x1440 59.95*+ 143.98\n 1920x1080 60.00\n"
XRANDR_GARBAGE = "segfault\n"
class TestOutputEnumerator:
"""Tests for the backend ladder."""
def test_prefers_randr(self, mock_root: MagicMock) -> None:
"""When RandR answers, xrandr is never consulted."""
backend = MagicMock()
backend.scan.return_value = (
Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)),
)
scan = OutputEnumerator(mock_root, backend=backend).scan()
assert scan.source == "randr"
assert scan.ok is True
def test_falls_back_to_xrandr(self, mock_root: MagicMock) -> None:
"""A failed RandR scan drops to xrandr."""
backend = MagicMock()
backend.scan.return_value = None
with patch(
"gatelock._outputs.scan_xrandr",
return_value=(Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)),),
):
scan = OutputEnumerator(mock_root, backend=backend).scan()
assert scan.source == "xrandr"
def test_falls_back_to_tk(self, mock_root: MagicMock) -> None:
"""With neither backend, Tk's screen size is the floor."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = OutputEnumerator(mock_root, backend=None).scan()
assert scan.source == "tk"
assert scan.live[0].rect == OutputRect(0, 0, 1920, 1080)
def test_total_failure_is_not_ok(self) -> None:
"""With no root and no backend, the scan reports failure."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = OutputEnumerator(None, backend=None).scan()
assert scan.ok is False
assert scan.source == "none"
assert scan.live == ()
def test_backend_property_and_close(self) -> None:
"""close() releases the backend once and is safe to repeat."""
backend = MagicMock()
enumerator = OutputEnumerator(None, backend=backend)
assert enumerator.backend is backend
enumerator.close()
enumerator.close()
backend.close.assert_called_once_with()
def test_enumerate_outputs_helper(self, mock_root: MagicMock) -> None:
"""The one-shot helper works and cleans up after itself."""
with patch("gatelock._outputs.scan_xrandr", return_value=None):
scan = enumerate_outputs(mock_root)
assert scan.source == "tk"
class TestOutputScan:
"""Tests for OutputScan's live filter."""
def test_live_filters_dark_outputs(self) -> None:
"""Only connected-with-mode outputs are live."""
scan = OutputScan(
outputs=(
Output("DP-0", connected=True, rect=None),
Output("HDMI-0", connected=True, rect=OutputRect(0, 0, 1, 1)),
Output("DP-1", connected=False, rect=None),
),
source="randr",
ok=True,
)
assert [o.name for o in scan.live] == ["HDMI-0"]

204
gatelock/gatelock/tests/test_outputs_randr.py Normal file
View File

@ -0,0 +1,204 @@
"""Tests for the python-xlib RandR backend and the enumerator over it.
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
from gatelock._outputs import (
OutputRect,
RandrBackend,
)
# Bound at import, so the autouse hermetic patch (which replaces these
# attributes) does not hide the real implementations from their own tests.
REAL_CREATE = RandrBackend.create.__func__
XRANDR_REAL = """\
Screen 0: minimum 8 x 8, current 6400 x 2160, maximum 32767 x 32767
HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 597mm x 336mm
2560x1440 59.95*+ 143.98 120.00
1920x1080 60.00 59.94
DP-0 connected primary 3840x2160+0+0 (normal left inverted right x axis) 621mm x 341mm
3840x2160 60.00*+ 144.00
DP-1 disconnected (normal left inverted right x axis y axis)
"""
XRANDR_INCIDENT = """\
Screen 0: minimum 8 x 8, current 6400 x 1440, maximum 32767 x 32767
HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 600mm x 330mm
2560x1440 59.95*+
DP-0 connected primary (normal left inverted right x axis y axis) 621mm x 341mm
3840x2160 60.00 +
"""
XRANDR_ALL_DARK = """\
HDMI-0 connected (normal left inverted right x axis)
DP-0 connected primary (normal left inverted right x axis y axis)
"""
XRANDR_ROTATED = "DP-0 connected primary 1080x1920+0+0 left (normal left inverted)\n"
XRANDR_NEGATIVE = "DP-0 connected 1920x1080+-1920+0 (normal left inverted)\n"
XRANDR_ZERO_SIZE = "DP-0 connected 0x0+0+0 (normal left inverted)\n"
XRANDR_MODE_LINES_ONLY = " 2560x1440 59.95*+ 143.98\n 1920x1080 60.00\n"
XRANDR_GARBAGE = "segfault\n"
def _reply(**fields: object) -> MagicMock:
"""An Xlib reply exposing its fields as attributes, as python-xlib does."""
reply = MagicMock()
for key, value in fields.items():
setattr(reply, key, value)
return reply
def _randr_display(*, outputs: dict[int, dict[str, object]]) -> MagicMock:
"""Build a fake Xlib display exposing ``outputs``."""
display = MagicMock()
root = MagicMock()
root.xrandr_get_screen_resources.return_value = _reply(
config_timestamp=1, outputs=list(outputs)
)
root.xrandr_get_output_primary.return_value = _reply(output=1)
display.screen.return_value.root = root
def output_info(output_id: int, _ts: int) -> MagicMock:
return _reply(**outputs[output_id]["info"])
def crtc_info(crtc: int, _ts: int) -> MagicMock:
return _reply(**outputs[crtc]["crtc"])
display.xrandr_get_output_info.side_effect = output_info
display.xrandr_get_crtc_info.side_effect = crtc_info
return display
class TestRandrBackend:
"""Tests for the RandR backend."""
def test_scan_maps_crtc_to_rect(self) -> None:
"""A connected output with a CRTC and mode becomes a live rect."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 55, "x": 0, "y": 0, "width": 3840, "height": 2160},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.name == "DP-0"
assert output.live is True
assert output.primary is True
assert output.rect == OutputRect(0, 0, 3840, 2160)
def test_no_crtc_is_not_live(self) -> None:
"""crtc == 0 is the dark-monitor case."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}}
)
(output,) = RandrBackend(display).scan() or ()
assert output.connected is True
assert output.rect is None
def test_crtc_without_mode_is_not_live(self) -> None:
"""A CRTC exists but carries no mode: still dark."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 0, "x": 0, "y": 0, "width": 0, "height": 0},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.rect is None
def test_zero_size_crtc_is_not_live(self) -> None:
"""A mode with no pixels is not a screen."""
display = _randr_display(
outputs={
1: {
"info": {"name": "DP-0", "connection": 0, "crtc": 1},
"crtc": {"mode": 55, "x": 0, "y": 0, "width": 0, "height": 1080},
}
}
)
(output,) = RandrBackend(display).scan() or ()
assert output.rect is None
def test_disconnected(self) -> None:
"""connection != 0 means disconnected."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-1", "connection": 1, "crtc": 0}}}
)
(output,) = RandrBackend(display).scan() or ()
assert output.connected is False
def test_scan_failure_returns_none(self) -> None:
"""A protocol error degrades to None instead of raising."""
display = MagicMock()
display.screen.side_effect = OSError("connection reset")
assert RandrBackend(display).scan() is None
def test_primary_query_failure_is_tolerated(self) -> None:
"""RandR refusing to name a primary is not fatal."""
display = _randr_display(
outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}}
)
root = display.screen.return_value.root
root.xrandr_get_output_primary.side_effect = OSError("nope")
(output,) = RandrBackend(display).scan() or ()
assert output.primary is False
def test_close_swallows_errors(self) -> None:
"""Teardown never raises."""
display = MagicMock()
display.close.side_effect = OSError("already gone")
RandrBackend(display).close()
def test_close_normal(self) -> None:
"""A clean close closes the display."""
display = MagicMock()
RandrBackend(display).close()
display.close.assert_called_once_with()
class TestCreateWithoutXlib:
"""RandrBackend.create's degradation path, exercised without the patch."""
def test_import_error_returns_none(self) -> None:
"""A missing python-xlib degrades to None, never an exception.
This is the branch the function-scope import exists to keep reachable.
"""
with patch.dict("sys.modules", {"Xlib": None}):
assert REAL_CREATE(RandrBackend) is None
def test_display_error_returns_none(self) -> None:
"""A display that refuses to open returns None."""
fake_display = MagicMock()
fake_display.Display.side_effect = OSError("cannot connect")
# `from Xlib import display` reads the attribute off the parent
# module, so wiring sys.modules alone is not enough.
fake_xlib = MagicMock(display=fake_display)
with patch.dict(
"sys.modules",
{"Xlib": fake_xlib, "Xlib.display": fake_display},
):
assert REAL_CREATE(RandrBackend) is None
def test_success_returns_backend(self) -> None:
"""A working python-xlib yields a usable backend."""
fake_display = MagicMock()
fake_xlib = MagicMock(display=fake_display)
with patch.dict(
"sys.modules",
{"Xlib": fake_xlib, "Xlib.display": fake_display},
):
backend = REAL_CREATE(RandrBackend)
assert isinstance(backend, RandrBackend)

143
gatelock/gatelock/tests/test_scrollable.py
View File

@ -20,8 +20,6 @@ import tkinter as tk
import pytest
from gatelock._scrollable import ScrollableSurface
from gatelock._scrollkeys import bind_scroll_keys
from gatelock._userintent import viewport_targets
from gatelock._window import LockConfig
pytestmark = pytest.mark.skipif(
@ -231,144 +229,3 @@ def test_a_surface_without_a_scrollbar_still_scrolls() -> None:
assert made.canvas.yview()[0] > 0.0
finally:
root.destroy()
def test_scroll_into_view_ignores_an_unmapped_widget(
surface: ScrollableSurface,
) -> None:
"""A widget Tk has not laid out yet has no position to scroll to."""
ghost = tk.Entry(surface.content) # never packed: zero height
before = _offset(surface)
surface.scroll_into_view(ghost)
assert _offset(surface) == before
def test_scroll_into_view_leaves_an_already_visible_widget_alone(
surface: ScrollableSurface,
) -> None:
"""Nothing moves when the widget is already fully on screen."""
before = _offset(surface)
surface.scroll_into_view(surface.content.winfo_children()[0])
assert _offset(surface) == before
def test_scroll_into_view_reaches_backwards(surface: ScrollableSurface) -> None:
"""Revealing a widget above the viewport scrolls up, not down."""
surface.canvas.yview_moveto(1.0)
surface.canvas.update()
surface.scroll_into_view(surface.content.winfo_children()[0])
assert _offset(surface) == 0.0
def test_paging_degrades_to_the_canvas_without_a_toplevel(
surface: ScrollableSurface, caplog: pytest.LogCaptureFixture
) -> None:
"""A widget tree with no real toplevel still scrolls by keyboard.
Reachable under a mocked Tk root. Binding nothing at all would leave the
overflow unreachable, so it degrades to canvas-local bindings and says so
-- a silent degradation here is content the user cannot get to.
"""
message = "no toplevel"
def _no_toplevel() -> None:
raise tk.TclError(message)
surface.canvas.winfo_toplevel = _no_toplevel
with caplog.at_level("WARNING"):
bind_scroll_keys(surface.canvas, surface.content)
surface._intent.bind(viewport_targets(surface.canvas, surface.content))
assert "paging keys bound to the canvas only" in caplog.text
assert "user-input tracking is bound to the viewport only" in caplog.text
surface.canvas.focus_set()
surface.canvas.event_generate("<Next>", when="now")
surface.canvas.update()
assert _offset(surface) > 0.0
def test_a_second_finalize_does_not_stack_handlers(
surface: ScrollableSurface,
) -> None:
"""finalize() is documented idempotent, and a repaint calls it again.
An earlier version re-bound <FocusIn> on every call, so handlers piled up
one per repaint across the ~19 repaint sites of a single lock session.
"""
bound_once = set(surface._focus_bound)
surface.finalize()
assert set(surface._focus_bound) == bound_once
def test_scroll_into_view_does_nothing_when_everything_fits() -> None:
"""Nothing to reveal on a screen that is fully visible already."""
root = tk.Tk()
try:
root.geometry(f"400x{_VIEWPORT_H}+0+0")
root.focus_force()
root.update_idletasks()
made = ScrollableSurface(root, LockConfig())
made.container.place(relx=0, rely=0, relwidth=1, relheight=1)
entry = tk.Entry(made.content)
entry.pack()
made.finalize()
root.update()
made.scroll_into_view(entry)
assert made.canvas.yview() == (0.0, 1.0)
finally:
root.destroy()
def test_scroll_into_view_holds_still_for_a_widget_in_the_middle(
surface: ScrollableSurface,
) -> None:
"""A widget already fully on screen is not dragged to an edge.
The margin logic has three outcomes -- scroll up, scroll down, hold -- and
"hold" is the one a viewport that jitters would get wrong.
"""
middle = surface.content.winfo_children()[3]
before = _offset(surface)
surface.scroll_into_view(middle)
assert _offset(surface) == before
def test_the_wheel_scrolls_from_over_the_content(surface: ScrollableSurface) -> None:
"""Wheel-over-text scrolls, which is the only way anyone uses a mouse here.
Tk dispatches through *bindtags*, not the parent chain: a child's tags are
``(itself, its class, the toplevel, "all")``, so a binding on ``content``
is never consulted for an event over a Label sitting on top of it. Content
that fills the viewport covers that frame entirely, so binding the wheel
only on the canvas and the content frame makes wheel scrolling dead
everywhere the user would actually point.
"""
child = surface.content.winfo_children()[10]
before = surface.canvas.yview()[0]
child.event_generate("<Button-5>")
surface.canvas.update()
assert surface.canvas.yview()[0] > before
def test_the_wheel_scrolls_back_up_from_over_the_content(
surface: ScrollableSurface,
) -> None:
"""The up notch is bound on descendants too, not just the down one."""
child = surface.content.winfo_children()[10]
child.event_generate("<Button-5>")
surface.canvas.update()
scrolled = surface.canvas.yview()[0]
child.event_generate("<Button-4>")
surface.canvas.update()
assert surface.canvas.yview()[0] < scrolled

197
gatelock/gatelock/tests/test_scrollable_into_view.py Normal file
View File

@ -0,0 +1,197 @@
"""Tests for scrolling a specific widget into view.
Split from ``test_scrollable.py`` (250-line cap), which keeps the fit states
and the scroll gestures themselves. These cover ``scroll_into_view`` and the
focus-following that calls it.
"""
from __future__ import annotations
import os
import tkinter as tk
import pytest
from gatelock._scrollable import ScrollableSurface
from gatelock._scrollkeys import bind_scroll_keys
from gatelock._userintent import viewport_targets
from gatelock._window import LockConfig
pytestmark = pytest.mark.skipif(
not os.environ.get("DISPLAY"),
reason="needs a real X display; run under xvfb-run in a headless checkout",
)
_VIEWPORT_H = 200
_TALL = 60
@pytest.fixture
def surface() -> tk.Misc:
"""A 400x200 viewport holding 20 stacked entries -- deliberately taller."""
root = tk.Tk()
# A managed window has its geometry rewritten wholesale by a real window
# manager (i3 confirmed: a plain geometry() request came back tiled to
# the pane size instead of 400x200, so nothing ever overflowed and every
# scroll assertion in this file failed against a real WM session) --
# exactly the reason production lock windows are overrideredirect too.
root.overrideredirect(boolean=True)
root.geometry(f"400x{_VIEWPORT_H}+0+0")
# No window manager under Xvfb, so the toplevel must claim X input focus
# before any synthetic key event is delivered anywhere.
root.focus_force()
root.update_idletasks()
made = ScrollableSurface(root, LockConfig())
made.container.place(relx=0, rely=0, relwidth=1, relheight=1)
for index in range(_TALL):
tk.Entry(made.content, name=f"field{index}").pack()
made.finalize()
root.update()
yield made
root.destroy()
def _offset(made: ScrollableSurface) -> float:
"""The viewport's current scroll fraction."""
return made.canvas.yview()[0]
def test_scroll_into_view_ignores_an_unmapped_widget(
surface: ScrollableSurface,
) -> None:
"""A widget Tk has not laid out yet has no position to scroll to."""
ghost = tk.Entry(surface.content) # never packed: zero height
before = _offset(surface)
surface.scroll_into_view(ghost)
assert _offset(surface) == before
def test_scroll_into_view_leaves_an_already_visible_widget_alone(
surface: ScrollableSurface,
) -> None:
"""Nothing moves when the widget is already fully on screen."""
before = _offset(surface)
surface.scroll_into_view(surface.content.winfo_children()[0])
assert _offset(surface) == before
def test_scroll_into_view_reaches_backwards(surface: ScrollableSurface) -> None:
"""Revealing a widget above the viewport scrolls up, not down."""
surface.canvas.yview_moveto(1.0)
surface.canvas.update()
surface.scroll_into_view(surface.content.winfo_children()[0])
assert _offset(surface) == 0.0
def test_paging_degrades_to_the_canvas_without_a_toplevel(
surface: ScrollableSurface, caplog: pytest.LogCaptureFixture
) -> None:
"""A widget tree with no real toplevel still scrolls by keyboard.
Reachable under a mocked Tk root. Binding nothing at all would leave the
overflow unreachable, so it degrades to canvas-local bindings and says so
-- a silent degradation here is content the user cannot get to.
"""
message = "no toplevel"
def _no_toplevel() -> None:
raise tk.TclError(message)
surface.canvas.winfo_toplevel = _no_toplevel
with caplog.at_level("WARNING"):
bind_scroll_keys(surface.canvas, surface.content)
surface._intent.bind(viewport_targets(surface.canvas, surface.content))
assert "paging keys bound to the canvas only" in caplog.text
assert "user-input tracking is bound to the viewport only" in caplog.text
surface.canvas.focus_set()
surface.canvas.event_generate("<Next>", when="now")
surface.canvas.update()
assert _offset(surface) > 0.0
def test_a_second_finalize_does_not_stack_handlers(
surface: ScrollableSurface,
) -> None:
"""finalize() is documented idempotent, and a repaint calls it again.
An earlier version re-bound <FocusIn> on every call, so handlers piled up
one per repaint across the ~19 repaint sites of a single lock session.
"""
bound_once = set(surface._focus_bound)
surface.finalize()
assert set(surface._focus_bound) == bound_once
def test_scroll_into_view_does_nothing_when_everything_fits() -> None:
"""Nothing to reveal on a screen that is fully visible already."""
root = tk.Tk()
try:
root.geometry(f"400x{_VIEWPORT_H}+0+0")
root.focus_force()
root.update_idletasks()
made = ScrollableSurface(root, LockConfig())
made.container.place(relx=0, rely=0, relwidth=1, relheight=1)
entry = tk.Entry(made.content)
entry.pack()
made.finalize()
root.update()
made.scroll_into_view(entry)
assert made.canvas.yview() == (0.0, 1.0)
finally:
root.destroy()
def test_scroll_into_view_holds_still_for_a_widget_in_the_middle(
surface: ScrollableSurface,
) -> None:
"""A widget already fully on screen is not dragged to an edge.
The margin logic has three outcomes -- scroll up, scroll down, hold -- and
"hold" is the one a viewport that jitters would get wrong.
"""
middle = surface.content.winfo_children()[3]
before = _offset(surface)
surface.scroll_into_view(middle)
assert _offset(surface) == before
def test_the_wheel_scrolls_from_over_the_content(surface: ScrollableSurface) -> None:
"""Wheel-over-text scrolls, which is the only way anyone uses a mouse here.
Tk dispatches through *bindtags*, not the parent chain: a child's tags are
``(itself, its class, the toplevel, "all")``, so a binding on ``content``
is never consulted for an event over a Label sitting on top of it. Content
that fills the viewport covers that frame entirely, so binding the wheel
only on the canvas and the content frame makes wheel scrolling dead
everywhere the user would actually point.
"""
child = surface.content.winfo_children()[10]
before = surface.canvas.yview()[0]
child.event_generate("<Button-5>")
surface.canvas.update()
assert surface.canvas.yview()[0] > before
def test_the_wheel_scrolls_back_up_from_over_the_content(
surface: ScrollableSurface,
) -> None:
"""The up notch is bound on descendants too, not just the down one."""
child = surface.content.winfo_children()[10]
child.event_generate("<Button-5>")
surface.canvas.update()
scrolled = surface.canvas.yview()[0]
child.event_generate("<Button-4>")
surface.canvas.update()
assert surface.canvas.yview()[0] < scrolled

118
gatelock/gatelock/tests/test_surfaces.py
View File

@ -8,10 +8,7 @@ import pytest
from gatelock._outputs import Output, OutputRect, OutputScan
from gatelock._surfaces import (
SurfaceInfo,
SurfaceSet,
TextMirror,
mirror_text_widgets,
needs_backdrop_root,
)
from gatelock._window import LockConfig
@ -165,118 +162,3 @@ class TestEnforce:
surfaces.raise_all()
for surface in surfaces._surfaces.values():
surface.window.lift.assert_called()
class TestFocus:
"""Which surface takes the keyboard."""
def test_focus_surface_by_index(self, surfaces: SurfaceSet) -> None:
"""Focusing an index returns that surface's info."""
surfaces.apply(scan(DP0, HDMI))
info = surfaces.focus_surface(1)
assert info is not None
assert info.output_name == "HDMI-0"
def test_focus_unknown_index(self, surfaces: SurfaceSet) -> None:
"""An index with no surface returns None."""
surfaces.apply(scan(DP0))
assert surfaces.focus_surface(9) is None
def test_preferred_index_is_the_primary(self, surfaces: SurfaceSet) -> None:
"""The live primary output takes initial focus."""
surfaces.apply(scan(HDMI, DP0))
assert surfaces.preferred_focus_index() == 1
def test_preferred_index_defaults_to_first(self, surfaces: SurfaceSet) -> None:
"""With no primary marked, the first surface wins."""
surfaces.apply(scan(HDMI))
assert surfaces.preferred_focus_index() == 0
class TestBackdropAndTeardown:
"""The root backdrop and full teardown."""
def test_backdrop_covers_the_whole_screen(
self, surfaces: SurfaceSet, mock_root: MagicMock
) -> None:
"""The backdrop spans the X screen, covering dead columns."""
surfaces.update_backdrop()
mock_root.geometry.assert_called_once_with("1920x1080+0+0")
def test_soft_mode_has_no_backdrop(self, mock_root: MagicMock) -> None:
"""Without a grab or overrideredirect there is nothing to back."""
SurfaceSet(mock_root, LockConfig(mode="soft"), MagicMock()).update_backdrop()
mock_root.geometry.assert_not_called()
def test_destroy_all(self, surfaces: SurfaceSet) -> None:
"""Closing tears down every surface."""
surfaces.apply(scan(DP0, HDMI))
surfaces.destroy_all()
assert surfaces.names() == frozenset()
class TestTextMirror:
"""tk.Text has no textvariable, so it needs explicit mirroring."""
def _widget(self, text: str = "") -> MagicMock:
widget = MagicMock()
state = {"text": text}
widget.get.side_effect = lambda *_a: state["text"]
widget.delete.side_effect = lambda *_a: state.update(text="")
widget.insert.side_effect = lambda _i, value: state.update(text=value)
widget._state = state
return widget
def test_typing_in_one_updates_the_other(self) -> None:
"""A key release in surface A reaches surface B."""
a, b = self._widget("hello"), self._widget()
var = MagicMock()
var.get.return_value = "hello"
mirror = TextMirror([a, b], var)
event = MagicMock()
event.widget = a
mirror._on_key_release(event)
var.set.assert_called_once_with("hello")
assert b._state["text"] == "hello"
def test_var_write_fans_out(self) -> None:
"""Setting the shared variable updates every widget."""
a, b = self._widget(), self._widget()
var = MagicMock()
var.get.return_value = "typed"
mirror = TextMirror([a, b], var)
mirror._on_var_write("x", "", "w")
assert a._state["text"] == "typed"
assert b._state["text"] == "typed"
def test_reentrancy_is_blocked(self) -> None:
"""A fan-out must not trigger another fan-out."""
a = self._widget()
var = MagicMock()
var.get.return_value = "v"
mirror = TextMirror([a], var)
mirror._syncing = True
mirror._on_var_write("x", "", "w")
assert a._state["text"] == ""
event = MagicMock()
event.widget = a
mirror._on_key_release(event)
var.set.assert_not_called()
def test_helper_returns_mirror(self) -> None:
"""The public helper wires the widgets up."""
widget = self._widget()
var = MagicMock()
assert isinstance(mirror_text_widgets([widget], var), TextMirror)
widget.bind.assert_called_once()
class TestSurfaceInfo:
"""SurfaceInfo equality drives the move-vs-noop decision."""
def test_identical_infos_compare_equal(self) -> None:
"""Equality is by value, so an unchanged layout is a no-op."""
rect = OutputRect(0, 0, 100, 100)
first = SurfaceInfo("DP-0", rect, 0, is_primary=True)
second = SurfaceInfo("DP-0", rect, 0, is_primary=True)
assert first == second

153
gatelock/gatelock/tests/test_surfaces_focus.py Normal file
View File

@ -0,0 +1,153 @@
"""Tests for focus handling, the backdrop, teardown and the text mirror.
Split from ``test_surfaces.py`` to hold the shared 250-line cap.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from gatelock._outputs import Output, OutputRect, OutputScan
from gatelock._surfaces import (
SurfaceInfo,
SurfaceSet,
TextMirror,
mirror_text_widgets,
)
from gatelock._window import LockConfig
DP0 = Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True)
HDMI = Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440))
DP0_DARK = Output("DP-0", connected=True, rect=None, primary=True)
DP0_MOVED = Output(
"DP-0", connected=True, rect=OutputRect(0, 0, 1920, 1080), primary=True
)
def scan(*outputs: Output) -> OutputScan:
"""Build a successful scan over ``outputs``."""
return OutputScan(outputs=outputs, source="randr", ok=True)
@pytest.fixture
def surfaces(mock_root: MagicMock) -> SurfaceSet:
"""A hard-mode surface set over a mock root."""
return SurfaceSet(mock_root, LockConfig(mode="hard"), MagicMock())
class TestFocus:
"""Which surface takes the keyboard."""
def test_focus_surface_by_index(self, surfaces: SurfaceSet) -> None:
"""Focusing an index returns that surface's info."""
surfaces.apply(scan(DP0, HDMI))
info = surfaces.focus_surface(1)
assert info is not None
assert info.output_name == "HDMI-0"
def test_focus_unknown_index(self, surfaces: SurfaceSet) -> None:
"""An index with no surface returns None."""
surfaces.apply(scan(DP0))
assert surfaces.focus_surface(9) is None
def test_preferred_index_is_the_primary(self, surfaces: SurfaceSet) -> None:
"""The live primary output takes initial focus."""
surfaces.apply(scan(HDMI, DP0))
assert surfaces.preferred_focus_index() == 1
def test_preferred_index_defaults_to_first(self, surfaces: SurfaceSet) -> None:
"""With no primary marked, the first surface wins."""
surfaces.apply(scan(HDMI))
assert surfaces.preferred_focus_index() == 0
class TestBackdropAndTeardown:
"""The root backdrop and full teardown."""
def test_backdrop_covers_the_whole_screen(
self, surfaces: SurfaceSet, mock_root: MagicMock
) -> None:
"""The backdrop spans the X screen, covering dead columns."""
surfaces.update_backdrop()
mock_root.geometry.assert_called_once_with("1920x1080+0+0")
def test_soft_mode_has_no_backdrop(self, mock_root: MagicMock) -> None:
"""Without a grab or overrideredirect there is nothing to back."""
SurfaceSet(mock_root, LockConfig(mode="soft"), MagicMock()).update_backdrop()
mock_root.geometry.assert_not_called()
def test_destroy_all(self, surfaces: SurfaceSet) -> None:
"""Closing tears down every surface."""
surfaces.apply(scan(DP0, HDMI))
surfaces.destroy_all()
assert surfaces.names() == frozenset()
class TestTextMirror:
"""tk.Text has no textvariable, so it needs explicit mirroring."""
def _widget(self, text: str = "") -> MagicMock:
widget = MagicMock()
state = {"text": text}
widget.get.side_effect = lambda *_a: state["text"]
widget.delete.side_effect = lambda *_a: state.update(text="")
widget.insert.side_effect = lambda _i, value: state.update(text=value)
widget._state = state
return widget
def test_typing_in_one_updates_the_other(self) -> None:
"""A key release in surface A reaches surface B."""
a, b = self._widget("hello"), self._widget()
var = MagicMock()
var.get.return_value = "hello"
mirror = TextMirror([a, b], var)
event = MagicMock()
event.widget = a
mirror._on_key_release(event)
var.set.assert_called_once_with("hello")
assert b._state["text"] == "hello"
def test_var_write_fans_out(self) -> None:
"""Setting the shared variable updates every widget."""
a, b = self._widget(), self._widget()
var = MagicMock()
var.get.return_value = "typed"
mirror = TextMirror([a, b], var)
mirror._on_var_write("x", "", "w")
assert a._state["text"] == "typed"
assert b._state["text"] == "typed"
def test_reentrancy_is_blocked(self) -> None:
"""A fan-out must not trigger another fan-out."""
a = self._widget()
var = MagicMock()
var.get.return_value = "v"
mirror = TextMirror([a], var)
mirror._syncing = True
mirror._on_var_write("x", "", "w")
assert a._state["text"] == ""
event = MagicMock()
event.widget = a
mirror._on_key_release(event)
var.set.assert_not_called()
def test_helper_returns_mirror(self) -> None:
"""The public helper wires the widgets up."""
widget = self._widget()
var = MagicMock()
assert isinstance(mirror_text_widgets([widget], var), TextMirror)
widget.bind.assert_called_once()
class TestSurfaceInfo:
"""SurfaceInfo equality drives the move-vs-noop decision."""
def test_identical_infos_compare_equal(self) -> None:
"""Equality is by value, so an unchanged layout is a no-op."""
rect = OutputRect(0, 0, 100, 100)
first = SurfaceInfo("DP-0", rect, 0, is_primary=True)
second = SurfaceInfo("DP-0", rect, 0, is_primary=True)
assert first == second

126
gatelock/gatelock/tests/test_window.py
View File

@ -141,129 +141,3 @@ class TestGrabInput:
mock_root.grab_set.assert_not_called()
mock_root.grab_set_global.assert_not_called()
class TestAcquireGlobalGrab:
"""Tests for LockWindow._acquire_global_grab."""
def test_success_focuses_and_notifies(self, mock_root: MagicMock) -> None:
"""A successful grab forces focus and notifies the hook."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1)
mock_root.grab_set_global.assert_called_once_with()
mock_root.focus_force.assert_called_once_with()
hooks.on_focus_ready.assert_called_once_with(None)
def test_success_swallows_tclerror_from_focus(self, mock_root: MagicMock) -> None:
"""A TclError while focusing after a successful grab is swallowed."""
mock_root.focus_force.side_effect = tk.TclError("gone")
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1) # must not raise
hooks.on_focus_ready.assert_not_called()
def test_failure_with_retry_zero_falls_back_to_local(
self, mock_root: MagicMock
) -> None:
"""grab_retry_ms=0 falls back to a local grab on the first failure."""
mock_root.grab_set_global.side_effect = tk.TclError("held by another client")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=0)
)
window._acquire_global_grab(attempt=1)
mock_root.grab_set.assert_called_once_with()
mock_root.after.assert_not_called()
def test_failure_with_retry_zero_swallows_local_grab_tclerror(
self, mock_root: MagicMock
) -> None:
"""The local-grab fallback itself swallows a TclError too."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
mock_root.grab_set.side_effect = tk.TclError("also gone")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=0)
)
window._acquire_global_grab(attempt=1) # must not raise
def test_failure_with_default_retry_schedules_retry(
self, mock_root: MagicMock
) -> None:
"""Default (None) retry interval reschedules every 200ms, logging every 25th."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=24)
assert mock_root.after.call_count == 1
scheduled_delay = mock_root.after.call_args[0][0]
assert scheduled_delay == 200
def test_failure_logs_every_grab_log_every_attempts(
self, mock_root: MagicMock
) -> None:
"""A warning is logged only when attempt is a multiple of grab_log_every."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_log_every=5)
)
with patch("gatelock._preempt._logger") as mock_logger:
window._acquire_global_grab(attempt=5)
mock_logger.warning.assert_called_once()
with patch("gatelock._preempt._logger") as mock_logger:
window._acquire_global_grab(attempt=3)
mock_logger.warning.assert_not_called()
def test_failure_with_custom_retry_ms_uses_it(self, mock_root: MagicMock) -> None:
"""An explicit positive grab_retry_ms is used as the reschedule delay."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=50)
)
window._acquire_global_grab(attempt=1)
scheduled_delay = mock_root.after.call_args[0][0]
assert scheduled_delay == 50
def test_rescheduled_callback_increments_attempt(
self, mock_root: MagicMock
) -> None:
"""The rescheduled callback re-invokes with attempt + 1."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1)
scheduled_callback = mock_root.after.call_args[0][1]
with patch("gatelock._arming.acquire_global_grab") as mock_acquire:
scheduled_callback()
mock_acquire.assert_called_once_with(window.root, window._arming, attempt=2)
class TestNotifyFocusReady:
"""Tests for LockWindow._notify_focus_ready."""
def test_calls_hook(self, mock_root: MagicMock) -> None:
"""The on_focus_ready hook is invoked."""
window, hooks = make_window(mock_root)
window._notify_focus_ready()
hooks.on_focus_ready.assert_called_once_with(None)
def test_swallows_tclerror_from_hook(self, mock_root: MagicMock) -> None:
"""A TclError raised by the hook (widget already destroyed) is swallowed."""
hooks = MagicMock()
hooks.on_focus_ready.side_effect = tk.TclError("destroyed")
window, _hooks = make_window(mock_root, hooks=hooks)
window._notify_focus_ready() # must not raise

139
gatelock/gatelock/tests/test_window_grab.py Normal file
View File

@ -0,0 +1,139 @@
"""Tests for acquiring the global input grab and its retry loop.
Split from ``test_window.py`` to hold the shared 250-line cap.
"""
from __future__ import annotations
import tkinter as tk
from unittest.mock import MagicMock, patch
from gatelock._window import LockConfig
from gatelock.tests.conftest import make_window
class TestAcquireGlobalGrab:
"""Tests for LockWindow._acquire_global_grab."""
def test_success_focuses_and_notifies(self, mock_root: MagicMock) -> None:
"""A successful grab forces focus and notifies the hook."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1)
mock_root.grab_set_global.assert_called_once_with()
mock_root.focus_force.assert_called_once_with()
hooks.on_focus_ready.assert_called_once_with(None)
def test_success_swallows_tclerror_from_focus(self, mock_root: MagicMock) -> None:
"""A TclError while focusing after a successful grab is swallowed."""
mock_root.focus_force.side_effect = tk.TclError("gone")
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1) # must not raise
hooks.on_focus_ready.assert_not_called()
def test_failure_with_retry_zero_falls_back_to_local(
self, mock_root: MagicMock
) -> None:
"""grab_retry_ms=0 falls back to a local grab on the first failure."""
mock_root.grab_set_global.side_effect = tk.TclError("held by another client")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=0)
)
window._acquire_global_grab(attempt=1)
mock_root.grab_set.assert_called_once_with()
mock_root.after.assert_not_called()
def test_failure_with_retry_zero_swallows_local_grab_tclerror(
self, mock_root: MagicMock
) -> None:
"""The local-grab fallback itself swallows a TclError too."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
mock_root.grab_set.side_effect = tk.TclError("also gone")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=0)
)
window._acquire_global_grab(attempt=1) # must not raise
def test_failure_with_default_retry_schedules_retry(
self, mock_root: MagicMock
) -> None:
"""Default (None) retry interval reschedules every 200ms, logging every 25th."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=24)
assert mock_root.after.call_count == 1
scheduled_delay = mock_root.after.call_args[0][0]
assert scheduled_delay == 200
def test_failure_logs_every_grab_log_every_attempts(
self, mock_root: MagicMock
) -> None:
"""A warning is logged only when attempt is a multiple of grab_log_every."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_log_every=5)
)
with patch("gatelock._preempt._logger") as mock_logger:
window._acquire_global_grab(attempt=5)
mock_logger.warning.assert_called_once()
with patch("gatelock._preempt._logger") as mock_logger:
window._acquire_global_grab(attempt=3)
mock_logger.warning.assert_not_called()
def test_failure_with_custom_retry_ms_uses_it(self, mock_root: MagicMock) -> None:
"""An explicit positive grab_retry_ms is used as the reschedule delay."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", grab_retry_ms=50)
)
window._acquire_global_grab(attempt=1)
scheduled_delay = mock_root.after.call_args[0][0]
assert scheduled_delay == 50
def test_rescheduled_callback_increments_attempt(
self, mock_root: MagicMock
) -> None:
"""The rescheduled callback re-invokes with attempt + 1."""
mock_root.grab_set_global.side_effect = tk.TclError("held")
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._acquire_global_grab(attempt=1)
scheduled_callback = mock_root.after.call_args[0][1]
with patch("gatelock._arming.acquire_global_grab") as mock_acquire:
scheduled_callback()
mock_acquire.assert_called_once_with(window.root, window._arming, attempt=2)
class TestNotifyFocusReady:
"""Tests for LockWindow._notify_focus_ready."""
def test_calls_hook(self, mock_root: MagicMock) -> None:
"""The on_focus_ready hook is invoked."""
window, hooks = make_window(mock_root)
window._notify_focus_ready()
hooks.on_focus_ready.assert_called_once_with(None)
def test_swallows_tclerror_from_hook(self, mock_root: MagicMock) -> None:
"""A TclError raised by the hook (widget already destroyed) is swallowed."""
hooks = MagicMock()
hooks.on_focus_ready.side_effect = tk.TclError("destroyed")
window, _hooks = make_window(mock_root, hooks=hooks)
window._notify_focus_ready() # must not raise

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

@ -2,15 +2,11 @@
from __future__ import annotations
import queue
import signal
from unittest.mock import MagicMock, patch
from gatelock import _preempt
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Claim,
)
from gatelock._detect import _RandrEventSource
@ -112,230 +108,3 @@ class TestGrabBlockedLogging:
with patch("gatelock._preempt._logger") as logger:
window._log_grab_blocked(50)
assert "fullscreen game" in str(logger.warning.call_args)
class TestPreemptWeakerHolder:
"""A weaker incumbent must not block a stronger claim indefinitely."""
def _arbiter_with(
self, *, our_rank: int, holder_rank: int, holder_pid: int
) -> MagicMock:
arbiter = MagicMock()
arbiter.claim = Claim(
app="ours",
rank=our_rank,
pid=1,
started="2026-08-21T12:00:00+00:00",
grab="global",
disable_vt=True,
instance_id="ours",
)
arbiter.describe_holder.return_value = Claim(
app="holder",
rank=holder_rank,
pid=holder_pid,
started="2026-08-21T07:04:07+00:00",
grab="global",
disable_vt=True,
instance_id="theirs",
)
return arbiter
def test_signals_a_weaker_holder(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_called_once_with(4075, signal.SIGTERM)
def test_never_signals_a_stronger_holder(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_DIET_GUARD, holder_rank=RANK_SCREEN_LOCKER, holder_pid=4137
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_signals_the_same_holder_only_once(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
window._log_grab_blocked(50)
kill.assert_called_once_with(4075, signal.SIGTERM)
def test_disabled_by_config(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=False)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_off_by_default(self, mock_root: MagicMock) -> None:
"""Preemption is opt-in, so no app can evict anyone by accident.
wake_alarm outranks screen_locker and builds a plain hard LockConfig.
While the default was True, that combination SIGTERMed the *armed
workout lock*, unlocking enforcement with no workout logged.
"""
arbiter = self._arbiter_with(
our_rank=RANK_WAKE_ALARM,
holder_rank=RANK_SCREEN_LOCKER,
holder_pid=4137,
)
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_no_arbiter_means_no_preemption(self, mock_root: MagicMock) -> None:
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
with patch("gatelock._preempt.os.kill") as kill:
_preempt.maybe_preempt(
Claim(
app="holder",
rank=RANK_DIET_GUARD,
pid=4075,
started="2026-08-21T07:04:07+00:00",
grab="global",
disable_vt=True,
instance_id="theirs",
),
arbiter=None,
config=window._config,
preempted_pids=set(),
)
kill.assert_not_called()
def test_a_holder_that_is_already_gone_is_reported_at_info(
self, mock_root: MagicMock
) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with (
patch("gatelock._preempt.os.kill", side_effect=ProcessLookupError),
patch("gatelock._preempt._logger") as logger,
):
window._log_grab_blocked(25)
assert logger.info.called
def test_an_unsignalable_holder_is_reported_not_raised(
self, mock_root: MagicMock
) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with (
patch("gatelock._preempt.os.kill", side_effect=OSError("nope")),
patch("gatelock._preempt._logger") as logger,
):
window._log_grab_blocked(25)
assert logger.exception.called
class TestFocusNotifiedOnce:
"""Both call sites are load-bearing, so the guard is on the callee."""
def test_second_notification_is_suppressed(self, mock_root: MagicMock) -> None:
"""The hook fires once even though two timings can reach it."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._notify_focus_ready()
window._notify_focus_ready()
hooks.on_focus_ready.assert_called_once()
class TestCloseReleasesTheScreen:
"""A clean exit must hand the screen to the next app."""
def test_arbiter_is_released(self, mock_root: MagicMock) -> None:
"""Without this the alarm -> workout handoff strands the workout lock."""
arbiter = MagicMock()
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._arbiter = arbiter
window.close()
arbiter.release.assert_called_once_with()
def test_close_without_arbiter(self, mock_root: MagicMock) -> None:
"""An unarbitrated lock still closes cleanly."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window.close()
hooks.on_close.assert_called_once_with()
class TestRemainingBranches:
"""Small paths the other suites do not reach."""
def test_claim_read_oserror(self, tmp_path: object) -> None:
"""A claim whose contents cannot be read is treated as absent."""
from gatelock._arbiter import Arbiter
arbiter = Arbiter("a", RANK_DIET_GUARD, grab="none", disable_vt=False)
arbiter.publish()
handle = MagicMock()
handle.__enter__.return_value = handle
handle.read.side_effect = OSError("io error")
with (
patch("gatelock._arbiter.Path.open", return_value=handle),
patch("gatelock._arbiter._try_lock", return_value=False),
):
assert arbiter.live_claims() == ()
arbiter.release()
def test_randr_start_spawns_thread(self) -> None:
"""A successful connect starts the watcher thread."""
source = _RandrEventSource(queue.Queue())
with (
patch.object(source, "_connect", return_value=True),
patch("gatelock._detect.threading.Thread") as thread_cls,
):
assert REAL_RANDR_START(source) is True
thread_cls.return_value.start.assert_called_once_with()
def test_randr_loop_ignores_empty_batch(self) -> None:
"""A readable socket with no events posts no notice."""
sink: queue.Queue[str] = queue.Queue()
source = _RandrEventSource(sink)
source._display = MagicMock()
source._display.pending_events.return_value = 0
calls = {"n": 0}
def is_set() -> bool:
calls["n"] += 1
return calls["n"] > 1
source._stop = MagicMock()
source._stop.is_set.side_effect = is_set
with patch("gatelock._detect.select.select", return_value=([1], [], [])):
source._loop()
assert sink.empty()

104
gatelock/gatelock/tests/test_window_surfaces_lifecycle.py Normal file
View File

@ -0,0 +1,104 @@
"""Tests for focus notification and close on a lock window.
Split from ``test_window_surfaces.py`` to hold the shared 250-line cap.
"""
from __future__ import annotations
import queue
from unittest.mock import MagicMock, patch
from gatelock._arbiter import (
RANK_DIET_GUARD,
)
from gatelock._detect import _RandrEventSource
from gatelock._outputs import Output, OutputRect
from gatelock._window import LockConfig
from gatelock.tests.conftest import make_window
# Captured before the autouse hermetic fixture replaces it.
REAL_RANDR_START = _RandrEventSource.start
DARK_ONLY = (Output("DP-0", connected=True, rect=None, primary=True),)
TWO_LIVE = (
Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True),
Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)),
)
class TestFocusNotifiedOnce:
"""Both call sites are load-bearing, so the guard is on the callee."""
def test_second_notification_is_suppressed(self, mock_root: MagicMock) -> None:
"""The hook fires once even though two timings can reach it."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._notify_focus_ready()
window._notify_focus_ready()
hooks.on_focus_ready.assert_called_once()
class TestCloseReleasesTheScreen:
"""A clean exit must hand the screen to the next app."""
def test_arbiter_is_released(self, mock_root: MagicMock) -> None:
"""Without this the alarm -> workout handoff strands the workout lock."""
arbiter = MagicMock()
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._arbiter = arbiter
window.close()
arbiter.release.assert_called_once_with()
def test_close_without_arbiter(self, mock_root: MagicMock) -> None:
"""An unarbitrated lock still closes cleanly."""
window, hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window.close()
hooks.on_close.assert_called_once_with()
class TestRemainingBranches:
"""Small paths the other suites do not reach."""
def test_claim_read_oserror(self, tmp_path: object) -> None:
"""A claim whose contents cannot be read is treated as absent."""
from gatelock._arbiter import Arbiter
arbiter = Arbiter("a", RANK_DIET_GUARD, grab="none", disable_vt=False)
arbiter.publish()
handle = MagicMock()
handle.__enter__.return_value = handle
handle.read.side_effect = OSError("io error")
with (
patch("gatelock._arbiter.Path.open", return_value=handle),
patch("gatelock._arbiter._try_lock", return_value=False),
):
assert arbiter.live_claims() == ()
arbiter.release()
def test_randr_start_spawns_thread(self) -> None:
"""A successful connect starts the watcher thread."""
source = _RandrEventSource(queue.Queue())
with (
patch.object(source, "_connect", return_value=True),
patch("gatelock._detect.threading.Thread") as thread_cls,
):
assert REAL_RANDR_START(source) is True
thread_cls.return_value.start.assert_called_once_with()
def test_randr_loop_ignores_empty_batch(self) -> None:
"""A readable socket with no events posts no notice."""
sink: queue.Queue[str] = queue.Queue()
source = _RandrEventSource(sink)
source._display = MagicMock()
source._display.pending_events.return_value = 0
calls = {"n": 0}
def is_set() -> bool:
calls["n"] += 1
return calls["n"] > 1
source._stop = MagicMock()
source._stop.is_set.side_effect = is_set
with patch("gatelock._detect.select.select", return_value=([1], [], [])):
source._loop()
assert sink.empty()

182
gatelock/gatelock/tests/test_window_surfaces_preempt.py Normal file
View File

@ -0,0 +1,182 @@
"""Tests for standing a weaker grab holder down.
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
from unittest.mock import MagicMock, patch
from gatelock import _preempt
from gatelock._arbiter import (
RANK_DIET_GUARD,
RANK_SCREEN_LOCKER,
RANK_WAKE_ALARM,
Claim,
)
from gatelock._detect import _RandrEventSource
from gatelock._outputs import Output, OutputRect
from gatelock._window import LockConfig
from gatelock.tests.conftest import make_window
# Captured before the autouse hermetic fixture replaces it.
REAL_RANDR_START = _RandrEventSource.start
DARK_ONLY = (Output("DP-0", connected=True, rect=None, primary=True),)
TWO_LIVE = (
Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True),
Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)),
)
class TestPreemptWeakerHolder:
"""A weaker incumbent must not block a stronger claim indefinitely."""
def _arbiter_with(
self, *, our_rank: int, holder_rank: int, holder_pid: int
) -> MagicMock:
arbiter = MagicMock()
arbiter.claim = Claim(
app="ours",
rank=our_rank,
pid=1,
started="2026-08-21T12:00:00+00:00",
grab="global",
disable_vt=True,
instance_id="ours",
)
arbiter.describe_holder.return_value = Claim(
app="holder",
rank=holder_rank,
pid=holder_pid,
started="2026-08-21T07:04:07+00:00",
grab="global",
disable_vt=True,
instance_id="theirs",
)
return arbiter
def test_signals_a_weaker_holder(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_called_once_with(4075, signal.SIGTERM)
def test_never_signals_a_stronger_holder(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_DIET_GUARD, holder_rank=RANK_SCREEN_LOCKER, holder_pid=4137
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_signals_the_same_holder_only_once(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
window._log_grab_blocked(50)
kill.assert_called_once_with(4075, signal.SIGTERM)
def test_disabled_by_config(self, mock_root: MagicMock) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=False)
)
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_off_by_default(self, mock_root: MagicMock) -> None:
"""Preemption is opt-in, so no app can evict anyone by accident.
wake_alarm outranks screen_locker and builds a plain hard LockConfig.
While the default was True, that combination SIGTERMed the *armed
workout lock*, unlocking enforcement with no workout logged.
"""
arbiter = self._arbiter_with(
our_rank=RANK_WAKE_ALARM,
holder_rank=RANK_SCREEN_LOCKER,
holder_pid=4137,
)
window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
window._arbiter = arbiter
with patch("gatelock._preempt.os.kill") as kill:
window._log_grab_blocked(25)
kill.assert_not_called()
def test_no_arbiter_means_no_preemption(self, mock_root: MagicMock) -> None:
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
with patch("gatelock._preempt.os.kill") as kill:
_preempt.maybe_preempt(
Claim(
app="holder",
rank=RANK_DIET_GUARD,
pid=4075,
started="2026-08-21T07:04:07+00:00",
grab="global",
disable_vt=True,
instance_id="theirs",
),
arbiter=None,
config=window._config,
preempted_pids=set(),
)
kill.assert_not_called()
def test_a_holder_that_is_already_gone_is_reported_at_info(
self, mock_root: MagicMock
) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with (
patch("gatelock._preempt.os.kill", side_effect=ProcessLookupError),
patch("gatelock._preempt._logger") as logger,
):
window._log_grab_blocked(25)
assert logger.info.called
def test_an_unsignalable_holder_is_reported_not_raised(
self, mock_root: MagicMock
) -> None:
arbiter = self._arbiter_with(
our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
)
window, _hooks = make_window(
mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=True)
)
window._arbiter = arbiter
with (
patch("gatelock._preempt.os.kill", side_effect=OSError("nope")),
patch("gatelock._preempt._logger") as logger,
):
window._log_grab_blocked(25)
assert logger.exception.called