From 0411116f24b9ccec6a88942b60bc763ea18af328 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Tue, 10 Nov 2026 16:08:40 +0100 Subject: [PATCH] Harden RunnerUp retirement after review Silent-failure checker, mypy, pylint now clean. Hidden packages count as installed (pm list -u), pull gets a long timeout, wifi entries are matched by ro.serialno, and a missing log/drop dir is a warning. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WzxdWZ9CKZsNxtD5Jr9bYD --- screen_locker/_runnerup_retire.py | 90 +++++--------- screen_locker/_runnerup_retire_adb.py | 62 ++++++++-- screen_locker/_runnerup_retire_result.py | 62 ++++++++++ screen_locker/_runnerup_retire_state.py | 10 +- screen_locker/_runnerup_retire_tcx.py | 26 ++-- screen_locker/runnerup_retire.py | 10 +- screen_locker/tests/_retire_helpers.py | 9 +- screen_locker/tests/test_runnerup_retire.py | 51 ++++++-- .../tests/test_runnerup_retire_adb.py | 111 +++++++++++------ .../tests/test_runnerup_retire_cli.py | 28 +++-- .../tests/test_runnerup_retire_marker.py | 117 ++++++++++++++++++ .../tests/test_runnerup_retire_state.py | 52 ++++++++ .../tests/test_runnerup_retire_tcx.py | 113 +++-------------- 13 files changed, 504 insertions(+), 237 deletions(-) create mode 100644 screen_locker/_runnerup_retire_result.py create mode 100644 screen_locker/tests/test_runnerup_retire_marker.py create mode 100644 screen_locker/tests/test_runnerup_retire_state.py diff --git a/screen_locker/_runnerup_retire.py b/screen_locker/_runnerup_retire.py index d139c2b..9cf2159 100644 --- a/screen_locker/_runnerup_retire.py +++ b/screen_locker/_runnerup_retire.py @@ -16,17 +16,17 @@ from __future__ import annotations from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta -from enum import StrEnum import logging from pathlib import Path from typing import TYPE_CHECKING from screen_locker._runnerup_retire_adb import PACKAGE, AdbError +from screen_locker._runnerup_retire_result import LEVEL, Outcome, RetireResult from screen_locker._runnerup_retire_state import load_state, save_state -from screen_locker._runnerup_retire_tcx import check_marker, find_anchor +from screen_locker._runnerup_retire_tcx import app_run_ends, check_marker, find_anchor if TYPE_CHECKING: - from screen_locker._runnerup_retire_adb import Phone + from screen_locker._runnerup_retire_adb import PhoneLike from screen_locker._runnerup_retire_state import RetireState _logger = logging.getLogger(__name__) @@ -38,57 +38,6 @@ LOG_FILE = Path(__file__).resolve().parent / "log.json" MARKER_NAME = ".workout_app_import.json" -class Outcome(StrEnum): - """How a run ended; the value is what the journal shows.""" - - UNINSTALLED = "uninstalled" - ALREADY_GONE = "already-gone" - ALREADY_DONE = "already-done" - NO_ANCHOR = "no-anchor" - NOT_DUE = "not-due" - DRY_RUN = "dry-run" - DEVICE_ABSENT = "device-absent" - MARKER_INVALID = "marker-invalid" - ADB_MISSING = "adb-missing" - PHONE_ERROR = "phone-error" - BACKUP_FAILED = "backup-failed" - UNINSTALL_FAILED = "uninstall-failed" - - -# Level per outcome; the failures also make the unit exit non-zero. -_LEVEL = { - Outcome.DEVICE_ABSENT: logging.WARNING, - Outcome.MARKER_INVALID: logging.WARNING, - Outcome.ADB_MISSING: logging.ERROR, - Outcome.PHONE_ERROR: logging.ERROR, - Outcome.BACKUP_FAILED: logging.ERROR, - Outcome.UNINSTALL_FAILED: logging.ERROR, -} -_NONZERO = frozenset( - { - Outcome.MARKER_INVALID, - Outcome.ADB_MISSING, - Outcome.PHONE_ERROR, - Outcome.BACKUP_FAILED, - Outcome.UNINSTALL_FAILED, - } -) - - -@dataclass(frozen=True) -class RetireResult: - """What happened and why, in a sentence a human can act on.""" - - outcome: Outcome - reason: str - logged: bool = False # the failure was already logged where it was caught - - @property - def ok(self) -> bool: - """False for outcomes the unit should surface as failed.""" - return self.outcome not in _NONZERO - - @dataclass(frozen=True) class Paths: """Every on-disk location, so tests (and ``--status``) can redirect them.""" @@ -128,7 +77,7 @@ def _finish(paths: Paths, state: RetireState, now: datetime, why: str) -> None: ) -def _probe(phone: Phone) -> RetireResult | tuple[str, bool]: +def _probe(phone: PhoneLike) -> RetireResult | tuple[str, bool]: """``(serial, installed)`` if the phone answers, else the result explaining why.""" if not phone.available(): return RetireResult( @@ -149,7 +98,7 @@ def _probe(phone: Phone) -> RetireResult | tuple[str, bool]: def _backup( - paths: Paths, phone: Phone, state: RetireState, serial: str, now: datetime + paths: Paths, phone: PhoneLike, state: RetireState, serial: str, now: datetime ) -> RetireState | RetireResult: """Back the phone's exports up once; a later retry reuses a finished backup.""" if state.backup_path and Path(state.backup_path).is_dir(): @@ -167,7 +116,7 @@ def _backup( def _retire( - paths: Paths, phone: Phone, state: RetireState, now: datetime, *, dry_run: bool + paths: Paths, phone: PhoneLike, state: RetireState, now: datetime, *, dry_run: bool ) -> RetireResult: """Everything after "it is due and the marker is good": the phone steps.""" probed = _probe(phone) @@ -199,6 +148,21 @@ def _retire( ) +def _no_anchor(paths: Paths) -> RetireResult: + """Why there is no anchor; a missing source is a warning, not "nothing yet".""" + absent = [str(p) for p in (paths.watch_dir, paths.log_file) if not p.exists()] + if absent: + return RetireResult( + Outcome.SOURCES_MISSING, + f"cannot look for app runs, missing: {', '.join(absent)}", + ) + seen = len(app_run_ends(paths.anchor_dirs())) + return RetireResult( + Outcome.NO_ANCHOR, + f"no app-recorded run credited yet ({seen} app TCX on disk); nothing to retire", + ) + + def _gate( paths: Paths, state: RetireState, now: datetime, *, dry_run: bool ) -> RetireState | RetireResult: @@ -209,9 +173,7 @@ def _gate( ) anchor = derive_anchor(paths, state) if anchor is None: - return RetireResult( - Outcome.NO_ANCHOR, "no app-recorded run credited yet; nothing to retire" - ) + return _no_anchor(paths) if not state.anchor_at: state = replace(state, anchor_at=anchor.isoformat()) if not dry_run: @@ -230,7 +192,11 @@ def _gate( def run_retire( - paths: Paths, phone: Phone, now: datetime | None = None, *, dry_run: bool = False + paths: Paths, + phone: PhoneLike, + now: datetime | None = None, + *, + dry_run: bool = False, ) -> RetireResult: """One pass: uninstall RunnerUp if (and only if) every condition holds.""" when = now or datetime.now(UTC) @@ -241,7 +207,7 @@ def run_retire( result = _retire(paths, phone, gated, when, dry_run=dry_run) if not result.logged: _logger.log( - _LEVEL.get(result.outcome, logging.INFO), + LEVEL.get(result.outcome, logging.INFO), "RunnerUp retirement: %s: %s", result.outcome.value, result.reason, diff --git a/screen_locker/_runnerup_retire_adb.py b/screen_locker/_runnerup_retire_adb.py index 54d25a7..2157a1c 100644 --- a/screen_locker/_runnerup_retire_adb.py +++ b/screen_locker/_runnerup_retire_adb.py @@ -13,7 +13,7 @@ import logging from pathlib import Path import shutil import subprocess -from typing import TYPE_CHECKING, Any, NoReturn +from typing import TYPE_CHECKING, Any, NoReturn, Protocol if TYPE_CHECKING: from collections.abc import Callable @@ -24,6 +24,9 @@ PHONE_SERIAL = "23181JEGR08034" PACKAGE = "org.runnerup.free" REMOTE_DIR = "/sdcard/Documents/RunnerUp" _ADB_TIMEOUT_S = 50 +# The whole export history (TCX + GPX) over wifi adb; the oneshot unit has no +# start timeout, so a slow pull is allowed to finish. +_PULL_TIMEOUT_S = 900 class AdbError(RuntimeError): @@ -42,6 +45,25 @@ class Device: reason: str +class PhoneLike(Protocol): + """What the retirement needs from a phone (tests substitute a fake).""" + + def available(self) -> bool: + """Whether adb can be run at all.""" + + def find_device(self) -> Device: + """The attached phone, or why there is none.""" + + def installed(self, dev: str, *, hidden: bool = True) -> bool: + """Whether RunnerUp is on the phone.""" + + def backup(self, dev: str, final: Path) -> tuple[Path, int]: + """Pull the exports to ``final``.""" + + def uninstall(self, dev: str) -> None: + """Uninstall RunnerUp or raise ``AdbError``.""" + + class Phone: """adb calls against one phone, with the process runner injectable.""" @@ -61,11 +83,11 @@ class Phone: """Whether an ``adb`` binary is on PATH.""" return self._which("adb") is not None - def _adb(self, *args: str) -> str: + def _adb(self, *args: str, timeout: int = _ADB_TIMEOUT_S) -> str: cmd = ["adb", *args] try: done = self._run( - cmd, capture_output=True, text=True, timeout=_ADB_TIMEOUT_S, check=False + cmd, capture_output=True, text=True, timeout=timeout, check=False ) except (OSError, subprocess.TimeoutExpired) as exc: msg = f"{' '.join(cmd)} did not run: {exc}" @@ -75,13 +97,30 @@ class Phone: _fail(f"{' '.join(cmd)} exited {done.returncode}: {detail}") return str(done.stdout) + def _is_phone(self, serial: str) -> bool: + """Whether an ``adb devices`` entry is the phone. + + USB and mDNS serials embed the phone's serial; a plain ``ip:port`` + wifi entry does not, so that one is asked for its ``ro.serialno``. + """ + if self.serial in serial: + return True + if ":" not in serial: + return False + try: + asked = self._adb("-s", serial, "shell", "getprop", "ro.serialno") + except AdbError as exc: + _logger.warning("Could not read the serial of %s (%s)", serial, exc) + return False + return asked.strip() == self.serial + def find_device(self) -> Device: """The attached phone, over USB or wifi (a wireless serial embeds it).""" listing = self._adb("devices") states: list[str] = [] for line in listing.splitlines()[1:]: serial, _, state = line.partition("\t") - if self.serial not in serial: + if not self._is_phone(serial): continue if state.strip() == "device": return Device(serial, "attached") @@ -90,9 +129,14 @@ class Phone: return Device(None, f"phone is {'/'.join(states)}, not ready") return Device(None, f"phone {self.serial} is not attached") - def installed(self, dev: str) -> bool: - """Whether the package is on the phone (exact ``package:`` line).""" - out = self._adb("-s", dev, "shell", "pm", "list", "packages", PACKAGE) + def installed(self, dev: str, *, hidden: bool = True) -> bool: + """Whether the package is on the phone (exact ``package:`` line). + + ``hidden`` adds ``-u`` so an app that focus mode hides or disables + still counts as present; it must not be mistaken for an uninstall. + """ + flag = ["-u"] if hidden else [] + out = self._adb("-s", dev, "shell", "pm", "list", "packages", *flag, PACKAGE) return f"package:{PACKAGE}" in out.split() def backup(self, dev: str, final: Path) -> tuple[Path, int]: @@ -108,7 +152,7 @@ class Phone: staging = final.with_name(final.name + ".partial") shutil.rmtree(staging, ignore_errors=True) staging.mkdir(parents=True) - self._adb("-s", dev, "pull", REMOTE_DIR, str(staging)) + self._adb("-s", dev, "pull", REMOTE_DIR, str(staging), timeout=_PULL_TIMEOUT_S) pulled = staging / Path(REMOTE_DIR).name got = len(list(pulled.rglob("*.tcx"))) if pulled.is_dir() else -1 if got < wanted: @@ -132,6 +176,6 @@ class Phone: raise AdbError(msg) from exc if "Success" not in out: _fail(f"uninstall did not report Success: {out.strip()!r}") - if self.installed(dev): + if self.installed(dev, hidden=False): _fail(f"{PACKAGE} is still listed after uninstall") _logger.info("Uninstalled %s from %s", PACKAGE, dev) diff --git a/screen_locker/_runnerup_retire_result.py b/screen_locker/_runnerup_retire_result.py new file mode 100644 index 0000000..28955a1 --- /dev/null +++ b/screen_locker/_runnerup_retire_result.py @@ -0,0 +1,62 @@ +"""Outcomes of one RunnerUp-retirement pass, and how loudly each is logged.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +import logging + +__all__ = ["LEVEL", "Outcome", "RetireResult"] + + +class Outcome(StrEnum): + """How a run ended; the value is what the journal shows.""" + + UNINSTALLED = "uninstalled" + ALREADY_GONE = "already-gone" + ALREADY_DONE = "already-done" + NO_ANCHOR = "no-anchor" + SOURCES_MISSING = "sources-missing" + NOT_DUE = "not-due" + DRY_RUN = "dry-run" + DEVICE_ABSENT = "device-absent" + MARKER_INVALID = "marker-invalid" + ADB_MISSING = "adb-missing" + PHONE_ERROR = "phone-error" + BACKUP_FAILED = "backup-failed" + UNINSTALL_FAILED = "uninstall-failed" + + +# Level per outcome; the failures also make the unit exit non-zero. +LEVEL = { + Outcome.DEVICE_ABSENT: logging.WARNING, + Outcome.SOURCES_MISSING: logging.WARNING, + Outcome.MARKER_INVALID: logging.WARNING, + Outcome.ADB_MISSING: logging.ERROR, + Outcome.PHONE_ERROR: logging.ERROR, + Outcome.BACKUP_FAILED: logging.ERROR, + Outcome.UNINSTALL_FAILED: logging.ERROR, +} +_NONZERO = frozenset( + { + Outcome.MARKER_INVALID, + Outcome.ADB_MISSING, + Outcome.PHONE_ERROR, + Outcome.BACKUP_FAILED, + Outcome.UNINSTALL_FAILED, + } +) + + +@dataclass(frozen=True) +class RetireResult: + """What happened and why, in a sentence a human can act on.""" + + outcome: Outcome + reason: str + logged: bool = False # the failure was already logged where it was caught + + @property + def ok(self) -> bool: + """False for outcomes the unit should surface as failed.""" + return self.outcome not in _NONZERO diff --git a/screen_locker/_runnerup_retire_state.py b/screen_locker/_runnerup_retire_state.py index 97186f0..c16b151 100644 --- a/screen_locker/_runnerup_retire_state.py +++ b/screen_locker/_runnerup_retire_state.py @@ -8,7 +8,7 @@ and never moved: a later, earlier-looking credit must not push the deadline. from __future__ import annotations -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, fields from datetime import datetime import json import logging @@ -36,10 +36,10 @@ class RetireState: def load_state(path: Path) -> RetireState: """Read the state; a missing file is a fresh start, a bad one is loud.""" + if not path.exists(): + return RetireState() try: raw = json.loads(path.read_text(encoding="utf-8")) - except FileNotFoundError: - return RetireState() except (OSError, ValueError) as exc: _logger.warning( "Retire state %s is unreadable (%s): starting from scratch, so the " @@ -51,8 +51,8 @@ def load_state(path: Path) -> RetireState: if not isinstance(raw, dict): _logger.warning("Retire state %s is not a JSON object: ignored", path) return RetireState() - fields = RetireState.__dataclass_fields__ - clean = {k: v for k, v in raw.items() if k in fields and isinstance(v, str)} + known = {f.name for f in fields(RetireState)} + clean = {k: v for k, v in raw.items() if k in known and isinstance(v, str)} try: state = RetireState(**clean) _ = state.anchor diff --git a/screen_locker/_runnerup_retire_tcx.py b/screen_locker/_runnerup_retire_tcx.py index 885da7a..d9d51a2 100644 --- a/screen_locker/_runnerup_retire_tcx.py +++ b/screen_locker/_runnerup_retire_tcx.py @@ -116,16 +116,19 @@ def _credit_ends(log_file: Path) -> list[tuple[float, datetime]]: return found -def find_anchor(log_file: Path, dirs: Iterable[Path]) -> datetime | None: - """When the PC first credited an app-recorded run, or None if it never did.""" - ends = [ +def app_run_ends(dirs: Iterable[Path]) -> list[float]: + """The end time of every app-authored TCX that has a usable one.""" + return [ end for directory in dirs for path in list_tcx(directory) if is_app_tcx(path) and (end := _ended_at(path)) is not None ] - if not ends: - return None + + +def find_anchor(log_file: Path, dirs: Iterable[Path]) -> datetime | None: + """When the PC first credited an app-recorded run, or None if it never did.""" + ends = app_run_ends(dirs) credited = [ when for run_end, when in _credit_ends(log_file) @@ -164,12 +167,15 @@ def _marker_names(marker: object) -> tuple[list[str] | None, str]: def _read_marker(marker_file: Path) -> tuple[object, str]: """The parsed marker JSON, or ``(None, why it could not be read)``.""" - try: - return json.loads(marker_file.read_text(encoding="utf-8")), "" - except FileNotFoundError: + if not marker_file.exists(): why = f"{marker_file} is missing: the app has not imported yet" - except (OSError, ValueError) as exc: - why = f"{marker_file} is unreadable ({exc})" + else: + try: + return json.loads(marker_file.read_text(encoding="utf-8")), "" + except (OSError, ValueError) as exc: + why = f"{marker_file} is unreadable ({exc})" + _logger.warning("Import marker problem: %s", why) + return None, why _logger.warning("Import marker problem: %s", why) return None, why diff --git a/screen_locker/runnerup_retire.py b/screen_locker/runnerup_retire.py index 646eaeb..14cba01 100644 --- a/screen_locker/runnerup_retire.py +++ b/screen_locker/runnerup_retire.py @@ -14,6 +14,7 @@ import argparse from datetime import UTC, datetime import logging import sys +from typing import TYPE_CHECKING from screen_locker._runnerup_retire import ( GRACE, @@ -25,12 +26,15 @@ from screen_locker._runnerup_retire_adb import PHONE_SERIAL, AdbError, Phone from screen_locker._runnerup_retire_state import load_state from screen_locker._runnerup_retire_tcx import check_marker +if TYPE_CHECKING: + from screen_locker._runnerup_retire_adb import PhoneLike + __all__ = ["main", "status_lines"] _logger = logging.getLogger(__name__) -def _installed_line(phone: Phone, *, query: bool) -> str: +def _installed_line(phone: PhoneLike, *, query: bool) -> str: if not query: return "installed: not queried (add --query-phone)" try: @@ -44,7 +48,7 @@ def _installed_line(phone: Phone, *, query: bool) -> str: def status_lines( - paths: Paths, phone: Phone, now: datetime, *, query_phone: bool = False + paths: Paths, phone: PhoneLike, now: datetime, *, query_phone: bool = False ) -> list[str]: """The ``--status`` report. Read-only: nothing is created or saved.""" state = load_state(paths.state_file) @@ -69,7 +73,7 @@ def status_lines( def main(argv: list[str] | None = None) -> int: """CLI entry; returns the process exit code.""" - parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser = argparse.ArgumentParser(description=__doc__.split("\n\n", maxsplit=1)[0]) parser.add_argument( "--status", action="store_true", help="print state, change nothing" ) diff --git a/screen_locker/tests/_retire_helpers.py b/screen_locker/tests/_retire_helpers.py index 50798c0..883dd45 100644 --- a/screen_locker/tests/_retire_helpers.py +++ b/screen_locker/tests/_retire_helpers.py @@ -87,23 +87,27 @@ class FakePhone: device: Device = field(default_factory=lambda: Device("S", "attached")) present: bool = True ask_error: AdbError | None = None - backup_error: AdbError | OSError | None = None + backup_error: Exception | None = None uninstall_error: AdbError | None = None calls: list[str] = field(default_factory=list) def available(self) -> bool: + """Available.""" return self.adb def find_device(self) -> Device: + """Find device.""" if self.ask_error: raise self.ask_error return self.device - def installed(self, dev: str) -> bool: + def installed(self, dev: str, *, hidden: bool = True) -> bool: + """Installed.""" self.calls.append(f"installed {dev}") return self.present def backup(self, dev: str, final: Path) -> tuple[Path, int]: + """Backup.""" self.calls.append("backup") if self.backup_error: raise self.backup_error @@ -111,6 +115,7 @@ class FakePhone: return final, 3 def uninstall(self, dev: str) -> None: + """Uninstall.""" self.calls.append("uninstall") if self.uninstall_error: raise self.uninstall_error diff --git a/screen_locker/tests/test_runnerup_retire.py b/screen_locker/tests/test_runnerup_retire.py index 1ccbfd0..82f69b0 100644 --- a/screen_locker/tests/test_runnerup_retire.py +++ b/screen_locker/tests/test_runnerup_retire.py @@ -25,23 +25,46 @@ if TYPE_CHECKING: from pathlib import Path -def go(paths: Paths, phone: FakePhone, now: datetime = DUE, **kw: bool): - return run_retire(paths, phone, now, **kw) # type: ignore[arg-type] +def go(paths: Paths, phone: FakePhone, now: datetime = DUE, *, dry_run: bool = False): + """Go.""" + return run_retire(paths, phone, now, dry_run=dry_run) def test_no_anchor_never_touches_anything(tmp_path: Path) -> None: + """Test no anchor never touches anything.""" paths = world(tmp_path, anchored=False) phone = FakePhone() result = go(paths, phone, DUE + timedelta(days=900)) assert result.outcome is Outcome.NO_ANCHOR assert result.ok - assert phone.calls == [] + assert not phone.calls assert not paths.state_file.exists() +def test_missing_sources_are_a_warning_not_a_quiet_nothing( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Test missing sources are a warning not a quiet nothing.""" + paths = Paths(tmp_path / "s.json", tmp_path / "nodrop", tmp_path / "nolog.json") + with caplog.at_level(logging.INFO): + result = go(paths, FakePhone()) + assert result.outcome is Outcome.SOURCES_MISSING + assert result.ok + assert "nodrop" in result.reason + assert "nolog.json" in result.reason + assert caplog.records[-1].levelno == logging.WARNING + + +def test_no_anchor_says_how_many_app_runs_it_saw(tmp_path: Path) -> None: + """Test no anchor says how many app runs it saw.""" + paths = world(tmp_path, anchored=False) + assert "(1 app TCX on disk)" in go(paths, FakePhone()).reason + + def test_not_due_persists_the_anchor_and_says_when( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test not due persists the anchor and says when.""" paths = world(tmp_path) with caplog.at_level(logging.INFO): result = go(paths, FakePhone(), DUE - timedelta(minutes=1)) @@ -52,6 +75,7 @@ def test_not_due_persists_the_anchor_and_says_when( def test_stored_anchor_wins_over_the_log(tmp_path: Path) -> None: + """Test stored anchor wins over the log.""" paths = world(tmp_path, anchored=False) save_state(paths.state_file, RetireState(anchor_at=ANCHOR.isoformat())) assert go(paths, FakePhone(), ANCHOR).outcome is Outcome.NOT_DUE @@ -60,6 +84,7 @@ def test_stored_anchor_wins_over_the_log(tmp_path: Path) -> None: def test_dry_run_writes_nothing_and_stops_before_the_phone_changes( tmp_path: Path, ) -> None: + """Test dry run writes nothing and stops before the phone changes.""" paths = world(tmp_path) phone = FakePhone() result = go(paths, phone, dry_run=True) @@ -75,6 +100,7 @@ def test_dry_run_writes_nothing_and_stops_before_the_phone_changes( def test_invalid_marker_blocks_the_uninstall_loudly( tmp_path: Path, caplog: pytest.LogCaptureFixture, variant: str ) -> None: + """Test invalid marker blocks the uninstall loudly.""" paths = world(tmp_path, marker=variant == "mismatch") if variant == "mismatch": put_marker(paths.watch_dir, [RUN], count=2) @@ -84,11 +110,12 @@ def test_invalid_marker_blocks_the_uninstall_loudly( assert result.outcome is Outcome.MARKER_INVALID assert not result.ok assert "NOT uninstalling" in result.reason - assert phone.calls == [] + assert not phone.calls assert caplog.records[-1].levelno == logging.WARNING def test_adb_missing_is_a_failure(tmp_path: Path) -> None: + """Test adb missing is a failure.""" result = go(world(tmp_path), FakePhone(adb=False)) assert result.outcome is Outcome.ADB_MISSING assert not result.ok @@ -97,18 +124,20 @@ def test_adb_missing_is_a_failure(tmp_path: Path) -> None: def test_absent_phone_retries_quietly_but_warns( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test absent phone retries quietly but warns.""" phone = FakePhone(device=Device(None, "phone S is not attached")) with caplog.at_level(logging.WARNING): result = go(world(tmp_path), phone) assert result.outcome is Outcome.DEVICE_ABSENT assert result.ok assert "not attached" in caplog.text - assert phone.calls == [] + assert not phone.calls def test_adb_error_while_asking_is_a_logged_failure( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test adb error while asking is a logged failure.""" phone = FakePhone(ask_error=AdbError("server died")) with caplog.at_level(logging.ERROR): result = go(world(tmp_path), phone) @@ -118,6 +147,7 @@ def test_adb_error_while_asking_is_a_logged_failure( def test_already_uninstalled_is_recorded_done(tmp_path: Path) -> None: + """Test already uninstalled is recorded done.""" paths = world(tmp_path) result = go(paths, FakePhone(present=False)) assert result.outcome is Outcome.ALREADY_GONE @@ -126,6 +156,7 @@ def test_already_uninstalled_is_recorded_done(tmp_path: Path) -> None: def test_full_run_backs_up_then_uninstalls_then_never_again(tmp_path: Path) -> None: + """Test full run backs up then uninstalls then never again.""" paths = world(tmp_path) phone = FakePhone() result = go(paths, phone) @@ -133,21 +164,22 @@ def test_full_run_backs_up_then_uninstalls_then_never_again(tmp_path: Path) -> N assert phone.calls == ["installed S", "backup", "uninstall"] backup = paths.watch_dir / "runnerup_phone_backup_2026-11-08" assert backup.is_dir() - assert list(paths.watch_dir.glob("*.tcx")) == [] # nothing top-level + assert not list(paths.watch_dir.glob("*.tcx")) # nothing top-level state = load_state(paths.state_file) assert state.anchor == ANCHOR assert state.backup_path == str(backup) assert state.done_at again = FakePhone() assert go(paths, again).outcome is Outcome.ALREADY_DONE - assert again.calls == [] + assert not again.calls @pytest.mark.parametrize("error", [AdbError("short pull"), OSError("disk full")]) def test_failed_backup_blocks_the_uninstall( tmp_path: Path, caplog: pytest.LogCaptureFixture, error: Exception ) -> None: - phone = FakePhone(backup_error=error) # type: ignore[arg-type] + """Test failed backup blocks the uninstall.""" + phone = FakePhone(backup_error=error) with caplog.at_level(logging.ERROR): result = go(world(tmp_path), phone) assert result.outcome is Outcome.BACKUP_FAILED @@ -157,6 +189,7 @@ def test_failed_backup_blocks_the_uninstall( def test_failed_uninstall_keeps_the_backup_for_the_retry(tmp_path: Path) -> None: + """Test failed uninstall keeps the backup for the retry.""" paths = world(tmp_path) first = FakePhone(uninstall_error=AdbError("DELETE_FAILED_DEVICE_POLICY_MANAGER")) result = go(paths, first) @@ -169,6 +202,7 @@ def test_failed_uninstall_keeps_the_backup_for_the_retry(tmp_path: Path) -> None def test_default_paths_are_the_module_constants() -> None: + """Test default paths are the module constants.""" paths = Paths.default() assert paths.state_file == rr.STATE_FILE assert paths.marker == rr.WATCH_DIR / rr.MARKER_NAME @@ -177,5 +211,6 @@ def test_default_paths_are_the_module_constants() -> None: def test_run_retire_defaults_to_the_real_clock(tmp_path: Path) -> None: + """Test run retire defaults to the real clock.""" paths = world(tmp_path) assert run_retire(paths, FakePhone(), dry_run=True).outcome is Outcome.DRY_RUN diff --git a/screen_locker/tests/test_runnerup_retire_adb.py b/screen_locker/tests/test_runnerup_retire_adb.py index 3786a89..65625f0 100644 --- a/screen_locker/tests/test_runnerup_retire_adb.py +++ b/screen_locker/tests/test_runnerup_retire_adb.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json from pathlib import Path import subprocess from types import SimpleNamespace @@ -11,7 +10,6 @@ from typing import Any import pytest from screen_locker import _runnerup_retire_adb as ra -from screen_locker._runnerup_retire_state import RetireState, load_state, save_state SERIAL = ra.PHONE_SERIAL PKG_LINE = f"package:{ra.PACKAGE}\n" @@ -21,10 +19,12 @@ class FakeAdb: """A ``subprocess.run`` stand-in: replies by the adb sub-command.""" def __init__(self, replies: dict[str, Any] | None = None) -> None: + """Init .""" self.replies = replies or {} self.calls: list[list[str]] = [] def __call__(self, cmd: list[str], **_: object) -> SimpleNamespace: + """Call .""" self.calls.append(cmd) for key, reply in self.replies.items(): if key in " ".join(cmd): @@ -36,10 +36,12 @@ class FakeAdb: def phone(fake: FakeAdb, adb: str | None = "/usr/bin/adb") -> ra.Phone: + """Phone.""" return ra.Phone(run=fake, which=lambda _name: adb) def test_available_follows_which() -> None: + """Test available follows which.""" assert phone(FakeAdb()).available() is True assert phone(FakeAdb(), adb=None).available() is False @@ -48,11 +50,13 @@ def test_available_follows_which() -> None: "boom", [FileNotFoundError("adb"), subprocess.TimeoutExpired("adb", 1)] ) def test_adb_failing_to_run_is_an_adb_error(boom: BaseException) -> None: + """Test adb failing to run is an adb error.""" with pytest.raises(ra.AdbError, match="did not run"): phone(FakeAdb({"devices": boom})).find_device() def test_nonzero_exit_carries_the_output() -> None: + """Test nonzero exit carries the output.""" fake = FakeAdb({"devices": (1, "", "daemon not running")}) with pytest.raises(ra.AdbError, match="daemon not running"): phone(fake).find_device() @@ -73,12 +77,69 @@ def test_nonzero_exit_carries_the_output() -> None: ], ) def test_find_device(listing: str, serial: str | None, reason: str) -> None: + """Test find device.""" device = phone(FakeAdb({"devices": (0, listing, "")})).find_device() assert device.serial == serial assert reason in device.reason +def test_wifi_entry_is_asked_for_its_hardware_serial() -> None: + """Test wifi entry is asked for its hardware serial.""" + listing = "List of devices attached\n192.168.1.5:5555\tdevice\n" + fake = FakeAdb({"devices": (0, listing, ""), "getprop": (0, f"{SERIAL}\n", "")}) + device = phone(fake).find_device() + assert device.serial == "192.168.1.5:5555" + assert ["adb", "-s", "192.168.1.5:5555", "shell", "getprop", "ro.serialno"] in ( + fake.calls + ) + + +def test_wifi_entry_of_another_phone_is_not_ours() -> None: + """Test wifi entry of another phone is not ours.""" + listing = "List of devices attached\n10.0.0.2:5555\tdevice\n" + fake = FakeAdb({"devices": (0, listing, ""), "getprop": (0, "OTHER\n", "")}) + assert phone(fake).find_device().serial is None + + +def test_wifi_entry_that_will_not_answer_is_warned_about( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test wifi entry that will not answer is warned about.""" + listing = "List of devices attached\n10.0.0.2:5555\tdevice\n" + fake = FakeAdb({"devices": (0, listing, ""), "getprop": (1, "", "offline")}) + with caplog.at_level("WARNING"): + assert phone(fake).find_device().serial is None + assert "Could not read the serial" in caplog.text + + +def test_presence_probe_sees_hidden_packages_but_the_final_check_does_not() -> None: + """Test presence probe sees hidden packages but the final check does not.""" + fake = FakeAdb({"pm list": (0, PKG_LINE, "")}) + phone(fake).installed("dev") + phone(fake).installed("dev", hidden=False) + assert "-u" in fake.calls[0] + assert "-u" not in fake.calls[1] + + +def test_pull_gets_a_long_timeout_and_other_calls_the_short_one( + tmp_path: Path, +) -> None: + """Test pull gets a long timeout and other calls the short one.""" + timeouts: dict[str, object] = {} + make = _pull_creates(1) + + def run(cmd: list[str], **kw: object) -> SimpleNamespace: + """Run.""" + timeouts[cmd[3]] = kw["timeout"] + code, out, err = make(cmd) if cmd[3] == "pull" else (0, "a.tcx\n", "") + return SimpleNamespace(returncode=code, stdout=out, stderr=err) + + ra.Phone(run=run, which=lambda _n: "adb").backup("dev", tmp_path / "b") + assert timeouts == {"shell": 50, "pull": 900} + + def test_installed_needs_the_exact_package_line() -> None: + """Test installed needs the exact package line.""" yes = FakeAdb({"pm list": (0, PKG_LINE, "")}) assert phone(yes).installed("dev") is True assert yes.calls[0][:3] == ["adb", "-s", "dev"] @@ -87,7 +148,10 @@ def test_installed_needs_the_exact_package_line() -> None: def _pull_creates(count: int, root: str = "RunnerUp"): + """Pull creates.""" + def make(cmd: list[str]) -> tuple[int, str, str]: + """Make.""" folder = Path(cmd[-1]) / root folder.mkdir(parents=True) for i in range(count): @@ -98,6 +162,7 @@ def _pull_creates(count: int, root: str = "RunnerUp"): def test_backup_stages_then_renames_and_counts(tmp_path: Path) -> None: + """Test backup stages then renames and counts.""" fake = FakeAdb({"ls": (0, "a.tcx\nb.tcx\nb.gpx\n", ""), "pull": _pull_creates(2)}) final = tmp_path / "backup_2026" target, count = phone(fake).backup("dev", final) @@ -107,6 +172,7 @@ def test_backup_stages_then_renames_and_counts(tmp_path: Path) -> None: def test_backup_never_overwrites_an_existing_folder(tmp_path: Path) -> None: + """Test backup never overwrites an existing folder.""" final = tmp_path / "backup" final.mkdir() (tmp_path / "backup.partial").mkdir() # a stale half-pull is cleared @@ -116,6 +182,7 @@ def test_backup_never_overwrites_an_existing_folder(tmp_path: Path) -> None: def test_backup_short_pull_is_an_error_and_not_renamed(tmp_path: Path) -> None: + """Test backup short pull is an error and not renamed.""" fake = FakeAdb({"ls": (0, "a.tcx\nb.tcx\n", ""), "pull": _pull_creates(1)}) with pytest.raises(ra.AdbError, match="pulled 1 of 2"): phone(fake).backup("dev", tmp_path / "b") @@ -123,18 +190,21 @@ def test_backup_short_pull_is_an_error_and_not_renamed(tmp_path: Path) -> None: def test_backup_pull_that_creates_nothing_is_an_error(tmp_path: Path) -> None: + """Test backup pull that creates nothing is an error.""" fake = FakeAdb({"ls": (0, "a.tcx\n", "")}) with pytest.raises(ra.AdbError, match="pulled -1 of 1"): phone(fake).backup("dev", tmp_path / "b") def test_uninstall_succeeds_only_when_the_package_is_gone() -> None: + """Test uninstall succeeds only when the package is gone.""" fake = FakeAdb({"uninstall": (0, "Success\n", ""), "pm list": (0, "", "")}) phone(fake).uninstall("dev") assert ["adb", "-s", "dev", "uninstall", ra.PACKAGE] in fake.calls def test_uninstall_refusal_keeps_adbs_reason() -> None: + """Test uninstall refusal keeps adbs reason.""" fail = "Failure [DELETE_FAILED_DEVICE_POLICY_MANAGER]" fake = FakeAdb({"uninstall": (1, fail, "")}) with pytest.raises(ra.AdbError, match="DELETE_FAILED_DEVICE_POLICY_MANAGER"): @@ -142,47 +212,14 @@ def test_uninstall_refusal_keeps_adbs_reason() -> None: def test_uninstall_without_success_text_is_an_error() -> None: + """Test uninstall without success text is an error.""" fake = FakeAdb({"uninstall": (0, "weird\n", "")}) with pytest.raises(ra.AdbError, match="did not report Success"): phone(fake).uninstall("dev") def test_uninstall_that_leaves_the_package_is_an_error() -> None: + """Test uninstall that leaves the package is an error.""" fake = FakeAdb({"uninstall": (0, "Success\n", ""), "pm list": (0, PKG_LINE, "")}) with pytest.raises(ra.AdbError, match="still listed"): phone(fake).uninstall("dev") - - -def test_state_roundtrip_and_anchor(tmp_path: Path) -> None: - path = tmp_path / "deep" / "s.json" - assert load_state(path) == RetireState() - save_state(path, RetireState(anchor_at="2026-11-01T10:45:00+00:00")) - assert load_state(path).anchor is not None - assert RetireState().anchor is None - - -@pytest.mark.parametrize( - "body", ["{nope", "[]", json.dumps({"anchor_at": "not a date"})] -) -def test_bad_state_is_loud_and_ignored( - tmp_path: Path, body: str, caplog: pytest.LogCaptureFixture -) -> None: - path = tmp_path / "s.json" - path.write_text(body) - with caplog.at_level("WARNING"): - assert load_state(path) == RetireState() - assert "Retire state" in caplog.text - - -def test_unreadable_state_is_loud( - tmp_path: Path, caplog: pytest.LogCaptureFixture -) -> None: - with caplog.at_level("WARNING"): - assert load_state(tmp_path) == RetireState() - assert "unreadable" in caplog.text - - -def test_state_ignores_unknown_and_non_string_fields(tmp_path: Path) -> None: - path = tmp_path / "s.json" - path.write_text(json.dumps({"done_at": 5, "extra": "x", "backup_path": "/b"})) - assert load_state(path) == RetireState(backup_path="/b") diff --git a/screen_locker/tests/test_runnerup_retire_cli.py b/screen_locker/tests/test_runnerup_retire_cli.py index 9b5a2ea..b1f589f 100644 --- a/screen_locker/tests/test_runnerup_retire_cli.py +++ b/screen_locker/tests/test_runnerup_retire_cli.py @@ -19,12 +19,14 @@ if TYPE_CHECKING: def _tree(root: Path) -> set[str]: + """Tree.""" return {str(p.relative_to(root)) for p in root.rglob("*")} def test_status_without_anchor(tmp_path: Path) -> None: + """Test status without anchor.""" paths = world(tmp_path, anchored=False, marker=False) - lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type] + lines = cli.status_lines(paths, FakePhone(), DUE) assert lines[0].startswith("anchor: none") assert lines[1] == "due: never" assert "INVALID" in lines[2] @@ -35,9 +37,10 @@ def test_status_without_anchor(tmp_path: Path) -> None: def test_status_is_strictly_read_only_and_marks_a_derived_anchor( tmp_path: Path, ) -> None: + """Test status is strictly read only and marks a derived anchor.""" paths = world(tmp_path) before = _tree(tmp_path) - lines = cli.status_lines(paths, FakePhone(), DUE - timedelta(days=1)) # type: ignore[arg-type] + lines = cli.status_lines(paths, FakePhone(), DUE - timedelta(days=1)) assert _tree(tmp_path) == before assert not paths.state_file.parent.exists() assert "derived, not yet stored" in lines[0] @@ -46,12 +49,13 @@ def test_status_is_strictly_read_only_and_marks_a_derived_anchor( def test_status_with_stored_anchor_and_done(tmp_path: Path) -> None: + """Test status with stored anchor and done.""" paths = world(tmp_path, anchored=False) save_state( paths.state_file, RetireState(anchor_at=ANCHOR.isoformat(), done_at="2026-11-09T00:00:00+00:00"), ) - lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type] + lines = cli.status_lines(paths, FakePhone(), DUE) assert "(stored)" in lines[0] assert lines[3] == "done: 2026-11-09T00:00:00+00:00" @@ -69,15 +73,17 @@ def test_status_with_stored_anchor_and_done(tmp_path: Path) -> None: ], ) def test_status_query_phone(tmp_path: Path, phone: FakePhone, expected: str) -> None: + """Test status query phone.""" paths = world(tmp_path) - lines = cli.status_lines(paths, phone, DUE, query_phone=True) # type: ignore[arg-type] + lines = cli.status_lines(paths, phone, DUE, query_phone=True) assert lines[4] == expected def test_main_status_prints_and_exits_zero( capsys: pytest.CaptureFixture[str], ) -> None: - assert cli.main(["--status"]) == 0 # the isolated defaults: an empty world + """Test main status prints and exits zero.""" + assert not cli.main(["--status"]) # the isolated defaults: an empty world out = capsys.readouterr().out assert "anchor: none" in out assert "not queried" in out @@ -86,14 +92,16 @@ def test_main_status_prints_and_exits_zero( def test_main_status_query_phone_uses_the_serial( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: + """Test main status query phone uses the serial.""" seen: list[str] = [] def fake_phone(serial: str) -> FakePhone: + """Fake phone.""" seen.append(serial) return FakePhone() monkeypatch.setattr(cli, "Phone", fake_phone) - assert cli.main(["--status", "--query-phone", "--serial", "XYZ"]) == 0 + assert not cli.main(["--status", "--query-phone", "--serial", "XYZ"]) assert seen == ["XYZ"] assert "installed: yes" in capsys.readouterr().out @@ -105,9 +113,11 @@ def test_main_status_query_phone_uses_the_serial( def test_main_exit_code_follows_the_outcome( monkeypatch: pytest.MonkeyPatch, outcome: Outcome, code: int ) -> None: + """Test main exit code follows the outcome.""" calls: list[bool] = [] def fake_run(paths: Paths, phone: object, *, dry_run: bool) -> RetireResult: + """Fake run.""" calls.append(dry_run) return RetireResult(outcome, "why") @@ -117,13 +127,15 @@ def test_main_exit_code_follows_the_outcome( def test_main_with_nothing_to_retire_is_a_clean_noop() -> None: - assert cli.main([]) == 0 + """Test main with nothing to retire is a clean noop.""" + assert not cli.main([]) def test_status_marker_line_for_a_valid_marker_uses_real_files( tmp_path: Path, ) -> None: + """Test status marker line for a valid marker uses real files.""" paths = world(tmp_path, marker=False) put_marker(paths.watch_dir, [RUN]) now = datetime.now(UTC) - assert cli.status_lines(paths, FakePhone(), now)[2].startswith("marker: valid") # type: ignore[arg-type] + assert cli.status_lines(paths, FakePhone(), now)[2].startswith("marker: valid") diff --git a/screen_locker/tests/test_runnerup_retire_marker.py b/screen_locker/tests/test_runnerup_retire_marker.py new file mode 100644 index 0000000..f2ecedd --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire_marker.py @@ -0,0 +1,117 @@ +"""RunnerUp retirement: validating the app's import marker against processed/.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import pytest + +from screen_locker import _runnerup_retire_tcx as rt +from screen_locker.tests._retire_helpers import APP, put_marker, put_tcx + +if TYPE_CHECKING: + from pathlib import Path + + +def _processed(tmp_path: Path) -> Path: + """Processed.""" + processed = tmp_path / "processed" + put_tcx(processed, "RunnerUp_2026-08-12-17-42-11_Running.tcx") + put_tcx(processed, "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx") + put_tcx(processed, "RunnerUp_2026-08-14-00-58-45_Running.tcx") + put_tcx(processed, "app_2026-11-01.tcx", author=APP) + return processed + + +_LISTED = [ + "RunnerUp_2026-08-12-17-42-11_Running.tcx", + "RunnerUp_2026-08-14-00-58-45_Running.tcx", +] + + +def test_marker_valid_counts_device_twins_and_ignores_app_files( + tmp_path: Path, +) -> None: + """Test marker valid counts device twins and ignores app files.""" + processed = _processed(tmp_path) + marker = put_marker(tmp_path, _LISTED) + check = rt.check_marker(marker, processed) + assert check.ok + assert "2 runs" in check.reason + + +def test_marker_may_name_the_device_prefixed_copy(tmp_path: Path) -> None: + """Test marker may name the device prefixed copy.""" + processed = _processed(tmp_path) + names = ["RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx", _LISTED[1]] + assert rt.check_marker(put_marker(tmp_path, names), processed).ok + + +def test_marker_missing_is_not_valid( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Test marker missing is not valid.""" + with caplog.at_level(logging.WARNING): + check = rt.check_marker(tmp_path / "m.json", tmp_path) + assert not check.ok + assert "has not imported yet" in check.reason + assert "marker problem" in caplog.text + + +def test_marker_unreadable_and_bad_json(tmp_path: Path) -> None: + """Test marker unreadable and bad json.""" + folder_as_marker = tmp_path / "m" + folder_as_marker.mkdir() + assert "unreadable" in rt.check_marker(folder_as_marker, tmp_path).reason + bad = tmp_path / "bad.json" + bad.write_text("{nope") + assert "unreadable" in rt.check_marker(bad, tmp_path).reason + + +@pytest.mark.parametrize( + ("body", "fragment"), + [ + ("[]", "not a JSON object"), + ('{"count": 0, "tcx_names": []}', "no completed_at"), + ('{"completed_at": "", "count": 0, "tcx_names": []}', "no completed_at"), + ('{"completed_at": "t", "count": 0, "tcx_names": "x"}', "not a list"), + ('{"completed_at": "t", "count": 0, "tcx_names": [1]}', "not a list"), + ('{"completed_at": "t", "count": 3, "tcx_names": ["a"]}', "count 3"), + ('{"completed_at": "t", "count": true, "tcx_names": ["a"]}', "count True"), + ('{"completed_at": "t", "count": "1", "tcx_names": ["a"]}', "count '1'"), + ('{"completed_at": "t", "count": 1, "tcx_names": ["x/a"]}', "bare file"), + ('{"completed_at": "t", "count": 1, "tcx_names": [".a"]}', "bare file"), + ], +) +def test_marker_shape_errors(tmp_path: Path, body: str, fragment: str) -> None: + """Test marker shape errors.""" + marker = tmp_path / "m.json" + marker.write_text(body) + check = rt.check_marker(marker, tmp_path) + assert not check.ok + assert fragment in check.reason + + +def test_marker_listing_one_run_twice_is_invalid(tmp_path: Path) -> None: + """Test marker listing one run twice is invalid.""" + processed = _processed(tmp_path) + twin = "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx" + marker = put_marker(tmp_path, [_LISTED[0], twin]) + assert "more than once" in rt.check_marker(marker, processed).reason + + +def test_marker_naming_a_run_not_in_processed_is_invalid(tmp_path: Path) -> None: + """Test marker naming a run not in processed is invalid.""" + processed = _processed(tmp_path) + marker = put_marker(tmp_path, [*_LISTED, "RunnerUp_2099-01-01-00-00-00_X.tcx"]) + reason = rt.check_marker(marker, processed).reason + assert "not in processed/" in reason + + +def test_marker_missing_a_processed_run_asks_for_a_reimport(tmp_path: Path) -> None: + """Test marker missing a processed run asks for a reimport.""" + processed = _processed(tmp_path) + marker = put_marker(tmp_path, _LISTED[:1]) + reason = rt.check_marker(marker, processed).reason + assert "re-run the app's import" in reason diff --git a/screen_locker/tests/test_runnerup_retire_state.py b/screen_locker/tests/test_runnerup_retire_state.py new file mode 100644 index 0000000..56afca5 --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire_state.py @@ -0,0 +1,52 @@ +"""RunnerUp retirement: the state file round-trip and its loud failures.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from screen_locker._runnerup_retire_state import RetireState, load_state, save_state + +if TYPE_CHECKING: + from pathlib import Path + + +def test_state_roundtrip_and_anchor(tmp_path: Path) -> None: + """Test state roundtrip and anchor.""" + path = tmp_path / "deep" / "s.json" + assert load_state(path) == RetireState() + save_state(path, RetireState(anchor_at="2026-11-01T10:45:00+00:00")) + assert load_state(path).anchor is not None + assert RetireState().anchor is None + + +@pytest.mark.parametrize( + "body", ["{nope", "[]", json.dumps({"anchor_at": "not a date"})] +) +def test_bad_state_is_loud_and_ignored( + tmp_path: Path, body: str, caplog: pytest.LogCaptureFixture +) -> None: + """Test bad state is loud and ignored.""" + path = tmp_path / "s.json" + path.write_text(body) + with caplog.at_level("WARNING"): + assert load_state(path) == RetireState() + assert "Retire state" in caplog.text + + +def test_unreadable_state_is_loud( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Test unreadable state is loud.""" + with caplog.at_level("WARNING"): + assert load_state(tmp_path) == RetireState() + assert "unreadable" in caplog.text + + +def test_state_ignores_unknown_and_non_string_fields(tmp_path: Path) -> None: + """Test state ignores unknown and non string fields.""" + path = tmp_path / "s.json" + path.write_text(json.dumps({"done_at": 5, "extra": "x", "backup_path": "/b"})) + assert load_state(path) == RetireState(backup_path="/b") diff --git a/screen_locker/tests/test_runnerup_retire_tcx.py b/screen_locker/tests/test_runnerup_retire_tcx.py index 17d8982..171611c 100644 --- a/screen_locker/tests/test_runnerup_retire_tcx.py +++ b/screen_locker/tests/test_runnerup_retire_tcx.py @@ -14,7 +14,6 @@ from screen_locker.tests._retire_helpers import ( END_UNIX, credit, put_log, - put_marker, put_tcx, ) @@ -25,6 +24,7 @@ _AT = "2026-11-01T10:45:00+00:00" def test_run_key_drops_device_prefix_only() -> None: + """Test run key drops device prefix only.""" plain = "RunnerUp_2026-08-22-23-51-04_Running.tcx" twin = "RunnerUp_Pixel_6a_2026-08-22-23-51-04_Running.tcx" assert rt.run_key(twin) == plain @@ -32,6 +32,7 @@ def test_run_key_drops_device_prefix_only() -> None: def test_list_tcx_skips_dotfiles_other_types_and_missing_dir(tmp_path: Path) -> None: + """Test list tcx skips dotfiles other types and missing dir.""" assert rt.list_tcx(tmp_path / "nope") == [] put_tcx(tmp_path, "a.tcx") put_tcx(tmp_path, ".partial.tcx") @@ -52,10 +53,12 @@ def test_list_tcx_skips_dotfiles_other_types_and_missing_dir(tmp_path: Path) -> def test_is_app_tcx_reads_author_and_creator( tmp_path: Path, kwargs: dict[str, str], expected: bool ) -> None: + """Test is app tcx reads author and creator.""" assert rt.is_app_tcx(put_tcx(tmp_path, "r.tcx", **kwargs)) is expected def test_is_app_tcx_ignores_the_name_in_free_text(tmp_path: Path) -> None: + """Test is app tcx ignores the name in free text.""" path = tmp_path / "n.tcx" path.write_text(f"{APP}") assert rt.is_app_tcx(path) is False @@ -64,18 +67,21 @@ def test_is_app_tcx_ignores_the_name_in_free_text(tmp_path: Path) -> None: def test_is_app_tcx_unreadable_is_a_loud_non_app( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test is app tcx unreadable is a loud non app.""" with caplog.at_level(logging.WARNING): assert rt.is_app_tcx(tmp_path) is False assert "Cannot read" in caplog.text def _app_run(tmp_path: Path, name: str = "app.tcx", where: str = "") -> Path: + """App run.""" return put_tcx(tmp_path / where if where else tmp_path, name, author=APP) def test_anchor_is_the_credit_time_of_the_entry_matching_the_run_end( tmp_path: Path, ) -> None: + """Test anchor is the credit time of the entry matching the run end.""" _app_run(tmp_path, where="processed") log = tmp_path / "log.json" put_log( @@ -92,6 +98,7 @@ def test_anchor_is_the_credit_time_of_the_entry_matching_the_run_end( def test_anchor_accepts_a_stringified_end_and_the_earliest_credit( tmp_path: Path, ) -> None: + """Test anchor accepts a stringified end and the earliest credit.""" _app_run(tmp_path) log = tmp_path / "log.json" put_log( @@ -106,6 +113,7 @@ def test_anchor_accepts_a_stringified_end_and_the_earliest_credit( def test_same_day_runnerup_credit_does_not_anchor_the_app_run(tmp_path: Path) -> None: + """Test same day runnerup credit does not anchor the app run.""" _app_run(tmp_path) log = tmp_path / "log.json" put_log(log, {"2026-11-01": [credit(_AT, END_UNIX - 3600)]}) @@ -113,6 +121,7 @@ def test_same_day_runnerup_credit_does_not_anchor_the_app_run(tmp_path: Path) -> def test_no_app_run_means_no_anchor(tmp_path: Path) -> None: + """Test no app run means no anchor.""" put_tcx(tmp_path, "r.tcx") log = tmp_path / "log.json" put_log(log, {"d": [credit(_AT, END_UNIX)]}) @@ -122,6 +131,7 @@ def test_no_app_run_means_no_anchor(tmp_path: Path) -> None: def test_unreadable_or_trackless_app_tcx_cannot_anchor( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test unreadable or trackless app tcx cannot anchor.""" (tmp_path / "bad.tcx").write_text(f"{APP}") log = tmp_path / "log.json" put_log(log, {"d": [credit(_AT, END_UNIX)]}) @@ -133,6 +143,7 @@ def test_unreadable_or_trackless_app_tcx_cannot_anchor( def test_log_entries_that_cannot_match_are_skipped( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: + """Test log entries that cannot match are skipped.""" _app_run(tmp_path) log = tmp_path / "log.json" put_log( @@ -160,6 +171,7 @@ def test_log_entries_that_cannot_match_are_skipped( def test_naive_credit_timestamps_are_read_as_local_time(tmp_path: Path) -> None: + """Test naive credit timestamps are read as local time.""" _app_run(tmp_path) log = tmp_path / "log.json" put_log(log, {"d": [credit("2026-11-01T10:45:00", END_UNIX)]}) @@ -168,95 +180,10 @@ def test_naive_credit_timestamps_are_read_as_local_time(tmp_path: Path) -> None: assert found.tzinfo is not None -def _processed(tmp_path: Path) -> Path: - processed = tmp_path / "processed" - put_tcx(processed, "RunnerUp_2026-08-12-17-42-11_Running.tcx") - put_tcx(processed, "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx") - put_tcx(processed, "RunnerUp_2026-08-14-00-58-45_Running.tcx") - put_tcx(processed, "app_2026-11-01.tcx", author=APP) - return processed - - -_LISTED = [ - "RunnerUp_2026-08-12-17-42-11_Running.tcx", - "RunnerUp_2026-08-14-00-58-45_Running.tcx", -] - - -def test_marker_valid_counts_device_twins_and_ignores_app_files( - tmp_path: Path, -) -> None: - processed = _processed(tmp_path) - marker = put_marker(tmp_path, _LISTED) - check = rt.check_marker(marker, processed) - assert check.ok - assert "2 runs" in check.reason - - -def test_marker_may_name_the_device_prefixed_copy(tmp_path: Path) -> None: - processed = _processed(tmp_path) - names = ["RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx", _LISTED[1]] - assert rt.check_marker(put_marker(tmp_path, names), processed).ok - - -def test_marker_missing_is_not_valid( - tmp_path: Path, caplog: pytest.LogCaptureFixture -) -> None: - with caplog.at_level(logging.WARNING): - check = rt.check_marker(tmp_path / "m.json", tmp_path) - assert not check.ok - assert "has not imported yet" in check.reason - assert "marker problem" in caplog.text - - -def test_marker_unreadable_and_bad_json(tmp_path: Path) -> None: - folder_as_marker = tmp_path / "m" - folder_as_marker.mkdir() - assert "unreadable" in rt.check_marker(folder_as_marker, tmp_path).reason - bad = tmp_path / "bad.json" - bad.write_text("{nope") - assert "unreadable" in rt.check_marker(bad, tmp_path).reason - - -@pytest.mark.parametrize( - ("body", "fragment"), - [ - ("[]", "not a JSON object"), - ('{"count": 0, "tcx_names": []}', "no completed_at"), - ('{"completed_at": "", "count": 0, "tcx_names": []}', "no completed_at"), - ('{"completed_at": "t", "count": 0, "tcx_names": "x"}', "not a list"), - ('{"completed_at": "t", "count": 0, "tcx_names": [1]}', "not a list"), - ('{"completed_at": "t", "count": 3, "tcx_names": ["a"]}', "count 3"), - ('{"completed_at": "t", "count": true, "tcx_names": ["a"]}', "count True"), - ('{"completed_at": "t", "count": "1", "tcx_names": ["a"]}', "count '1'"), - ('{"completed_at": "t", "count": 1, "tcx_names": ["x/a"]}', "bare file"), - ('{"completed_at": "t", "count": 1, "tcx_names": [".a"]}', "bare file"), - ], -) -def test_marker_shape_errors(tmp_path: Path, body: str, fragment: str) -> None: - marker = tmp_path / "m.json" - marker.write_text(body) - check = rt.check_marker(marker, tmp_path) - assert not check.ok - assert fragment in check.reason - - -def test_marker_listing_one_run_twice_is_invalid(tmp_path: Path) -> None: - processed = _processed(tmp_path) - twin = "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx" - marker = put_marker(tmp_path, [_LISTED[0], twin]) - assert "more than once" in rt.check_marker(marker, processed).reason - - -def test_marker_naming_a_run_not_in_processed_is_invalid(tmp_path: Path) -> None: - processed = _processed(tmp_path) - marker = put_marker(tmp_path, [*_LISTED, "RunnerUp_2099-01-01-00-00-00_X.tcx"]) - reason = rt.check_marker(marker, processed).reason - assert "not in processed/" in reason - - -def test_marker_missing_a_processed_run_asks_for_a_reimport(tmp_path: Path) -> None: - processed = _processed(tmp_path) - marker = put_marker(tmp_path, _LISTED[:1]) - reason = rt.check_marker(marker, processed).reason - assert "re-run the app's import" in reason +def test_app_run_ends_lists_every_app_run_in_every_folder(tmp_path: Path) -> None: + """Test app run ends lists every app run in every folder.""" + _app_run(tmp_path, "a.tcx") + _app_run(tmp_path, "b.tcx", where="processed") + put_tcx(tmp_path, "r.tcx") + ends = rt.app_run_ends([tmp_path, tmp_path / "processed"]) + assert ends == [END_UNIX, END_UNIX]