diff --git a/README.md b/README.md index 01b0d4d..6a2b0df 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,13 @@ so a run finished after 20:00 pushes the hour later without the phone being plugged in; the per-minute shutdown check then lifts the night lockdown on its own once the hour is past "now" (digital-wellbeing's check script). +**Retiring RunnerUp.** A week after the PC first credits a run recorded by the +workout app (TCX author `kuhy workout_app`), `runnerup-retire.timer` (daily, retries +until the phone is attached) backs `/sdcard/Documents/RunnerUp` up to +`~/data/cloud/RunnerUp/runnerup_phone_backup_/` and uninstalls `org.runnerup.free`, +once the app's `.workout_app_import.json` matches `processed/`; no app run, no uninstall. +`python3 -m screen_locker.runnerup_retire --status` / `--dry-run` change nothing. + ## MCP server (Claude Code integration) screen-locker exposes a **read-only** MCP server (`screen_locker._mcp`) so diff --git a/install_systemd.sh b/install_systemd.sh index c4ce5df..f13b812 100755 --- a/install_systemd.sh +++ b/install_systemd.sh @@ -17,6 +17,8 @@ BONUS_SERVICE_NAME="earner-bonus.service" BONUS_PATH_NAME="earner-bonus.path" RUNNERUP_WATCH_NAME="runnerup-watch.service" WORKOUT_POKE_NAME="workout-poke.service" +RETIRE_SERVICE_NAME="runnerup-retire.service" +RETIRE_TIMER_NAME="runnerup-retire.timer" # Runtime dependencies. screen_lock.py imports tkinter at module scope. On Arch # the tkinter MODULE ships inside the `python` package, but the shared library @@ -147,6 +149,12 @@ cp "$SCRIPT_DIR/$RUNNERUP_WATCH_NAME" "$USER_SERVICE_DIR/$RUNNERUP_WATCH_NAME" bash "$SCRIPT_DIR/scripts/workout_poke_keygen.sh" cp "$SCRIPT_DIR/$WORKOUT_POKE_NAME" "$USER_SERVICE_DIR/$WORKOUT_POKE_NAME" +# The RunnerUp retirement: a week after the workout app's first credited run +# it backs the old app's exports up and uninstalls it from the phone. A daily +# timer, because the phone is not always attached; a no-op until then. +cp "$SCRIPT_DIR/$RETIRE_SERVICE_NAME" "$USER_SERVICE_DIR/$RETIRE_SERVICE_NAME" +cp "$SCRIPT_DIR/$RETIRE_TIMER_NAME" "$USER_SERVICE_DIR/$RETIRE_TIMER_NAME" + # Update paths in the service file to use absolute paths REPO_ROOT="$SCRIPT_DIR" sed -i "s|WorkingDirectory=.*|WorkingDirectory=$REPO_ROOT|" "$USER_SERVICE_DIR/$SERVICE_NAME" @@ -179,6 +187,9 @@ systemctl --user restart "$RUNNERUP_WATCH_NAME" systemctl --user enable "$WORKOUT_POKE_NAME" systemctl --user restart "$WORKOUT_POKE_NAME" +# Enable the daily RunnerUp retirement retry +systemctl --user enable --now "$RETIRE_TIMER_NAME" + # Verify enforcement is actually armed. enable can silently no-op when systemd # breaks an ordering cycle by deleting the timer's job -- exactly what happened # on 2026-08-04 -- so a successful `enable` is NOT proof the locker will run. diff --git a/remove_systemd.sh b/remove_systemd.sh index b1902ea..afd2afa 100755 --- a/remove_systemd.sh +++ b/remove_systemd.sh @@ -17,6 +17,8 @@ UNITS=( "earner-bonus.service" "runnerup-watch.service" "workout-poke.service" + "runnerup-retire.timer" + "runnerup-retire.service" ) for unit in "${UNITS[@]}"; do diff --git a/runnerup-retire.service b/runnerup-retire.service new file mode 100644 index 0000000..9b8fba1 --- /dev/null +++ b/runnerup-retire.service @@ -0,0 +1,15 @@ +[Unit] +Description=Back up and uninstall the stock RunnerUp app a week after the workout app's first credited run + +[Service] +Type=oneshot +# screen_locker is pip-installed (editable), like workout-sync.service, so no +# WorkingDirectory/PYTHONPATH. Headless: no DISPLAY, no Tk, no lock screen. +# +# Exit 0 for "not due yet", "no app run credited yet" and "phone not attached" +# (the timer retries), non-zero when a human is needed: invalid import marker, +# adb missing, a failed backup or a refused uninstall. That shows up in +# `systemctl --user --failed`. Every skip says why in the journal: +# journalctl --user -u runnerup-retire +# State and the anchor: python3 -m screen_locker.runnerup_retire --status +ExecStart=/usr/bin/python3 -m screen_locker.runnerup_retire diff --git a/runnerup-retire.timer b/runnerup-retire.timer new file mode 100644 index 0000000..c853fa6 --- /dev/null +++ b/runnerup-retire.timer @@ -0,0 +1,18 @@ +[Unit] +Description=Retry the RunnerUp retirement daily until the phone has been cleaned up + +# No After=graphical-session.target: timers.target is ordered before it, and +# that After= closes a cycle systemd resolves by deleting this timer's job +# (see early-bird-workout-check.timer for the 2026-08-04 incident). + +[Timer] +# The phone is not always on USB, so a failed or skipped pass is simply tried +# again: shortly after boot and once a day. Most passes are an instant no-op +# (not due yet, or already done), so the timer stays installed. +OnBootSec=10min +OnCalendar=*-*-* 19:00:00 +# Persistent=true: a PC that was off at 19:00 still gets that day's pass. +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/screen_locker/_runnerup_retire.py b/screen_locker/_runnerup_retire.py new file mode 100644 index 0000000..9cf2159 --- /dev/null +++ b/screen_locker/_runnerup_retire.py @@ -0,0 +1,215 @@ +"""Retire the stock RunnerUp app a week after the PC first credits an app run. + +kuhy's workout_app records runs itself now. Once the PC has credited one of +them, RunnerUp is redundant: seven days later (``GRACE``) it is backed up and +uninstalled, but only if the app has imported the RunnerUp history (a marker +the app writes) and that import matches what ``processed/`` holds. With no +anchor (no app run ever credited) it never uninstalls. + +The phone is not always on USB, so ``runnerup-retire.timer`` retries daily +until this succeeds. Every outcome is a typed :class:`RetireResult` whose +reason is logged at the level it deserves: ``info`` for the normal "not due +yet", ``warning`` for a skip that needs attention, ``error`` for a failure. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import UTC, datetime, timedelta +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 app_run_ends, check_marker, find_anchor + +if TYPE_CHECKING: + from screen_locker._runnerup_retire_adb import PhoneLike + from screen_locker._runnerup_retire_state import RetireState + +_logger = logging.getLogger(__name__) + +GRACE = timedelta(days=7) +STATE_FILE = Path.home() / ".local" / "state" / "screen-locker" / "runnerup_retire.json" +WATCH_DIR = Path.home() / "data" / "cloud" / "RunnerUp" +LOG_FILE = Path(__file__).resolve().parent / "log.json" +MARKER_NAME = ".workout_app_import.json" + + +@dataclass(frozen=True) +class Paths: + """Every on-disk location, so tests (and ``--status``) can redirect them.""" + + state_file: Path + watch_dir: Path + log_file: Path + + @classmethod + def default(cls) -> Paths: + """The live locations (read at call time so tests can patch them).""" + return cls(STATE_FILE, WATCH_DIR, LOG_FILE) + + @property + def processed(self) -> Path: + """Where endurain-import files the RunnerUp exports it has handled.""" + return self.watch_dir / "processed" + + @property + def marker(self) -> Path: + """The app's historic-import marker.""" + return self.watch_dir / MARKER_NAME + + def anchor_dirs(self) -> tuple[Path, Path]: + """Where an app run's TCX can sit: the drop dir, then ``processed/``.""" + return (self.watch_dir, self.processed) + + +def derive_anchor(paths: Paths, state: RetireState) -> datetime | None: + """The stored anchor, else the first credit of an app run found on disk.""" + return state.anchor or find_anchor(paths.log_file, paths.anchor_dirs()) + + +def _finish(paths: Paths, state: RetireState, now: datetime, why: str) -> None: + save_state( + paths.state_file, replace(state, done_at=now.isoformat(), done_reason=why) + ) + + +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( + Outcome.ADB_MISSING, "adb is not installed (pacman -S android-tools)" + ) + try: + device = phone.find_device() + if device.serial is None: + return RetireResult( + Outcome.DEVICE_ABSENT, f"{device.reason}; retrying next run" + ) + return device.serial, phone.installed(device.serial) + except AdbError as exc: + _logger.exception("RunnerUp retirement: asking the phone failed") + return RetireResult( + Outcome.PHONE_ERROR, f"asking the phone failed: {exc}", logged=True + ) + + +def _backup( + 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(): + return state + final = paths.watch_dir / f"runnerup_phone_backup_{now:%Y-%m-%d}" + try: + made, files = phone.backup(serial, final) + except (AdbError, OSError) as exc: + _logger.exception("RunnerUp retirement: backup failed") + return RetireResult(Outcome.BACKUP_FAILED, f"backup failed: {exc}", logged=True) + _logger.info("Backed up %d RunnerUp TCX files to %s", files, made) + state = replace(state, backup_path=str(made)) + save_state(paths.state_file, state) + return state + + +def _retire( + 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) + if isinstance(probed, RetireResult): + return probed + serial, present = probed + if not present: + if not dry_run: + _finish(paths, state, now, f"{PACKAGE} was already absent") + return RetireResult(Outcome.ALREADY_GONE, f"{PACKAGE} is already absent") + if dry_run: + return RetireResult( + Outcome.DRY_RUN, "would back up and uninstall (no changes made)" + ) + backed = _backup(paths, phone, state, serial, now) + if isinstance(backed, RetireResult): + return backed + try: + phone.uninstall(serial) + except AdbError as exc: + _logger.exception("RunnerUp retirement: uninstall failed") + return RetireResult( + Outcome.UNINSTALL_FAILED, f"uninstall failed: {exc}", logged=True + ) + _finish(paths, backed, now, f"uninstalled after backup to {backed.backup_path}") + return RetireResult( + Outcome.UNINSTALLED, + f"{PACKAGE} uninstalled; backup in {backed.backup_path}", + ) + + +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: + """The no-phone checks; a state back means "due, marker good: go on".""" + if state.done_at: + return RetireResult( + Outcome.ALREADY_DONE, f"done {state.done_at}: {state.done_reason}" + ) + anchor = derive_anchor(paths, state) + if anchor is None: + return _no_anchor(paths) + if not state.anchor_at: + state = replace(state, anchor_at=anchor.isoformat()) + if not dry_run: + save_state(paths.state_file, state) + due = anchor + GRACE + if now < due: + return RetireResult( + Outcome.NOT_DUE, f"due {due:%Y-%m-%d %H:%M} (anchor {anchor:%Y-%m-%d})" + ) + marker = check_marker(paths.marker, paths.processed) + if not marker.ok: + return RetireResult( + Outcome.MARKER_INVALID, f"{marker.reason}; NOT uninstalling" + ) + return state + + +def run_retire( + 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) + gated = _gate(paths, load_state(paths.state_file), when, dry_run=dry_run) + if isinstance(gated, RetireResult): + result = gated + else: + result = _retire(paths, phone, gated, when, dry_run=dry_run) + if not result.logged: + _logger.log( + LEVEL.get(result.outcome, logging.INFO), + "RunnerUp retirement: %s: %s", + result.outcome.value, + result.reason, + ) + return result diff --git a/screen_locker/_runnerup_retire_adb.py b/screen_locker/_runnerup_retire_adb.py new file mode 100644 index 0000000..2157a1c --- /dev/null +++ b/screen_locker/_runnerup_retire_adb.py @@ -0,0 +1,181 @@ +"""The phone side of the RunnerUp retirement: find it, back it up, uninstall. + +Every call carries ``-s `` and goes through one ``run`` seam, so tests +never touch a device. adb's exit code is not trusted for the uninstall: it +counts only when the output says ``Success`` AND ``pm list packages`` no +longer lists the package. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging +from pathlib import Path +import shutil +import subprocess +from typing import TYPE_CHECKING, Any, NoReturn, Protocol + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger = logging.getLogger(__name__) + +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): + """An adb call failed; the message says what and why.""" + + +def _fail(message: str) -> NoReturn: + raise AdbError(message) + + +@dataclass(frozen=True) +class Device: + """Outcome of looking for the phone: the adb serial to use, or why not.""" + + serial: str | None + 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.""" + + def __init__( + self, + serial: str = PHONE_SERIAL, + *, + which: Callable[[str], str | None] = shutil.which, + run: Callable[..., Any] = subprocess.run, + ) -> None: + """Pin the phone's serial; ``which``/``run`` are the test seams.""" + self.serial = serial + self._which = which + self._run = run + + def available(self) -> bool: + """Whether an ``adb`` binary is on PATH.""" + return self._which("adb") is not None + + 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=timeout, check=False + ) + except (OSError, subprocess.TimeoutExpired) as exc: + msg = f"{' '.join(cmd)} did not run: {exc}" + raise AdbError(msg) from exc + if done.returncode: + detail = (done.stdout + done.stderr).strip() + _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 not self._is_phone(serial): + continue + if state.strip() == "device": + return Device(serial, "attached") + states.append(state.strip()) + if states: + 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, *, 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]: + """Pull the phone's export folder next to ``final``; return (dir, tcx count). + + Pulled into a staging dir and renamed only once every phone ``.tcx`` + is on disk, so a half backup never looks like a finished one. The + result is a subfolder, never top-level: a top-level ``*.tcx`` would be + credited by runnerup-watch. + """ + listing = self._adb("-s", dev, "shell", "ls", REMOTE_DIR) + wanted = sum(1 for n in listing.split() if n.endswith(".tcx")) + 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), 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: + _fail(f"backup pulled {got} of {wanted} TCX files into {staging}") + target, n = final, 1 + while target.exists(): + n += 1 + target = final.with_name(f"{final.name}-{n}") + pulled.rename(target) + staging.rmdir() + return target, got + + def uninstall(self, dev: str) -> None: + """Uninstall the package and prove it is gone, or raise ``AdbError``.""" + try: + out = self._adb("-s", dev, "uninstall", PACKAGE) + except AdbError as exc: + # adb exits 1 with "Failure [DELETE_FAILED_...]": keep that reason + # (a device-owner policy can block it) in the error. + msg = f"uninstall refused: {exc}" + raise AdbError(msg) from exc + if "Success" not in out: + _fail(f"uninstall did not report Success: {out.strip()!r}") + 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 new file mode 100644 index 0000000..c16b151 --- /dev/null +++ b/screen_locker/_runnerup_retire_state.py @@ -0,0 +1,71 @@ +"""The retirement's own memory: the anchor, the backup, the done stamp. + +One small JSON file (path: ``_runnerup_retire.STATE_FILE``). Written +atomically and only by a real run (``--status`` and ``--dry-run`` never write), +so the file is also the proof that a run happened. The anchor is stored once +and never moved: a later, earlier-looking credit must not push the deadline. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, fields +from datetime import datetime +import json +import logging +import os +from pathlib import Path +import tempfile + +_logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class RetireState: + """What a previous run recorded (all ISO strings, empty until known).""" + + anchor_at: str = "" + backup_path: str = "" + done_at: str = "" + done_reason: str = "" + + @property + def anchor(self) -> datetime | None: + """The stored anchor as an aware datetime, or None.""" + return datetime.fromisoformat(self.anchor_at) if self.anchor_at else None + + +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 (OSError, ValueError) as exc: + _logger.warning( + "Retire state %s is unreadable (%s): starting from scratch, so the " + "anchor is re-derived and the uninstall re-checked", + path, + exc, + ) + return RetireState() + if not isinstance(raw, dict): + _logger.warning("Retire state %s is not a JSON object: ignored", path) + return RetireState() + 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 + except ValueError as exc: + _logger.warning("Retire state %s holds a bad anchor (%s): ignored", path, exc) + return RetireState() + return state + + +def save_state(path: Path, state: RetireState) -> None: + """Atomically replace the state file (creates the directory on first write).""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".retire-", suffix=".tmp") + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(asdict(state), handle, indent=2) + Path(tmp).replace(path) diff --git a/screen_locker/_runnerup_retire_tcx.py b/screen_locker/_runnerup_retire_tcx.py new file mode 100644 index 0000000..d9d51a2 --- /dev/null +++ b/screen_locker/_runnerup_retire_tcx.py @@ -0,0 +1,210 @@ +"""What the RunnerUp retirement reads off disk: app runs, the marker, the anchor. + +Split from :mod:`screen_locker._runnerup_retire` (250-line cap). Everything +here is read-only; nothing touches the phone. + +Telling the two sources apart is done on the TCX *content* (```` or +```` named ``kuhy workout_app``, the app's ``kRunSourceName``), never +on the file name: the app may name its files however it likes. + +The anchor is "when the PC first CREDITED an app run". The credit chokepoint +(``log.json``) stores a run's TCX end as ``completed_at`` on its +``runnerup_verified`` entry and stamps the entry itself with the moment the PC +wrote it. So the anchor is the ``timestamp`` of the entry whose +``completed_at`` equals an app TCX's last trackpoint. Joining on the end time +(not the day) is what makes a same-day RunnerUp run, which fills the day's one +credit slot first, correctly yield no anchor for the app run. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +import json +import logging +import re +from typing import TYPE_CHECKING, Any + +from screen_locker._log_io import load_workout_log +from screen_locker._runnerup_tcx import parse_tcx_file + +if TYPE_CHECKING: + from collections.abc import Iterable + from pathlib import Path + +_logger = logging.getLogger(__name__) + +APP_SOURCE_NAME = "kuhy workout_app" +_APP_RE = re.compile( + rb"<(?:Author|Creator)\b[^>]*>\s*\s*" + + re.escape(APP_SOURCE_NAME.encode()) + + rb"\s*" +) +# The WebDAV copy of a RunnerUp export may carry the device name +# (RunnerUp_Pixel_6a_2026-...) that the phone's own copy lacks. +_DEVICE_PREFIX = re.compile(r"^RunnerUp_(?:[A-Za-z0-9]+_)*?(?=\d{4}-\d{2}-\d{2}-)") +_END_TOLERANCE_S = 1.0 + + +def run_key(name: str) -> str: + """A run's identity: its file name without any device-name prefix.""" + return _DEVICE_PREFIX.sub("RunnerUp_", name) + + +def list_tcx(directory: Path) -> list[Path]: + """Finished ``*.tcx`` files directly in ``directory`` (dotfiles are partial).""" + if not directory.is_dir(): + return [] + return sorted( + p for p in directory.glob("*.tcx") if p.is_file() and not p.name.startswith(".") + ) + + +def is_app_tcx(path: Path) -> bool: + """Whether the TCX names the new app as its author or creator.""" + try: + data = path.read_bytes() + except OSError as exc: + _logger.warning( + "Cannot read %s (%s): counting it as a RunnerUp file", path, exc + ) + return False + return bool(_APP_RE.search(data)) + + +def _ended_at(path: Path) -> float | None: + """The run's end exactly as the credit path computed it (same parser).""" + data = parse_tcx_file(str(path)) + if data is None or data.get("ended_at") is None: + _logger.warning("App TCX %s has no usable end time: it cannot anchor", path) + return None + return float(data["ended_at"]) + + +def _credited_at(entry: dict[str, Any]) -> datetime | None: + try: + stamp = datetime.fromisoformat(str(entry["timestamp"])) + except (KeyError, ValueError) as exc: + _logger.warning("Credit entry has no usable timestamp (%r): skipped", exc) + return None + return stamp if stamp.tzinfo else stamp.astimezone() + + +def _as_unix(raw: object) -> float | None: + """A unix time from a number or numeric string (the log stores both).""" + if isinstance(raw, int | float) and not isinstance(raw, bool): + return float(raw) + if isinstance(raw, str) and re.fullmatch(r"\d+(\.\d+)?", raw): + return float(raw) + return None + + +def _credit_ends(log_file: Path) -> list[tuple[float, datetime]]: + """``(TCX end, PC credit time)`` of every credited run that recorded its end.""" + found: list[tuple[float, datetime]] = [] + for entries in load_workout_log(log_file).values(): + for entry in entries: + data = entry.get("workout_data") or {} + if data.get("type") != "runnerup_verified": + continue + end = _as_unix(data.get("completed_at")) + if end is None: + continue # an older entry that never stored its run's end + when = _credited_at(entry) + if when is not None: + found.append((end, when)) + return found + + +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 + ] + + +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) + if any(abs(run_end - end) <= _END_TOLERANCE_S for end in ends) + ] + return min(credited) if credited else None + + +@dataclass(frozen=True) +class MarkerCheck: + """Verdict on the app's historic-import marker.""" + + ok: bool + reason: str + + +def _bad(reason: str) -> MarkerCheck: + return MarkerCheck(ok=False, reason=reason) + + +def _marker_names(marker: object) -> tuple[list[str] | None, str]: + """The marker's ``tcx_names`` if its shape is valid, else a reason.""" + if not isinstance(marker, dict): + return None, "marker is not a JSON object" + names, count = marker.get("tcx_names"), marker.get("count") + if not isinstance(marker.get("completed_at"), str) or not marker["completed_at"]: + return None, "marker has no completed_at" + if not isinstance(names, list) or not all(isinstance(n, str) for n in names): + return None, "marker tcx_names is not a list of names" + if isinstance(count, bool) or not isinstance(count, int) or count != len(names): + return None, f"marker count {count!r} != {len(names)} listed names" + if any("/" in n or n.startswith(".") for n in names): + return None, "marker tcx_names must be bare file names" + return names, "" + + +def _read_marker(marker_file: Path) -> tuple[object, str]: + """The parsed marker JSON, or ``(None, why it could not be read)``.""" + if not marker_file.exists(): + why = f"{marker_file} is missing: the app has not imported yet" + 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 + + +def _compare(names: list[str], processed: Path) -> MarkerCheck: + listed = {run_key(n) for n in names} + if len(listed) != len(names): + return _bad("marker lists the same run more than once") + have = {run_key(p.name) for p in list_tcx(processed) if not is_app_tcx(p)} + if absent := sorted(listed - have): + return _bad(f"{len(absent)} listed run(s) not in processed/: {absent[:3]}") + if unlisted := sorted(have - listed): + return _bad( + f"{len(unlisted)} RunnerUp run(s) in processed/ are not in the " + f"marker ({unlisted[:3]}): re-run the app's import" + ) + return MarkerCheck(ok=True, reason=f"{len(listed)} runs match") + + +def check_marker(marker_file: Path, processed: Path) -> MarkerCheck: + """Validate the app's import marker against what ``processed/`` really holds. + + Valid means: the file parses, ``count`` equals the number of listed names, + and the listed runs are exactly the RunnerUp-recorded (non-app) runs in + ``processed/``. A run is counted once however many copies exist (the + device-prefixed WebDAV twin and the phone-named copy are one run). + """ + marker, why = _read_marker(marker_file) + if why: + return _bad(why) + names, reason = _marker_names(marker) + return _compare(names, processed) if names is not None else _bad(reason) diff --git a/screen_locker/_runnerup_tcx.py b/screen_locker/_runnerup_tcx.py index 0727478..bca4539 100644 --- a/screen_locker/_runnerup_tcx.py +++ b/screen_locker/_runnerup_tcx.py @@ -65,48 +65,53 @@ class RunnerUpTcxMixin: shutil.rmtree(tmp_dir, ignore_errors=True) def _parse_tcx(self, tcx_path: str) -> dict[str, Any] | None: - """Parse a local TCX file and return activity summary dict. + """Parse a local TCX file (see :func:`parse_tcx_file`).""" + return parse_tcx_file(tcx_path) - Sums ``TotalTimeSeconds`` and ``DistanceMeters`` across all Laps so - multi-segment runs (pause/resume) are counted in full. - """ - try: - tree = ET.parse(tcx_path) - except ET.ParseError as exc: - _logger.warning( - "TCX file %s is not valid XML (%s) — the run it describes CANNOT " - "be verified and will not count", - tcx_path, - exc, - ) - return None - root = tree.getroot() - activity = root.find(f".//{{{_TCX_NS}}}Activity") - if activity is None: - _logger.info("No Activity element in TCX file") - return None +def parse_tcx_file(tcx_path: str) -> dict[str, Any] | None: + """Parse a local TCX file and return activity summary dict. - sport_str = activity.get("Sport", "") - sport_int = _TCX_SPORT_TO_INT.get(sport_str, -1) + Sums ``TotalTimeSeconds`` and ``DistanceMeters`` across all Laps so + multi-segment runs (pause/resume) are counted in full. + """ + try: + tree = ET.parse(tcx_path) + except ET.ParseError as exc: + _logger.warning( + "TCX file %s is not valid XML (%s) — the run it describes CANNOT " + "be verified and will not count", + tcx_path, + exc, + ) + return None - total_seconds = 0.0 - total_distance = 0.0 - for lap in activity.findall(f"{{{_TCX_NS}}}Lap"): - t_elem = lap.find(f"{{{_TCX_NS}}}TotalTimeSeconds") - d_elem = lap.find(f"{{{_TCX_NS}}}DistanceMeters") - if t_elem is not None and t_elem.text: - total_seconds += float(t_elem.text) - if d_elem is not None and d_elem.text: - total_distance += float(d_elem.text) + root = tree.getroot() + activity = root.find(f".//{{{_TCX_NS}}}Activity") + if activity is None: + _logger.info("No Activity element in TCX file") + return None - return { - "sport": sport_int, - "duration_seconds": int(total_seconds), - "distance_m": total_distance, - "walking": is_walking_export(Path(tcx_path).name, sport_str), - **track_summary(activity), - } + sport_str = activity.get("Sport", "") + sport_int = _TCX_SPORT_TO_INT.get(sport_str, -1) + + total_seconds = 0.0 + total_distance = 0.0 + for lap in activity.findall(f"{{{_TCX_NS}}}Lap"): + t_elem = lap.find(f"{{{_TCX_NS}}}TotalTimeSeconds") + d_elem = lap.find(f"{{{_TCX_NS}}}DistanceMeters") + if t_elem is not None and t_elem.text: + total_seconds += float(t_elem.text) + if d_elem is not None and d_elem.text: + total_distance += float(d_elem.text) + + return { + "sport": sport_int, + "duration_seconds": int(total_seconds), + "distance_m": total_distance, + "walking": is_walking_export(Path(tcx_path).name, sport_str), + **track_summary(activity), + } def _mark_walk(data: dict[str, Any] | None, name: str) -> dict[str, Any] | None: diff --git a/screen_locker/runnerup_retire.py b/screen_locker/runnerup_retire.py new file mode 100644 index 0000000..14cba01 --- /dev/null +++ b/screen_locker/runnerup_retire.py @@ -0,0 +1,99 @@ +"""``python3 -m screen_locker.runnerup_retire``: uninstall RunnerUp, a week on. + +What ``runnerup-retire.service`` runs. ``--status`` prints anchor, due date, +marker validity and (with ``--query-phone``) whether RunnerUp is installed; +it never writes and never talks to the phone unless asked. ``--dry-run`` runs +every check, including the read-only adb queries, but writes nothing, pulls +nothing and uninstalls nothing. The exit code is non-zero only for outcomes +that need a human (bad marker, adb missing, failed backup or uninstall). +""" + +from __future__ import annotations + +import argparse +from datetime import UTC, datetime +import logging +import sys +from typing import TYPE_CHECKING + +from screen_locker._runnerup_retire import ( + GRACE, + Paths, + derive_anchor, + run_retire, +) +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: PhoneLike, *, query: bool) -> str: + if not query: + return "installed: not queried (add --query-phone)" + try: + device = phone.find_device() + if device.serial is None: + return f"installed: unknown ({device.reason})" + return f"installed: {'yes' if phone.installed(device.serial) else 'no'}" + except AdbError as exc: + _logger.warning("--status could not ask the phone: %s", exc) + return f"installed: unknown ({exc})" + + +def status_lines( + 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) + anchor = derive_anchor(paths, state) + if anchor is None: + anchor_line, due_line = "anchor: none (no app run credited yet)", "due: never" + else: + origin = "stored" if state.anchor_at else "derived, not yet stored" + due = anchor + GRACE + days = (due - now).total_seconds() / 86400 + anchor_line = f"anchor: {anchor.isoformat()} ({origin})" + due_line = f"due: {due:%Y-%m-%d %H:%M} ({days:+.1f} days)" + marker = check_marker(paths.marker, paths.processed) + return [ + anchor_line, + due_line, + f"marker: {'valid' if marker.ok else 'INVALID'} ({marker.reason})", + f"done: {state.done_at or 'no'}", + _installed_line(phone, query=query_phone), + ] + + +def main(argv: list[str] | None = None) -> int: + """CLI entry; returns the process exit code.""" + parser = argparse.ArgumentParser(description=__doc__.split("\n\n", maxsplit=1)[0]) + parser.add_argument( + "--status", action="store_true", help="print state, change nothing" + ) + parser.add_argument( + "--dry-run", action="store_true", help="check everything, change nothing" + ) + parser.add_argument( + "--query-phone", action="store_true", help="with --status: ask adb" + ) + parser.add_argument("--serial", default=PHONE_SERIAL, help="phone adb serial") + args = parser.parse_args(argv) + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + paths, phone = Paths.default(), Phone(args.serial) + if args.status: + now = datetime.now(UTC) + lines = status_lines(paths, phone, now, query_phone=args.query_phone) + sys.stdout.write("\n".join(lines) + "\n") + return 0 + return 0 if run_retire(paths, phone, dry_run=args.dry_run).ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/screen_locker/tests/_isolated_state.py b/screen_locker/tests/_isolated_state.py index 79995ed..8284313 100644 --- a/screen_locker/tests/_isolated_state.py +++ b/screen_locker/tests/_isolated_state.py @@ -105,4 +105,20 @@ ISOLATED_STATE: tuple[tuple[str, tuple[str, ...]], ...] = ( "workday_penalty.json", ("_constants.WORKDAY_PENALTY_FILE", "_workday_penalty.WORKDAY_PENALTY_FILE"), ), + # RunnerUp retirement: the anchor/done stamp, the dufs drop directory it + # reads the marker and TCX files from (and writes the phone backup into), + # and the credit log the anchor is derived from. A test must never see, or + # create files under, the user's real RunnerUp folder. + ( + "runnerup_retire.json", + ("_runnerup_retire.STATE_FILE",), + ), + ( + "RunnerUp", + ("_runnerup_retire.WATCH_DIR",), + ), + ( + "retire_log.json", + ("_runnerup_retire.LOG_FILE",), + ), ) diff --git a/screen_locker/tests/_retire_helpers.py b/screen_locker/tests/_retire_helpers.py new file mode 100644 index 0000000..883dd45 --- /dev/null +++ b/screen_locker/tests/_retire_helpers.py @@ -0,0 +1,137 @@ +"""Builders for the RunnerUp retirement tests: TCX files, marker, credit log.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +import json +from typing import TYPE_CHECKING, Any + +from screen_locker._runnerup_retire import Paths +from screen_locker._runnerup_retire_adb import AdbError, Device + +if TYPE_CHECKING: + from pathlib import Path + +_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" +APP = "kuhy workout_app" +# 2026-11-01T10:30:00Z, the end of every TCX built by default. +END_UNIX = 1793529000.0 + + +def tcx_text( + *, author: str = "RunnerUp", creator: str = "", end: str = "2026-11-01T10:30:00Z" +) -> str: + """A tiny TCX: two trackpoints (the last is the run's end) and an author.""" + creator_xml = f"{creator}" if creator else "" + return ( + f'' + '2026-11-01T10:00:00Z' + "1800" + "5000" + "" + "0" + f"5000" + f"{creator_xml}" + f'{author}' + "" + ) + + +def put_tcx(folder: Path, name: str, **kwargs: str) -> Path: + """Write a TCX named ``name`` into ``folder`` (created if needed).""" + folder.mkdir(parents=True, exist_ok=True) + path = folder / name + path.write_text(tcx_text(**kwargs), encoding="utf-8") + return path + + +def put_marker(watch: Path, names: list[str], count: int | None = None) -> Path: + """Write the app's import marker into the drop folder.""" + watch.mkdir(parents=True, exist_ok=True) + marker = watch / ".workout_app_import.json" + body = { + "completed_at": "2026-11-02T08:00:00Z", + "count": len(names) if count is None else count, + "tcx_names": names, + } + marker.write_text(json.dumps(body), encoding="utf-8") + return marker + + +def put_log(log_file: Path, entries: dict[str, list[dict[str, Any]]]) -> None: + """Write a day-keyed credit log.""" + log_file.write_text(json.dumps(entries), encoding="utf-8") + + +def credit( + timestamp: str, completed_at: object = None, kind: str = "runnerup_verified" +) -> dict[str, Any]: + """One log entry as the credit chokepoint writes it.""" + data: dict[str, Any] = {"type": kind} + if completed_at is not None: + data["completed_at"] = completed_at + return {"timestamp": timestamp, "workout_data": data} + + +ANCHOR = datetime(2026, 11, 1, 10, 45, tzinfo=UTC) +DUE = ANCHOR + timedelta(days=7) +RUN = "RunnerUp_2026-08-12-17-42-11_Running.tcx" + + +@dataclass +class FakePhone: + """Records what the retirement asked of the phone.""" + + adb: bool = True + device: Device = field(default_factory=lambda: Device("S", "attached")) + present: bool = True + ask_error: AdbError | 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, *, 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 + final.mkdir(parents=True) + return final, 3 + + def uninstall(self, dev: str) -> None: + """Uninstall.""" + self.calls.append("uninstall") + if self.uninstall_error: + raise self.uninstall_error + + +def world(tmp_path: Path, *, anchored: bool = True, marker: bool = True) -> Paths: + """A drop folder with one RunnerUp run, one app run, and the credit log.""" + paths = Paths( + tmp_path / "state" / "s.json", tmp_path / "drop", tmp_path / "log.json" + ) + put_tcx(paths.processed, RUN) + put_tcx(paths.processed, "app.tcx", author=APP) + put_log( + paths.log_file, + {"d": [credit(ANCHOR.isoformat(), END_UNIX)]} if anchored else {}, + ) + if marker: + put_marker(paths.watch_dir, [RUN]) + return paths diff --git a/screen_locker/tests/test_runnerup_retire.py b/screen_locker/tests/test_runnerup_retire.py new file mode 100644 index 0000000..82f69b0 --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire.py @@ -0,0 +1,216 @@ +"""RunnerUp retirement: when it fires, what it refuses, and what it records.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +import logging +from typing import TYPE_CHECKING + +import pytest + +from screen_locker import _runnerup_retire as rr +from screen_locker._runnerup_retire import Outcome, Paths, run_retire +from screen_locker._runnerup_retire_adb import AdbError, Device +from screen_locker._runnerup_retire_state import RetireState, load_state, save_state +from screen_locker.tests._retire_helpers import ( + ANCHOR, + DUE, + RUN, + FakePhone, + put_marker, + world, +) + +if TYPE_CHECKING: + from pathlib import Path + + +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 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)) + assert result.outcome is Outcome.NOT_DUE + assert "2026-11-08 10:45" in result.reason + assert load_state(paths.state_file).anchor == ANCHOR + assert [r.levelno for r in caplog.records] == [logging.INFO] + + +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 + + +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) + assert result.outcome is Outcome.DRY_RUN + assert phone.calls == ["installed S"] + assert not paths.state_file.exists() + gone = go(paths, FakePhone(present=False), dry_run=True) + assert gone.outcome is Outcome.ALREADY_GONE + assert not paths.state_file.exists() + + +@pytest.mark.parametrize("variant", ["missing", "mismatch"]) +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) + phone = FakePhone() + with caplog.at_level(logging.WARNING): + result = go(paths, phone) + assert result.outcome is Outcome.MARKER_INVALID + assert not result.ok + assert "NOT uninstalling" in result.reason + 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 + + +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 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) + assert result.outcome is Outcome.PHONE_ERROR + assert not result.ok + assert len(caplog.records) == 1 # logged once, where it was caught + + +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 + assert load_state(paths.state_file).done_at + assert go(paths, FakePhone()).outcome is Outcome.ALREADY_DONE + + +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) + assert result.outcome is Outcome.UNINSTALLED + assert phone.calls == ["installed S", "backup", "uninstall"] + backup = paths.watch_dir / "runnerup_phone_backup_2026-11-08" + assert backup.is_dir() + 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 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: + """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 + assert not result.ok + assert "uninstall" not in phone.calls + assert str(error) in caplog.text + + +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) + assert result.outcome is Outcome.UNINSTALL_FAILED + assert not result.ok + assert not load_state(paths.state_file).done_at + retry = FakePhone() + assert go(paths, retry, DUE + timedelta(days=1)).outcome is Outcome.UNINSTALLED + assert retry.calls == ["installed S", "uninstall"] # no second backup + + +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 + assert paths.processed == rr.WATCH_DIR / "processed" + assert paths.anchor_dirs() == (rr.WATCH_DIR, rr.WATCH_DIR / "processed") + + +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 new file mode 100644 index 0000000..65625f0 --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire_adb.py @@ -0,0 +1,225 @@ +"""RunnerUp retirement: the adb wrapper and the state file (no real adb).""" + +from __future__ import annotations + +from pathlib import Path +import subprocess +from types import SimpleNamespace +from typing import Any + +import pytest + +from screen_locker import _runnerup_retire_adb as ra + +SERIAL = ra.PHONE_SERIAL +PKG_LINE = f"package:{ra.PACKAGE}\n" + + +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): + if isinstance(reply, BaseException): + raise reply + code, out, err = reply(cmd) if callable(reply) else reply + return SimpleNamespace(returncode=code, stdout=out, stderr=err) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + +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 + + +@pytest.mark.parametrize( + "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() + + +@pytest.mark.parametrize( + ("listing", "serial", "reason"), + [ + (f"List of devices attached\n{SERIAL}\tdevice\n", SERIAL, "attached"), + ( + f"List of devices attached\nadb-{SERIAL}-x._adb-tls-connect._tcp\tdevice\n", + f"adb-{SERIAL}-x._adb-tls-connect._tcp", + "attached", + ), + (f"List of devices attached\n{SERIAL}\tunauthorized\n", None, "unauthorized"), + ("List of devices attached\nOTHER\tdevice\n", None, "not attached"), + ("List of devices attached\n", None, "not attached"), + ], +) +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"] + other = FakeAdb({"pm list": (0, f"package:{ra.PACKAGE}.pro\n", "")}) + assert phone(other).installed("dev") is False + + +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): + (folder / f"r{i}.tcx").write_text("x") + return 0, "", "" + + return make + + +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) + assert (target, count) == (final, 2) + assert len(list(final.glob("*.tcx"))) == 2 + assert not (tmp_path / "backup_2026.partial").exists() + + +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 + fake = FakeAdb({"ls": (0, "a.tcx\n", ""), "pull": _pull_creates(1)}) + target, _ = phone(fake).backup("dev", final) + assert target == tmp_path / "backup-2" + + +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") + assert not (tmp_path / "b").exists() + + +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"): + phone(fake).uninstall("dev") + + +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") diff --git a/screen_locker/tests/test_runnerup_retire_cli.py b/screen_locker/tests/test_runnerup_retire_cli.py new file mode 100644 index 0000000..b1f589f --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire_cli.py @@ -0,0 +1,141 @@ +"""RunnerUp retirement CLI: ``--status`` is read-only, exit codes follow outcomes.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING + +import pytest + +from screen_locker import runnerup_retire as cli +from screen_locker._runnerup_retire import Outcome, Paths, RetireResult +from screen_locker._runnerup_retire_adb import AdbError, Device +from screen_locker._runnerup_retire_state import RetireState, save_state +from screen_locker.tests._retire_helpers import put_marker +from screen_locker.tests.test_runnerup_retire import ANCHOR, DUE, RUN, FakePhone, world + +if TYPE_CHECKING: + from pathlib import Path + + +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) + assert lines[0].startswith("anchor: none") + assert lines[1] == "due: never" + assert "INVALID" in lines[2] + assert lines[3] == "done: no" + assert "not queried" in lines[4] + + +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)) + assert _tree(tmp_path) == before + assert not paths.state_file.parent.exists() + assert "derived, not yet stored" in lines[0] + assert "2026-11-08 10:45 (+1.0 days)" in lines[1] + assert lines[2] == "marker: valid (1 runs match)" + + +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) + assert "(stored)" in lines[0] + assert lines[3] == "done: 2026-11-09T00:00:00+00:00" + + +@pytest.mark.parametrize( + ("phone", "expected"), + [ + (FakePhone(present=True), "installed: yes"), + (FakePhone(present=False), "installed: no"), + ( + FakePhone(device=Device(None, "phone S is not attached")), + "installed: unknown (phone S is not attached)", + ), + (FakePhone(ask_error=AdbError("boom")), "installed: unknown (boom)"), + ], +) +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) + assert lines[4] == expected + + +def test_main_status_prints_and_exits_zero( + capsys: pytest.CaptureFixture[str], +) -> None: + """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 + + +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 not cli.main(["--status", "--query-phone", "--serial", "XYZ"]) + assert seen == ["XYZ"] + assert "installed: yes" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("outcome", "code"), + [(Outcome.NOT_DUE, 0), (Outcome.DEVICE_ABSENT, 0), (Outcome.UNINSTALL_FAILED, 1)], +) +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") + + monkeypatch.setattr(cli, "run_retire", fake_run) + assert cli.main(["--dry-run"]) == code + assert calls == [True] + + +def test_main_with_nothing_to_retire_is_a_clean_noop() -> None: + """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") 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 new file mode 100644 index 0000000..171611c --- /dev/null +++ b/screen_locker/tests/test_runnerup_retire_tcx.py @@ -0,0 +1,189 @@ +"""RunnerUp retirement: app-run detection, anchor derivation, marker validation.""" + +from __future__ import annotations + +from datetime import UTC, datetime +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, + END_UNIX, + credit, + put_log, + put_tcx, +) + +if TYPE_CHECKING: + from pathlib import Path + +_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 + assert rt.run_key(plain) == plain + + +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") + (tmp_path / "a.gpx").write_text("x") + (tmp_path / "dir.tcx").mkdir() + assert [p.name for p in rt.list_tcx(tmp_path)] == ["a.tcx"] + + +@pytest.mark.parametrize( + ("kwargs", "expected"), + [ + ({"author": APP}, True), + ({"creator": APP}, True), + ({"author": "RunnerUp"}, False), + ({"author": f"{APP} fork"}, False), + ], +) +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 + + +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( + log, + { + "2026-11-01": [credit(_AT, END_UNIX)], + "2026-11-02": [credit("2026-11-02T09:00:00+00:00", "1700000000.0")], + }, + ) + found = rt.find_anchor(log, [tmp_path, tmp_path / "processed"]) + assert found == datetime(2026, 11, 1, 10, 45, tzinfo=UTC) + + +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( + log, + { + "a": [credit("2026-11-03T00:00:00+00:00", str(END_UNIX))], + "b": [credit("2026-11-02T00:00:00+00:00", END_UNIX + 0.4)], + }, + ) + found = rt.find_anchor(log, [tmp_path]) + assert found == datetime(2026, 11, 2, tzinfo=UTC) + + +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)]}) + assert rt.find_anchor(log, [tmp_path]) is None + + +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)]}) + assert rt.find_anchor(log, [tmp_path]) is 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)]}) + with caplog.at_level(logging.WARNING): + assert rt.find_anchor(log, [tmp_path]) is None + assert "cannot anchor" in caplog.text + + +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( + log, + { + "d": [ + credit(_AT), # an older entry without completed_at + credit(_AT, True), + credit(_AT, "soon"), + credit(_AT, END_UNIX, kind="manual_workout"), + { + "timestamp": "garbage", + "workout_data": { + "type": "runnerup_verified", + "completed_at": END_UNIX, + }, + }, + {"workout_data": None}, + ] + }, + ) + with caplog.at_level(logging.WARNING): + assert rt.find_anchor(log, [tmp_path]) is None + assert "no usable timestamp" in caplog.text + + +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)]}) + found = rt.find_anchor(log, [tmp_path]) + assert found is not None + assert found.tzinfo is not None + + +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]