diff --git a/gatelock/gatelock/tests/test_arbiter.py b/gatelock/gatelock/tests/test_arbiter.py index 719a4a0..0ece730 100644 --- a/gatelock/gatelock/tests/test_arbiter.py +++ b/gatelock/gatelock/tests/test_arbiter.py @@ -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 diff --git a/gatelock/gatelock/tests/test_arbiter_lifecycle.py b/gatelock/gatelock/tests/test_arbiter_lifecycle.py new file mode 100644 index 0000000..ee3a8e6 --- /dev/null +++ b/gatelock/gatelock/tests/test_arbiter_lifecycle.py @@ -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() diff --git a/gatelock/gatelock/tests/test_arbiter_release.py b/gatelock/gatelock/tests/test_arbiter_release.py new file mode 100644 index 0000000..5bc2342 --- /dev/null +++ b/gatelock/gatelock/tests/test_arbiter_release.py @@ -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 diff --git a/gatelock/gatelock/tests/test_escape.py b/gatelock/gatelock/tests/test_escape.py index 5503ab5..76bcccb 100644 --- a/gatelock/gatelock/tests/test_escape.py +++ b/gatelock/gatelock/tests/test_escape.py @@ -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 diff --git a/gatelock/gatelock/tests/test_escape_budget.py b/gatelock/gatelock/tests/test_escape_budget.py new file mode 100644 index 0000000..e28c944 --- /dev/null +++ b/gatelock/gatelock/tests/test_escape_budget.py @@ -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 diff --git a/gatelock/gatelock/tests/test_outputs.py b/gatelock/gatelock/tests/test_outputs.py index 63accd3..45aba58 100644 --- a/gatelock/gatelock/tests/test_outputs.py +++ b/gatelock/gatelock/tests/test_outputs.py @@ -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"] diff --git a/gatelock/gatelock/tests/test_outputs_enumerator.py b/gatelock/gatelock/tests/test_outputs_enumerator.py new file mode 100644 index 0000000..0f24b62 --- /dev/null +++ b/gatelock/gatelock/tests/test_outputs_enumerator.py @@ -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"] diff --git a/gatelock/gatelock/tests/test_outputs_randr.py b/gatelock/gatelock/tests/test_outputs_randr.py new file mode 100644 index 0000000..6fc8fe6 --- /dev/null +++ b/gatelock/gatelock/tests/test_outputs_randr.py @@ -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) diff --git a/gatelock/gatelock/tests/test_scrollable.py b/gatelock/gatelock/tests/test_scrollable.py index 5002b82..d2f6e5c 100644 --- a/gatelock/gatelock/tests/test_scrollable.py +++ b/gatelock/gatelock/tests/test_scrollable.py @@ -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("", 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 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("") - 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("") - surface.canvas.update() - scrolled = surface.canvas.yview()[0] - - child.event_generate("") - surface.canvas.update() - - assert surface.canvas.yview()[0] < scrolled diff --git a/gatelock/gatelock/tests/test_scrollable_into_view.py b/gatelock/gatelock/tests/test_scrollable_into_view.py new file mode 100644 index 0000000..4efa64d --- /dev/null +++ b/gatelock/gatelock/tests/test_scrollable_into_view.py @@ -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("", 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 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("") + 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("") + surface.canvas.update() + scrolled = surface.canvas.yview()[0] + + child.event_generate("") + surface.canvas.update() + + assert surface.canvas.yview()[0] < scrolled diff --git a/gatelock/gatelock/tests/test_surfaces.py b/gatelock/gatelock/tests/test_surfaces.py index 3d7f08f..5a5ea57 100644 --- a/gatelock/gatelock/tests/test_surfaces.py +++ b/gatelock/gatelock/tests/test_surfaces.py @@ -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 diff --git a/gatelock/gatelock/tests/test_surfaces_focus.py b/gatelock/gatelock/tests/test_surfaces_focus.py new file mode 100644 index 0000000..2aebb1d --- /dev/null +++ b/gatelock/gatelock/tests/test_surfaces_focus.py @@ -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 diff --git a/gatelock/gatelock/tests/test_window.py b/gatelock/gatelock/tests/test_window.py index 6331615..86795cd 100644 --- a/gatelock/gatelock/tests/test_window.py +++ b/gatelock/gatelock/tests/test_window.py @@ -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 diff --git a/gatelock/gatelock/tests/test_window_grab.py b/gatelock/gatelock/tests/test_window_grab.py new file mode 100644 index 0000000..1c0541f --- /dev/null +++ b/gatelock/gatelock/tests/test_window_grab.py @@ -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 diff --git a/gatelock/gatelock/tests/test_window_surfaces.py b/gatelock/gatelock/tests/test_window_surfaces.py index 784e0b4..e34e41e 100644 --- a/gatelock/gatelock/tests/test_window_surfaces.py +++ b/gatelock/gatelock/tests/test_window_surfaces.py @@ -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() diff --git a/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py b/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py new file mode 100644 index 0000000..3d8bd78 --- /dev/null +++ b/gatelock/gatelock/tests/test_window_surfaces_lifecycle.py @@ -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() diff --git a/gatelock/gatelock/tests/test_window_surfaces_preempt.py b/gatelock/gatelock/tests/test_window_surfaces_preempt.py new file mode 100644 index 0000000..edd22d2 --- /dev/null +++ b/gatelock/gatelock/tests/test_window_surfaces_preempt.py @@ -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