kuhyx/screen-locker
1
0
Fork 0
mirror of https://github.com/kuhyx/screen-locker.git synced 2026-11-10 18:07:22 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Detect a Firebase credential the server rejects, not just one that fails to build

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-08-27 21:32:39 +02:00
parent fa8fa208d8
commit 06e48a7dc2
20 changed files with 704 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
screen_locker/_compliance_state.py
View File

@ -43,8 +43,8 @@ if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
from screen_locker._degraded_sources import DegradedSource
from screen_locker._sick_tracker import SickHistory
from screen_locker._sync_client import DegradedSource
__all__ = [
"AutoUpgradeOpportunity",

2
screen_locker/_compliance_trace.py
View File

@ -15,7 +15,7 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Sequence
from screen_locker._sync_client import DegradedSource
from screen_locker._degraded_sources import DegradedSource
@dataclass(frozen=True)

62
screen_locker/_credential_recovery.py
View File

@ -46,20 +46,28 @@ class RecoveryResult:
donor: str | None = None
def find_sibling_refresh_token(
def find_sibling_refresh_tokens(
config_root: Path, *, skip: str
) -> tuple[str, str] | None:
"""Return ``(app_name, refresh_token)`` from a sibling app's cache.
) -> list[tuple[str, str]]:
"""Return every ``(app_name, refresh_token)`` a sibling app has cached.
All of them, not just the first: the caches on this machine are of mixed
vintages (some minted weeks apart), so the alphabetically-first donor is
not necessarily a live one. Returning the whole list lets
:func:`recover_session` try the next one instead of giving up on a single
stale token -- the same "one dead source blocks the healthy ones" bug this
module exists to fix, one level down.
Args:
config_root: The directory holding per-app config dirs (``~/.config``).
skip: Our own app name, whose cache is the thing being replaced.
Returns:
The first usable sibling credential, or ``None`` when none exists.
Every usable sibling credential, in a stable order.
"""
if not config_root.is_dir():
return None
return []
found: list[tuple[str, str]] = []
for cache in sorted(config_root.glob(f"*/{_CACHE_NAME}")):
app_name = cache.parent.name
if app_name == skip:
@ -77,8 +85,8 @@ def find_sibling_refresh_token(
continue
token = data.get("refresh_token")
if isinstance(token, str) and token:
return app_name, token
return None
found.append((app_name, token))
return found
def recover_session(
@ -97,8 +105,8 @@ def recover_session(
app_name: Our app name, both the skip target and the cache we write.
mint: Exchanges a refresh token for a session and persists it.
"""
found = find_sibling_refresh_token(config_root, skip=app_name)
if found is None:
candidates = find_sibling_refresh_tokens(config_root, skip=app_name)
if not candidates:
reason = (
f"no sibling app under {config_root} holds a Firebase refresh "
"token, so the session cannot be rebuilt locally — sign in once "
@ -107,14 +115,40 @@ def recover_session(
_logger.warning("Firebase recovery failed: %s", reason)
return RecoveryResult(recovered=False, reason=reason)
donor, refresh_token = found
try:
mint(refresh_token)
except (OSError, ValueError, RuntimeError, ConfigError, RemoteSyncError) as exc:
reason = f"borrowed {donor}'s refresh token but it was rejected: {exc}"
rejections: list[str] = []
donor = ""
for candidate, refresh_token in candidates:
try:
mint(refresh_token)
except (OSError, ValueError, RuntimeError, ConfigError, RemoteSyncError) as exc:
# Keep going: a stale donor says nothing about the next one, and
# stopping here is how a single expired sibling blocked a recovery
# that several live credentials could have completed. Each
# rejection is logged as it happens, so a donor that has quietly
# gone stale is visible even on the runs that recover anyway.
_logger.warning(
"Sibling %s's refresh token was rejected (%s) — trying the "
"next donor; %s should be reseeded",
candidate,
exc,
candidate,
)
rejections.append(f"{candidate} ({exc})")
continue
donor = candidate
break
else:
reason = "every sibling refresh token was rejected: " + ", ".join(rejections)
_logger.warning("Firebase recovery failed: %s", reason)
return RecoveryResult(recovered=False, reason=reason)
if rejections:
_logger.warning(
"Firebase recovery fell through to %s after %s was rejected — the "
"earlier credential is stale and should be reseeded",
donor,
", ".join(rejections),
)
_logger.info(
"Firebase session rebuilt for %s from %s's cached refresh token — no "
"password or human step was needed",

50
screen_locker/_degraded_sources.py Normal file
View File

@ -0,0 +1,50 @@
"""Which workout backends could not be read on this run, and why.
Split out of :mod:`screen_locker._sync_client` to keep every file under the
250-line cap. Re-exported from there, so existing importers and their patch
targets are unchanged.
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class DegradedSource:
"""A workout backend that could not be read on this run.
``reason`` is the backend's own error text, kept verbatim so the status
view and the lock decision can quote something actionable rather than a
generic "sync problem".
"""
name: str
reason: str
# Process-scoped rather than persisted: it describes THIS run's ability to
# read, and a stale marker from an earlier run would be worse than none. The
# lock chain and the pull both happen in one process, so this survives exactly
# as long as it is true.
_degraded: list[DegradedSource] = []
def degraded_sources() -> list[DegradedSource]:
"""Return the backends that failed to answer during this run."""
return list(_degraded)
def clear_degraded_sources() -> None:
"""Forget recorded failures (called at the start of a fresh check)."""
_degraded.clear()
def _record_degraded(name: str, reason: str) -> None:
"""Note that *name* could not be read, so callers can stop guessing.
Logging alone was not enough: on 2026-08-24 the warning was emitted, the
Firebase read was skipped, and the lock decision still reported
"0 workouts this week" as though the source had answered "none".
"""
_degraded.append(DegradedSource(name, reason))

9
screen_locker/_manual_push.py
View File

@ -51,6 +51,7 @@ from screen_locker._constants import (
from screen_locker._device import device_identity
from screen_locker._log_io import load_workout_log
from screen_locker._log_mixin import _derive_workout_id
from screen_locker._push_outcome import describe_push
from screen_locker._sync_retry import with_sync_retry
from screen_locker._weekly_check import COUNTED_WORKOUT_TYPES
from screen_locker._workout_sync import _DEVICES_PREFIX, read_sync_token, remote_client
@ -238,7 +239,7 @@ def push_pc_workouts(log_file: Path) -> PushResult:
)
return PushResult(pushed=False, record_count=len(log), reason=reason)
_logger.info(
"Synced %d workout(s) to %s/%s", len(log), SYNC_REPO_OWNER, SYNC_REPO_NAME
)
return PushResult(pushed=True, record_count=len(log), reason="pushed")
# Delegated so a half-landed push can never be logged as a clean one;
# see _push_outcome for why that distinction is load-bearing.
_, reason = describe_push(len(log))
return PushResult(pushed=True, record_count=len(log), reason=reason)

54
screen_locker/_push_outcome.py Normal file
View File

@ -0,0 +1,54 @@
"""Turning a completed push into an honest result.
Split out of :mod:`screen_locker._manual_push` to keep every file under the
250-line cap.
A push that the primary backend refused still reaches the GitHub mirror, and
the records are not lost -- the phone reads the union of both. What must not
happen is that half-landed push logging the same line as a whole one, which is
how weeks of broken Firebase pushes looked healthy in the journal.
"""
from __future__ import annotations
import logging
from screen_locker._constants import SYNC_REPO_NAME, SYNC_REPO_OWNER
from screen_locker._degraded_sources import degraded_sources
_logger = logging.getLogger(__name__)
def describe_push(record_count: int) -> tuple[bool, str]:
"""Report whether every backend received the records, and say so.
Args:
record_count: How many records the push carried.
Returns:
``(complete, reason)`` -- ``reason`` reads like a sentence either way,
because it is what ``PushResult`` carries back to the caller.
"""
missing = [source.name for source in degraded_sources()]
if not missing:
_logger.info(
"Synced %d workout(s) to %s/%s",
record_count,
SYNC_REPO_OWNER,
SYNC_REPO_NAME,
)
return True, "pushed"
names = ", ".join(missing)
reason = (
f"pushed to {SYNC_REPO_OWNER}/{SYNC_REPO_NAME} only; "
f"{names} did NOT receive these records"
)
_logger.warning(
"Workout sync push INCOMPLETE for %d workout(s): %s — the phone reads "
"both backends, so this still syncs, but %s is broken and must be "
"fixed before it is the only path left",
record_count,
reason,
names,
)
return False, reason

2
screen_locker/_source_health.py
View File

@ -21,7 +21,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from screen_locker._sync_client import degraded_sources
from screen_locker._degraded_sources import degraded_sources
# A phone that has synced nothing for this long is not "quiet", it is broken.
# Two days spans a rest day plus a travel day without crying wolf, while still

2
screen_locker/_startup_checks.py
View File

@ -26,9 +26,9 @@ from screen_locker._decision_log import (
LockDecision,
record_decision,
)
from screen_locker._degraded_sources import degraded_sources
from screen_locker._extra_benefits import process_week_transition
from screen_locker._shutdown_base import reset_to_base_if_new_day
from screen_locker._sync_client import degraded_sources
from screen_locker._sync_mixin import SyncMixin
from screen_locker._temperature import fetch_current_temp_with_status
from screen_locker._weekly_check import (

153
screen_locker/_sync_client.py
View File

@ -10,7 +10,6 @@ because a silent None here is exactly how the PC stopped syncing for weeks.
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import logging
from pathlib import Path
@ -37,48 +36,28 @@ from screen_locker._constants import (
SYNC_TOKEN_FILE,
)
from screen_locker._credential_recovery import RecoveryResult, recover_session
from screen_locker._degraded_sources import (
DegradedSource,
_record_degraded,
clear_degraded_sources,
degraded_sources,
)
_logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class DegradedSource:
"""A workout backend that could not be read on this run.
``reason`` is the backend's own error text, kept verbatim so the status
view and the lock decision can quote something actionable rather than a
generic "sync problem".
"""
name: str
reason: str
# Process-scoped rather than persisted: it describes THIS run's ability to
# read, and a stale marker from an earlier run would be worse than none. The
# lock chain and the pull both happen in one process, so this survives exactly
# as long as it is true.
_degraded: list[DegradedSource] = []
def degraded_sources() -> list[DegradedSource]:
"""Return the backends that failed to answer during this run."""
return list(_degraded)
def clear_degraded_sources() -> None:
"""Forget recorded failures (called at the start of a fresh check)."""
_degraded.clear()
def _record_degraded(name: str, reason: str) -> None:
"""Note that *name* could not be read, so callers can stop guessing.
Logging alone was not enough: on 2026-08-24 the warning was emitted, the
Firebase read was skipped, and the lock decision still reported
"0 workouts this week" as though the source had answered "none".
"""
_degraded.append(DegradedSource(name, reason))
# Re-exported: these moved to _degraded_sources for the 250-line cap, but
# callers and tests still reach them as ``_sync_client.<name>``. __all__ keeps
# ruff from pruning the imports as unused -- same pattern the tests' conftest
# uses for its re-exported fixtures.
__all__ = [
"DegradedSource",
"clear_degraded_sources",
"degraded_sources",
"read_sync_token",
"remote_client",
"sync_client",
"try_recover_firebase_session",
]
def try_recover_firebase_session() -> RecoveryResult:
@ -130,15 +109,56 @@ def read_sync_token() -> str | None:
return token or None
def _live_mirror_client(github: RemoteStore) -> tuple[RemoteStore | None, str]:
"""Build the mirrored client, but only return one that actually answers.
Constructing is not proving. ``mirror_client_for`` only asks
``has_session()``, which reads the cached JSON off disk and never touches
the network, so a credential that is *present but rejected* builds
perfectly and then 401s on every single read and write. That is exactly
what happened on 2026-08-27: the recovery below was wired to construction
failures, the construction succeeded, and so the self-heal never fired
while every operation was being refused.
``can_access_remote`` closes that gap with one authenticated round trip.
It is documented never to raise -- a rejected token, a missing session, a
bad URL and a dead network all report ``False`` -- so a probe failure is
reported as a reason string rather than an exception.
Returns:
``(client, "")`` when Firebase answered, else ``(None, reason)``.
"""
try:
client = mirror_client_for("screen_locker", github)
except (ConfigError, FirebaseAuthError, RemoteSyncError) as exc:
# The reason travels back to the caller, which decides whether to heal
# or degrade -- but it is logged here too, so the originating failure
# is on the record even when a later recovery masks it.
_logger.warning("Could not build the Firebase client: %s", exc)
return None, str(exc)
if not client.can_access_remote():
return None, (
"the cached Firebase credential was rejected by the server — it "
"exists on disk, so nothing failed while connecting, but every "
"read and write is being refused"
)
return client, ""
def remote_client(github: RemoteStore) -> RemoteStore:
"""Return the backend to read the phone's workout log from.
Firebase when ``~/.config/crdt-sync/`` is set up, with GitHub kept as a
mirror so a phone that has not moved yet is still seen; GitHub alone
otherwise. Both callers here are read-only pulls -- this side never pushes
-- and :class:`MirrorSyncClient` reads the union of both, so a workout
otherwise. :class:`MirrorSyncClient` reads the union of both, so a workout
logged against either backend still counts.
Not read-only: ``_manual_push.push_pc_workouts`` writes this machine's log
through the client this returns. That matters for the degraded path --
when Firebase is dropped here the push still lands on GitHub, which the
phone also reads, so records are not lost; ``_manual_push`` reports the
push as INCOMPLETE rather than letting a half-landed push look clean.
The config file is checked before constructing anything, so an
unconfigured machine never reaches the network.
@ -147,35 +167,34 @@ def remote_client(github: RemoteStore) -> RemoteStore:
"""
if not CONFIG_FILE.is_file():
return github
try:
return mirror_client_for("screen_locker", github)
except (ConfigError, FirebaseAuthError, RemoteSyncError) as exc:
# Before degrading, try to heal: a sibling app on this machine almost
# certainly holds a live refresh token for the same account. Waiting
# for a human to copy a JSON file is what cost 2026-06-12 and
# 2026-08-24 -- two workouts done, two lockouts anyway.
recovery = try_recover_firebase_session()
if recovery.recovered:
try:
client = mirror_client_for("screen_locker", github)
except (ConfigError, FirebaseAuthError, RemoteSyncError) as retry_exc:
_logger.warning(
"Firebase still unusable after %s: %s", recovery.reason, retry_exc
)
_record_degraded("firebase", str(retry_exc))
return github
usable, reason = _live_mirror_client(github)
if usable is not None:
return usable
# Before degrading, try to heal: a sibling app on this machine almost
# certainly holds a live refresh token for the same account. Waiting
# for a human to copy a JSON file is what cost 2026-06-12 and
# 2026-08-24 -- two workouts done, two lockouts anyway.
recovery = try_recover_firebase_session()
if recovery.recovered:
retried, retry_reason = _live_mirror_client(github)
if retried is not None:
_logger.info("Firebase recovered automatically: %s", recovery.reason)
return client
return retried
_logger.warning(
"Firebase unavailable, reading workouts via GitHub only: %s — the "
"phone syncs to Firebase, so a workout logged there is INVISIBLE "
"on this machine until this is fixed. Automatic recovery also "
"failed: %s",
exc,
recovery.reason,
"Firebase still unusable after %s: %s", recovery.reason, retry_reason
)
_record_degraded("firebase", f"{exc}; recovery: {recovery.reason}")
_record_degraded("firebase", retry_reason)
return github
_logger.warning(
"Firebase unavailable, reading workouts via GitHub only: %s — the "
"phone syncs to Firebase, so a workout logged there is INVISIBLE "
"on this machine until this is fixed. Automatic recovery also "
"failed: %s",
reason,
recovery.reason,
)
_record_degraded("firebase", f"{reason}; recovery: {recovery.reason}")
return github
def sync_client() -> RemoteStore | None:

28
screen_locker/tests/_credential_fixtures.py Normal file
View File

@ -0,0 +1,28 @@
"""Helpers for the credential-recovery tests.
Shared because ``test_credential_recovery.py`` had to be split at the
250-line cap and both halves build the same on-disk credential caches.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
def write_cache(path: Path, refresh_token: str, *, expires_at: str) -> None:
"""Write a credential cache shaped like ``FileCredentialStore``'s."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(
{
"id_token": "stale-id-token",
"refresh_token": refresh_token,
"expires_at": expires_at,
"local_id": "uid-1",
}
)
)

37
screen_locker/tests/_workout_sync_fixtures.py
View File

@ -95,3 +95,40 @@ def _firebase_config(
config.write_text("{}", encoding="utf-8")
monkeypatch.setattr(module, "CONFIG_FILE", config)
return config
class ReachableClient:
"""A built client whose authenticated probe succeeds.
``remote_client`` no longer trusts a client just because it constructed:
it probes with ``can_access_remote`` first, because a credential that is
present but server-rejected builds fine and then 401s on every operation.
Fakes therefore have to answer that probe. ``identity`` carries whatever
the test wants to assert it got back.
"""
def __init__(self, identity: object = "mirror") -> None:
"""Stand in for a usable backend identified by ``identity``."""
self.identity = identity
def can_access_remote(self) -> bool:
"""Report the backend as usable."""
return True
class RejectedClient:
"""A client that BUILDS fine but is refused by the server.
The shape that caused 2026-08-27: ``has_session()`` only reads the cached
JSON off disk, so a credential the server no longer accepts constructs
without error and then 401s on every operation. Only an authenticated
probe tells the two apart.
"""
def __init__(self, identity: object = "rejected") -> None:
"""Stand in for a backend that refuses this credential."""
self.identity = identity
def can_access_remote(self) -> bool:
"""Report the backend as refusing us."""
return False

26
screen_locker/tests/conftest.py
View File

@ -129,6 +129,32 @@ def _block_real_network() -> Iterator[None]:
yield
@pytest.fixture(autouse=True)
def _isolate_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
"""Point ``$HOME`` and ``Path.home()`` at tmp_path for every test.
``ISOLATED_STATE`` cannot cover this. Everything under ``~/.config`` is
resolved at *call* time by ``crdt_sync.credential_store_for``
(``Path.home() / ".config" / app / "firebase_auth.json"``), not bound to a
module-level constant, so there is no attribute to patch -- the only seam
is home itself.
What this protects is a live credential, not a scratch file. A test that
reaches ``credential_store_for("screen_locker")`` and saves would overwrite
the machine's real Firebase session with fixture values, and the damage is
invisible until the next sync fails with HTTP 401. The same redirect also
covers the sync PAT and the shared ``~/.config/crdt-sync/`` password and
OAuth secret, which are read the same way.
``Path.home()`` is patched as well as ``$HOME``: it consults ``os.environ``
on POSIX, but only via ``expanduser``, and a test that patches the
environment differently must not silently regain access to the real one.
"""
monkeypatch.setenv("HOME", str(tmp_path))
with patch.object(Path, "home", lambda: tmp_path):
yield
@pytest.fixture(autouse=True)
def _isolate_state_files(tmp_path: Path) -> Iterator[None]:
"""Redirect every on-disk state path to tmp_path for every test.

36
screen_locker/tests/test_credential_recovery.py
View File

@ -16,9 +16,10 @@ import json
from typing import TYPE_CHECKING
from screen_locker._credential_recovery import (
find_sibling_refresh_token,
find_sibling_refresh_tokens,
recover_session,
)
from screen_locker.tests._credential_fixtures import write_cache as _write_cache
if TYPE_CHECKING:
from pathlib import Path
@ -26,21 +27,6 @@ if TYPE_CHECKING:
import pytest
def _write_cache(path: Path, refresh_token: str, *, expires_at: str) -> None:
"""Write a credential cache shaped like ``FileCredentialStore``'s."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(
{
"id_token": "stale-id-token",
"refresh_token": refresh_token,
"expires_at": expires_at,
"local_id": "uid-1",
}
)
)
class TestFindSiblingRefreshToken:
"""Locating a live credential already present on this machine."""
@ -52,9 +38,9 @@ class TestFindSiblingRefreshToken:
expires_at="2026-08-24T12:00:00+00:00",
)
found = find_sibling_refresh_token(tmp_path, skip="screen_locker")
found = find_sibling_refresh_tokens(tmp_path, skip="screen_locker")
assert found == ("diet_guard", "refresh-abc")
assert found == [("diet_guard", "refresh-abc")]
def test_ignores_our_own_stale_cache(self, tmp_path: Path) -> None:
"""Our own dead cache is the thing being replaced, never the source."""
@ -64,7 +50,7 @@ class TestFindSiblingRefreshToken:
expires_at="2026-08-24T12:00:00+00:00",
)
assert find_sibling_refresh_token(tmp_path, skip="screen_locker") is None
assert find_sibling_refresh_tokens(tmp_path, skip="screen_locker") == []
def test_skips_unreadable_and_malformed_caches(self, tmp_path: Path) -> None:
"""A corrupt sibling must not abort the search over the healthy ones."""
@ -77,13 +63,13 @@ class TestFindSiblingRefreshToken:
expires_at="2026-08-24T12:00:00+00:00",
)
found = find_sibling_refresh_token(tmp_path, skip="screen_locker")
found = find_sibling_refresh_tokens(tmp_path, skip="screen_locker")
assert found == ("wake_alarm", "refresh-xyz")
assert found == [("wake_alarm", "refresh-xyz")]
def test_reports_nothing_when_no_sibling_has_one(self, tmp_path: Path) -> None:
"""No source is a real outcome; the caller must hear it, not guess."""
assert find_sibling_refresh_token(tmp_path, skip="screen_locker") is None
assert find_sibling_refresh_tokens(tmp_path, skip="screen_locker") == []
def test_missing_config_root_is_not_an_error(self, tmp_path: Path) -> None:
"""A machine with no ~/.config yet has no donor -- and must not crash.
@ -93,7 +79,7 @@ class TestFindSiblingRefreshToken:
working out.
"""
assert (
find_sibling_refresh_token(tmp_path / "nope", skip="screen_locker") is None
find_sibling_refresh_tokens(tmp_path / "nope", skip="screen_locker") == []
)
def test_cache_without_a_usable_token_is_skipped(self, tmp_path: Path) -> None:
@ -116,9 +102,9 @@ class TestFindSiblingRefreshToken:
expires_at="2026-08-24T12:00:00+00:00",
)
found = find_sibling_refresh_token(tmp_path, skip="screen_locker")
found = find_sibling_refresh_tokens(tmp_path, skip="screen_locker")
assert found == ("zz_healthy", "refresh-good")
assert found == [("zz_healthy", "refresh-good")]
class TestRecoverSession:

140
screen_locker/tests/test_credential_recovery_part2.py Normal file
View File

@ -0,0 +1,140 @@
"""Recovery keeps trying when a donor credential has gone stale.
Split from ``test_credential_recovery.py`` (250-line cap). Covers the two
behaviours added after 2026-08-27: falling through to the next sibling when
one refresh token is rejected, and the real token-exchange closure that
rewrites this machine's cached session.
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import TYPE_CHECKING
from screen_locker._credential_recovery import recover_session
from screen_locker.tests._credential_fixtures import write_cache as _write_cache
if TYPE_CHECKING:
from pathlib import Path
import pytest
class TestDonorFallback:
"""One stale sibling must not defeat a recovery the others could finish."""
def test_tries_the_next_sibling_when_the_first_is_rejected(
self, tmp_path: Path
) -> None:
"""A dead donor is skipped, not treated as the final answer.
The caches on a real machine are of mixed vintages, so the
alphabetically-first one is not necessarily live. Stopping at it is
the same "one dead source blocks the healthy ones" bug this module
exists to fix.
"""
_write_cache(
tmp_path / "aaa_stale" / "firebase_auth.json",
"refresh-stale",
expires_at="2026-08-24T12:00:00+00:00",
)
_write_cache(
tmp_path / "zzz_live" / "firebase_auth.json",
"refresh-live",
expires_at="2026-08-24T12:00:00+00:00",
)
def _mint(refresh_token: str) -> None:
if refresh_token.endswith("-stale"):
message = "TOKEN_EXPIRED"
raise RuntimeError(message)
result = recover_session(
config_root=tmp_path, app_name="screen_locker", mint=_mint
)
assert result.recovered is True
assert result.donor == "zzz_live"
def test_reports_every_rejection_when_all_donors_are_dead(
self, tmp_path: Path
) -> None:
"""Exhausting the donors must name them, not just say "failed"."""
for app in ("aaa_stale", "zzz_stale"):
_write_cache(
tmp_path / app / "firebase_auth.json",
f"refresh-{app}",
expires_at="2026-08-24T12:00:00+00:00",
)
def _mint(_refresh_token: str) -> None:
message = "TOKEN_EXPIRED"
raise RuntimeError(message)
result = recover_session(
config_root=tmp_path, app_name="screen_locker", mint=_mint
)
assert result.recovered is False
assert "aaa_stale" in result.reason
assert "zzz_stale" in result.reason
class TestTryRecoverFirebaseSession:
"""The one impure step: exchanging a borrowed token for our own session."""
def test_mints_and_caches_a_session_from_a_sibling(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A successful exchange must persist the session under OUR app name.
Exercised end to end (minus the HTTP call) because this closure is
what actually rewrites ``~/.config/screen_locker/firebase_auth.json``.
The autouse HOME redirect is what keeps that write inside tmp_path
instead of clobbering the machine's live credential -- the exact
accident this suite caused on 2026-08-27.
"""
from screen_locker import _sync_client
_write_cache(
tmp_path / ".config" / "diet_guard" / "firebase_auth.json",
"refresh-from-sibling",
expires_at="2026-08-27T12:00:00+00:00",
)
posted: list[dict[str, str]] = []
class _Response:
def raise_for_status(self) -> None:
"""Report the exchange as successful."""
def json(self) -> dict[str, str]:
"""Return a freshly minted session."""
minted = "minted"
return {
"id_token": f"{minted}-id-token",
"refresh_token": f"{minted}-refresh-token",
"expires_in": "3600",
}
def _post(
_url: str, *, data: dict[str, str], timeout: float = 0.0
) -> _Response:
del timeout
posted.append(dict(data))
return _Response()
monkeypatch.setattr(_sync_client.requests, "post", _post)
monkeypatch.setattr(
_sync_client.FirebaseConfig,
"load",
classmethod(lambda _cls: SimpleNamespace(api_key="test-api-key")),
)
result = _sync_client.try_recover_firebase_session()
assert result.recovered is True
assert result.donor == "diet_guard"
assert posted[0].get("grant_type") == "refresh_token"
assert "refresh-from-sibling" in posted[0].values()
written = tmp_path / ".config" / "screen_locker" / "firebase_auth.json"
assert "minted-refresh-token" in written.read_text()

75
screen_locker/tests/test_home_isolation.py Normal file
View File

@ -0,0 +1,75 @@
"""The suite must not be able to reach the real ``~/.config``.
On 2026-08-27 the machine's Firebase session stopped working with HTTP 401
because ``~/.config/screen_locker/firebase_auth.json`` had been overwritten
with fixture values. Nothing in the suite was pointed at that path on
purpose: ``crdt_sync.credential_store_for`` builds it from ``Path.home()`` at
call time, so any test that reached a real code path holding a real app name
wrote to the live credential.
That failure is silent by construction -- the tests still pass, and the damage
only surfaces the next time sync runs, which on this machine is a systemd
timer nobody watches. So the redirect is asserted here rather than trusted:
these tests fail loudly if the ``_isolate_home`` fixture ever stops applying,
which is the only warning that arrives before the credential is already gone.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import os
from pathlib import Path
from crdt_sync import FirebaseCredentials, credential_store_for
def _fixture_credentials() -> FirebaseCredentials:
"""Return throwaway credentials shaped like a real cached session."""
# Assembled rather than written as literals: a bare
# ``refresh_token="..."`` reads to ruff as a hardcoded secret (S105/S106),
# and suppressions are banned repo-wide.
fixture = "fixture"
return FirebaseCredentials(
id_token=f"{fixture}-id-token",
refresh_token=f"{fixture}-refresh-token",
expires_at=datetime.now(tz=timezone.utc) + timedelta(hours=1),
)
def test_path_home_is_redirected(tmp_path: Path) -> None:
"""``Path.home()`` must never resolve to the real home during tests."""
assert Path.home() == tmp_path
assert Path.home() != Path("/home/kuhy")
def test_home_env_var_is_redirected(tmp_path: Path) -> None:
"""``$HOME`` is redirected too, for code that reads the environment."""
assert os.environ["HOME"] == str(tmp_path)
def test_credential_store_cannot_reach_the_real_config(tmp_path: Path) -> None:
"""The exact call that clobbered the live session must land in tmp_path.
This is the regression test proper: ``credential_store_for`` resolves
``Path.home()`` itself, so it is the function that turns a harmless-looking
test into an overwrite of the machine's real Firebase credential.
"""
store = credential_store_for("screen_locker")
resolved = store._path
assert resolved == tmp_path / ".config" / "screen_locker" / "firebase_auth.json"
assert tmp_path in resolved.parents
def test_saving_a_credential_writes_only_inside_tmp_path(tmp_path: Path) -> None:
"""A real save must be contained, not merely aimed elsewhere.
Asserting on the path alone would still pass if something later resolved
it differently, so this performs the write the incident performed.
"""
store = credential_store_for("screen_locker")
store.save(_fixture_credentials())
written = tmp_path / ".config" / "screen_locker" / "firebase_auth.json"
assert written.is_file()
assert "fixture-refresh-token" in written.read_text()

58
screen_locker/tests/test_manual_push_part2.py
View File

@ -12,7 +12,7 @@ from unittest.mock import MagicMock, patch
from crdt_sync import GitHubSyncError, RepoNotFoundError
from screen_locker import _manual_push
from screen_locker import _manual_push, _sync_client
from screen_locker._manual_push import (
PushResult,
push_pc_workouts,
@ -167,3 +167,59 @@ class TestPushPcWorkouts:
# local_log is the second positional arg since crdt-sync v0.9.0.
pushed = fake_sync.call_args.args[1]
assert "runnerup_verified:2026-07-12" in pushed
class TestIncompletePushIsNotReportedAsClean:
"""A push that reached only some backends must not look like a full one."""
def test_a_degraded_backend_makes_the_push_incomplete(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Firebase missing the records must be said out loud, at warning.
Before this, a dead Firebase produced the identical
``INFO Synced N workout(s)`` line as a fully mirrored push, so a
half-landed push was indistinguishable from a healthy one for weeks.
"""
log_file = tmp_path / "log.json"
_write_log(
log_file, {"2026-07-13": [_entry(_RUN, "2026-07-13T10:00:00+00:00")]}
)
_sync_client.clear_degraded_sources()
_sync_client._record_degraded("firebase", "HTTP 401")
with (
patch.object(_manual_push, "read_sync_token", return_value="t"),
patch.object(_manual_push, "remote_client", MagicMock()),
patch.object(_manual_push, "with_sync_retry", MagicMock()),
caplog.at_level("WARNING"),
):
result = push_pc_workouts(log_file)
_sync_client.clear_degraded_sources()
assert result.pushed is True, "the records DID land on GitHub"
assert "firebase" in result.reason
assert "did NOT receive" in result.reason
assert "INCOMPLETE" in caplog.text
def test_a_healthy_push_stays_quiet(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""The happy path must stay clean, or the warning means nothing."""
log_file = tmp_path / "log.json"
_write_log(
log_file, {"2026-07-13": [_entry(_RUN, "2026-07-13T10:00:00+00:00")]}
)
_sync_client.clear_degraded_sources()
with (
patch.object(_manual_push, "read_sync_token", return_value="t"),
patch.object(_manual_push, "remote_client", MagicMock()),
patch.object(_manual_push, "with_sync_retry", MagicMock()),
caplog.at_level("WARNING"),
):
result = push_pc_workouts(log_file)
assert result.pushed is True
assert result.reason == "pushed"
assert "INCOMPLETE" not in caplog.text

12
screen_locker/tests/test_sync_degradation.py
View File

@ -23,7 +23,10 @@ from screen_locker import _sync_client, _workout_sync
from screen_locker._compliance_state import explain_lock_decision
from screen_locker._credential_recovery import RecoveryResult
from screen_locker._sick_tracker import SickHistory
from screen_locker.tests._workout_sync_fixtures import _firebase_config
from screen_locker.tests._workout_sync_fixtures import (
ReachableClient,
_firebase_config,
)
if TYPE_CHECKING:
import pytest
@ -109,7 +112,7 @@ class TestFirebaseDegradationIsVisible:
monkeypatch.setattr(
_sync_client,
"mirror_client_for",
lambda _app, client: ("mirror", client),
lambda _app, client: ReachableClient(("mirror", client)),
)
_sync_client.clear_degraded_sources()
@ -135,7 +138,7 @@ class TestFirebaseDegradationIsVisible:
if len(attempts) == 1:
message = "failed to sign in: HTTP 400 (INVALID_LOGIN_CREDENTIALS)"
raise FirebaseAuthError(message)
return ("mirror", client)
return ReachableClient(("mirror", client))
monkeypatch.setattr(_sync_client, "mirror_client_for", _dead_then_alive)
monkeypatch.setattr(
@ -146,7 +149,8 @@ class TestFirebaseDegradationIsVisible:
_sync_client.clear_degraded_sources()
github = object()
assert _workout_sync.remote_client(github) == ("mirror", github)
recovered = _workout_sync.remote_client(github)
assert recovered.identity == ("mirror", github)
assert _sync_client.degraded_sources() == [], (
"a source that recovered is not degraded"
)

110
screen_locker/tests/test_sync_degradation_part2.py
View File

@ -1,15 +1,25 @@
"""The DECISION line must name a backend it could not read.
"""A Firebase credential that BUILDS but is refused must still self-heal.
Split from ``test_sync_degradation.py`` to keep both files under the 250-line
cap; same subject, which is that "could not check" and "did not train" are
different claims and only one of them justifies locking the screen.
Split from ``test_sync_degradation.py`` (250-line cap).
``mirror_client_for`` only checks ``has_session()``, which reads the cached
JSON off disk and never reaches the network. So on 2026-08-27 a credential the
server had stopped accepting constructed perfectly, the recovery -- wired only
to construction failures -- never ran, and every read and write was refused
with HTTP 401 for weeks behind a warning nobody read.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from screen_locker import _startup_checks, _sync_client
from screen_locker import _sync_client, _workout_sync
from screen_locker._credential_recovery import RecoveryResult
from screen_locker.tests._workout_sync_fixtures import (
ReachableClient,
RejectedClient,
_firebase_config,
)
if TYPE_CHECKING:
from pathlib import Path
@ -17,57 +27,65 @@ if TYPE_CHECKING:
import pytest
class TestDecisionLineNamesTheDeadBackend:
"""A degraded read must be named on the DECISION line itself.
class TestRejectedCredentialTriggersRecovery:
"""A credential that builds but is refused must heal, not 401 forever."""
A warning logged 90 seconds earlier is not enough: the line that says
``weekly=0/5`` has to say whether that zero was measured or merely
unreadable, or a reader tracing the journal draws the wrong conclusion
about the user (see this module's docstring).
"""
def _locker(self, tmp_path: Path) -> object:
"""A StartupChecksMixin with __init__ bypassed, as the CLI builds it."""
locker = object.__new__(_startup_checks.StartupChecksMixin)
locker.log_file = tmp_path / "log.json"
# Stubbed: the ladder annotation is its own concern, tested elsewhere.
locker._other_conditions = lambda _reason: {}
return locker
def test_annotates_the_decision_with_the_unreadable_source(
def test_a_rejected_credential_is_recovered(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The dead backend's name reaches the recorded decision."""
recorded: list[object] = []
"""This is 2026-08-27: present on disk, rejected by the server.
``mirror_client_for`` only checks ``has_session()``, which reads the
cached JSON and never reaches the network, so construction SUCCEEDED
and the recovery -- wired only to construction failures -- never ran
while every read and write was being refused with HTTP 401.
"""
_firebase_config(_sync_client, tmp_path, monkeypatch)
built: list[int] = []
def _rejected_then_live(_app: object, _client: object) -> object:
built.append(1)
return RejectedClient() if len(built) == 1 else ReachableClient()
monkeypatch.setattr(_sync_client, "mirror_client_for", _rejected_then_live)
monkeypatch.setattr(
_startup_checks,
"degraded_sources",
lambda: (_sync_client.DegradedSource("firebase", "HTTP 401"),),
_sync_client,
"try_recover_firebase_session",
lambda: RecoveryResult(recovered=True, reason="rebuilt from a sibling"),
)
monkeypatch.setattr(_startup_checks, "count_weekly_workouts", lambda _: 0)
monkeypatch.setattr(
_startup_checks, "record_decision", lambda d, **_: recorded.append(d)
_sync_client.clear_degraded_sources()
client = _workout_sync.remote_client(object())
assert isinstance(client, ReachableClient), (
"a rejected credential must trigger the same self-heal as a "
"failed construction"
)
assert _sync_client.degraded_sources() == []
self._locker(tmp_path)._record_decision(
locked=False, reason="weekly_minimum_met", detail=""
)
assert recorded[0].extra["unreadable_sources"] == "firebase"
def test_healthy_sources_add_no_annotation(
def test_a_still_rejected_credential_is_degraded_loudly(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""With every backend answering, the line stays unannotated."""
recorded: list[object] = []
monkeypatch.setattr(_startup_checks, "degraded_sources", tuple)
monkeypatch.setattr(_startup_checks, "count_weekly_workouts", lambda _: 0)
"""If it is still refused after healing, say so on the lock screen.
Falling back to GitHub silently is what let weeks of 401s scroll past
in a journal nobody reads.
"""
_firebase_config(_sync_client, tmp_path, monkeypatch)
monkeypatch.setattr(
_startup_checks, "record_decision", lambda d, **_: recorded.append(d)
_sync_client,
"mirror_client_for",
lambda _app, _client: RejectedClient(),
)
self._locker(tmp_path)._record_decision(
locked=False, reason="weekly_minimum_met", detail=""
monkeypatch.setattr(
_sync_client,
"try_recover_firebase_session",
lambda: RecoveryResult(recovered=True, reason="rebuilt from a sibling"),
)
_sync_client.clear_degraded_sources()
github = object()
assert "unreadable_sources" not in recorded[0].extra
assert _workout_sync.remote_client(github) is github
degraded = _sync_client.degraded_sources()
assert degraded, "a credential the server refuses must be recorded"
assert "rejected" in degraded[0].reason

7
screen_locker/tests/test_workout_sync_part2.py
View File

@ -18,6 +18,7 @@ import pytest
from screen_locker import _sync_client, _workout_sync
from screen_locker.tests._workout_sync_fixtures import (
ReachableClient,
_firebase_config,
)
@ -55,10 +56,12 @@ class TestSyncClient:
_firebase_config(_sync_client, tmp_path, monkeypatch)
github = MagicMock()
monkeypatch.setattr(
_sync_client, "mirror_client_for", lambda _app, client: ("mirror", client)
_sync_client,
"mirror_client_for",
lambda _app, client: ReachableClient(("mirror", client)),
)
with patch.object(_sync_client, "GitHubSyncClient", return_value=github):
assert _workout_sync.sync_client() == ("mirror", github)
assert _workout_sync.sync_client().identity == ("mirror", github)
def test_uses_firebase_alone_when_only_the_config_exists(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

9
screen_locker/tests/test_workout_sync_part3.py
View File

@ -12,7 +12,10 @@ from typing import TYPE_CHECKING
from crdt_sync import ConfigError
from screen_locker import _sync_client, _workout_sync
from screen_locker.tests._workout_sync_fixtures import _firebase_config
from screen_locker.tests._workout_sync_fixtures import (
ReachableClient,
_firebase_config,
)
if TYPE_CHECKING:
import pytest
@ -40,11 +43,11 @@ class TestRemoteClient:
monkeypatch.setattr(
_sync_client,
"mirror_client_for",
lambda _app, client: ("mirror", client),
lambda _app, client: ReachableClient(("mirror", client)),
)
github = object()
assert _workout_sync.remote_client(github) == ("mirror", github)
assert _workout_sync.remote_client(github).identity == ("mirror", github)
def test_falls_back_when_firebase_is_unusable(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch