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

Merge branch 'worktree-agent-acd53f670745f7d44'

Browse Source
This commit is contained in:
Krzysztof Rudnicki 2026-11-10 16:09:14 +01:00
commit b1ff91249b
20 changed files with 2026 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@ -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_<date>/` 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

11
install_systemd.sh
View File

@ -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.

2
remove_systemd.sh
View File

@ -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

15
runnerup-retire.service Normal file
View File

@ -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

18
runnerup-retire.timer Normal file
View File

@ -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

215
screen_locker/_runnerup_retire.py Normal file
View File

@ -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

181
screen_locker/_runnerup_retire_adb.py Normal file
View File

@ -0,0 +1,181 @@
"""The phone side of the RunnerUp retirement: find it, back it up, uninstall.
Every call carries ``-s <serial>`` 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)

62
screen_locker/_runnerup_retire_result.py Normal file
View File

@ -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

71
screen_locker/_runnerup_retire_state.py Normal file
View File

@ -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)

210
screen_locker/_runnerup_retire_tcx.py Normal file
View File

@ -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* (``<Author>`` or
``<Creator>`` 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*<Name>\s*"
+ re.escape(APP_SOURCE_NAME.encode())
+ rb"\s*</Name>"
)
# 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)

79
screen_locker/_runnerup_tcx.py
View File

@ -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:

99
screen_locker/runnerup_retire.py Normal file
View File

@ -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())

16
screen_locker/tests/_isolated_state.py
View File

@ -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",),
),
)

137
screen_locker/tests/_retire_helpers.py Normal file
View File

@ -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><Name>{creator}</Name></Creator>" if creator else ""
return (
f'<?xml version="1.0"?><TrainingCenterDatabase xmlns="{_NS}"><Activities>'
'<Activity Sport="Running"><Id>2026-11-01T10:00:00Z</Id>'
"<Lap><TotalTimeSeconds>1800</TotalTimeSeconds>"
"<DistanceMeters>5000</DistanceMeters><Track>"
"<Trackpoint><Time>2026-11-01T10:00:00Z</Time>"
"<DistanceMeters>0</DistanceMeters></Trackpoint>"
f"<Trackpoint><Time>{end}</Time><DistanceMeters>5000</DistanceMeters>"
f"</Trackpoint></Track></Lap>{creator_xml}</Activity></Activities>"
f'<Author xsi:type="Application_t" xmlns:xsi="x"><Name>{author}</Name>'
"</Author></TrainingCenterDatabase>"
)
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

216
screen_locker/tests/test_runnerup_retire.py Normal file
View File

@ -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

225
screen_locker/tests/test_runnerup_retire_adb.py Normal file
View File

@ -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")

141
screen_locker/tests/test_runnerup_retire_cli.py Normal file
View File

@ -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")

117
screen_locker/tests/test_runnerup_retire_marker.py Normal file
View File

@ -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

52
screen_locker/tests/test_runnerup_retire_state.py Normal file
View File

@ -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")

189
screen_locker/tests/test_runnerup_retire_tcx.py Normal file
View File

@ -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"<Notes>{APP}</Notes>")
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"<Author><Name>{APP}</Name></Author>")
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]