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

Add RunnerUp retirement logic

Browse Source 
Anchor = PC credit time of an app-authored run (joined on TCX end
time in log.json), due 7 days later, gated on a validated import
marker, then adb backup + uninstall. Typed results, --status/--dry-run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZ9CKZsNxtD5Jr9bYD
This commit is contained in:
Krzysztof Rudnicki 2026-11-10 16:01:17 +01:00
parent 1020421371
commit b174339979
12 changed files with 1706 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

249
screen_locker/_runnerup_retire.py Normal file
View File

@ -0,0 +1,249 @@
"""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
from enum import StrEnum
import logging
from pathlib import Path
from typing import TYPE_CHECKING
from screen_locker._runnerup_retire_adb import PACKAGE, AdbError
from screen_locker._runnerup_retire_state import load_state, save_state
from screen_locker._runnerup_retire_tcx import check_marker, find_anchor
if TYPE_CHECKING:
from screen_locker._runnerup_retire_adb import Phone
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"
class Outcome(StrEnum):
"""How a run ended; the value is what the journal shows."""
UNINSTALLED = "uninstalled"
ALREADY_GONE = "already-gone"
ALREADY_DONE = "already-done"
NO_ANCHOR = "no-anchor"
NOT_DUE = "not-due"
DRY_RUN = "dry-run"
DEVICE_ABSENT = "device-absent"
MARKER_INVALID = "marker-invalid"
ADB_MISSING = "adb-missing"
PHONE_ERROR = "phone-error"
BACKUP_FAILED = "backup-failed"
UNINSTALL_FAILED = "uninstall-failed"
# Level per outcome; the failures also make the unit exit non-zero.
_LEVEL = {
Outcome.DEVICE_ABSENT: logging.WARNING,
Outcome.MARKER_INVALID: logging.WARNING,
Outcome.ADB_MISSING: logging.ERROR,
Outcome.PHONE_ERROR: logging.ERROR,
Outcome.BACKUP_FAILED: logging.ERROR,
Outcome.UNINSTALL_FAILED: logging.ERROR,
}
_NONZERO = frozenset(
{
Outcome.MARKER_INVALID,
Outcome.ADB_MISSING,
Outcome.PHONE_ERROR,
Outcome.BACKUP_FAILED,
Outcome.UNINSTALL_FAILED,
}
)
@dataclass(frozen=True)
class RetireResult:
"""What happened and why, in a sentence a human can act on."""
outcome: Outcome
reason: str
logged: bool = False # the failure was already logged where it was caught
@property
def ok(self) -> bool:
"""False for outcomes the unit should surface as failed."""
return self.outcome not in _NONZERO
@dataclass(frozen=True)
class Paths:
"""Every on-disk location, so tests (and ``--status``) can redirect them."""
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: Phone) -> 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: Phone, 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: Phone, 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 _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 RetireResult(
Outcome.NO_ANCHOR, "no app-recorded run credited yet; nothing to retire"
)
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: Phone, 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

137
screen_locker/_runnerup_retire_adb.py Normal file
View File

@ -0,0 +1,137 @@
"""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
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
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 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) -> str:
cmd = ["adb", *args]
try:
done = self._run(
cmd, capture_output=True, text=True, timeout=_ADB_TIMEOUT_S, 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 find_device(self) -> Device:
"""The attached phone, over USB or wifi (a wireless serial embeds it)."""
listing = self._adb("devices")
states: list[str] = []
for line in listing.splitlines()[1:]:
serial, _, state = line.partition("\t")
if self.serial not in serial:
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) -> bool:
"""Whether the package is on the phone (exact ``package:`` line)."""
out = self._adb("-s", dev, "shell", "pm", "list", "packages", 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))
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):
_fail(f"{PACKAGE} is still listed after uninstall")
_logger.info("Uninstalled %s from %s", PACKAGE, dev)

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
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."""
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return RetireState()
except (OSError, ValueError) as exc:
_logger.warning(
"Retire state %s is unreadable (%s): starting from scratch, so the "
"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()
fields = RetireState.__dataclass_fields__
clean = {k: v for k, v in raw.items() if k in fields 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)

204
screen_locker/_runnerup_retire_tcx.py Normal file
View File

@ -0,0 +1,204 @@
"""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 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 = [
end
for directory in dirs
for path in list_tcx(directory)
if is_app_tcx(path) and (end := _ended_at(path)) is not None
]
if not ends:
return None
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)``."""
try:
return json.loads(marker_file.read_text(encoding="utf-8")), ""
except FileNotFoundError:
why = f"{marker_file} is missing: the app has not imported yet"
except (OSError, ValueError) as exc:
why = f"{marker_file} is unreadable ({exc})"
_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:

95
screen_locker/runnerup_retire.py Normal file
View File

@ -0,0 +1,95 @@
"""``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 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
__all__ = ["main", "status_lines"]
_logger = logging.getLogger(__name__)
def _installed_line(phone: Phone, *, 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: Phone, 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")[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",),
),
)

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

@ -0,0 +1,132 @@
"""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: AdbError | OSError | None = None
uninstall_error: AdbError | None = None
calls: list[str] = field(default_factory=list)
def available(self) -> bool:
return self.adb
def find_device(self) -> Device:
if self.ask_error:
raise self.ask_error
return self.device
def installed(self, dev: str) -> bool:
self.calls.append(f"installed {dev}")
return self.present
def backup(self, dev: str, final: Path) -> tuple[Path, int]:
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:
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

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

@ -0,0 +1,181 @@
"""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, **kw: bool):
return run_retire(paths, phone, now, **kw) # type: ignore[arg-type]
def test_no_anchor_never_touches_anything(tmp_path: Path) -> None:
paths = world(tmp_path, anchored=False)
phone = FakePhone()
result = go(paths, phone, DUE + timedelta(days=900))
assert result.outcome is Outcome.NO_ANCHOR
assert result.ok
assert phone.calls == []
assert not paths.state_file.exists()
def test_not_due_persists_the_anchor_and_says_when(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
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:
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:
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:
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 phone.calls == []
assert caplog.records[-1].levelno == logging.WARNING
def test_adb_missing_is_a_failure(tmp_path: Path) -> None:
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:
phone = FakePhone(device=Device(None, "phone S is not attached"))
with caplog.at_level(logging.WARNING):
result = go(world(tmp_path), phone)
assert result.outcome is Outcome.DEVICE_ABSENT
assert result.ok
assert "not attached" in caplog.text
assert phone.calls == []
def test_adb_error_while_asking_is_a_logged_failure(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
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:
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:
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 list(paths.watch_dir.glob("*.tcx")) == [] # nothing top-level
state = load_state(paths.state_file)
assert state.anchor == ANCHOR
assert state.backup_path == str(backup)
assert state.done_at
again = FakePhone()
assert go(paths, again).outcome is Outcome.ALREADY_DONE
assert again.calls == []
@pytest.mark.parametrize("error", [AdbError("short pull"), OSError("disk full")])
def test_failed_backup_blocks_the_uninstall(
tmp_path: Path, caplog: pytest.LogCaptureFixture, error: Exception
) -> None:
phone = FakePhone(backup_error=error) # type: ignore[arg-type]
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:
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:
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:
paths = world(tmp_path)
assert run_retire(paths, FakePhone(), dry_run=True).outcome is Outcome.DRY_RUN

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

@ -0,0 +1,188 @@
"""RunnerUp retirement: the adb wrapper and the state file (no real adb)."""
from __future__ import annotations
import json
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
from screen_locker._runnerup_retire_state import RetireState, load_state, save_state
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:
self.replies = replies or {}
self.calls: list[list[str]] = []
def __call__(self, cmd: list[str], **_: object) -> SimpleNamespace:
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:
return ra.Phone(run=fake, which=lambda _name: adb)
def test_available_follows_which() -> None:
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:
with pytest.raises(ra.AdbError, match="did not run"):
phone(FakeAdb({"devices": boom})).find_device()
def test_nonzero_exit_carries_the_output() -> None:
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:
device = phone(FakeAdb({"devices": (0, listing, "")})).find_device()
assert device.serial == serial
assert reason in device.reason
def test_installed_needs_the_exact_package_line() -> None:
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"):
def make(cmd: list[str]) -> tuple[int, str, str]:
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:
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:
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:
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:
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:
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:
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:
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:
fake = FakeAdb({"uninstall": (0, "Success\n", ""), "pm list": (0, PKG_LINE, "")})
with pytest.raises(ra.AdbError, match="still listed"):
phone(fake).uninstall("dev")
def test_state_roundtrip_and_anchor(tmp_path: Path) -> None:
path = tmp_path / "deep" / "s.json"
assert load_state(path) == RetireState()
save_state(path, RetireState(anchor_at="2026-11-01T10:45:00+00:00"))
assert load_state(path).anchor is not None
assert RetireState().anchor is None
@pytest.mark.parametrize(
"body", ["{nope", "[]", json.dumps({"anchor_at": "not a date"})]
)
def test_bad_state_is_loud_and_ignored(
tmp_path: Path, body: str, caplog: pytest.LogCaptureFixture
) -> None:
path = tmp_path / "s.json"
path.write_text(body)
with caplog.at_level("WARNING"):
assert load_state(path) == RetireState()
assert "Retire state" in caplog.text
def test_unreadable_state_is_loud(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level("WARNING"):
assert load_state(tmp_path) == RetireState()
assert "unreadable" in caplog.text
def test_state_ignores_unknown_and_non_string_fields(tmp_path: Path) -> None:
path = tmp_path / "s.json"
path.write_text(json.dumps({"done_at": 5, "extra": "x", "backup_path": "/b"}))
assert load_state(path) == RetireState(backup_path="/b")

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

@ -0,0 +1,129 @@
"""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]:
return {str(p.relative_to(root)) for p in root.rglob("*")}
def test_status_without_anchor(tmp_path: Path) -> None:
paths = world(tmp_path, anchored=False, marker=False)
lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type]
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:
paths = world(tmp_path)
before = _tree(tmp_path)
lines = cli.status_lines(paths, FakePhone(), DUE - timedelta(days=1)) # type: ignore[arg-type]
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:
paths = world(tmp_path, anchored=False)
save_state(
paths.state_file,
RetireState(anchor_at=ANCHOR.isoformat(), done_at="2026-11-09T00:00:00+00:00"),
)
lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type]
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:
paths = world(tmp_path)
lines = cli.status_lines(paths, phone, DUE, query_phone=True) # type: ignore[arg-type]
assert lines[4] == expected
def test_main_status_prints_and_exits_zero(
capsys: pytest.CaptureFixture[str],
) -> None:
assert cli.main(["--status"]) == 0 # the isolated defaults: an empty world
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:
seen: list[str] = []
def fake_phone(serial: str) -> FakePhone:
seen.append(serial)
return FakePhone()
monkeypatch.setattr(cli, "Phone", fake_phone)
assert cli.main(["--status", "--query-phone", "--serial", "XYZ"]) == 0
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:
calls: list[bool] = []
def fake_run(paths: Paths, phone: object, *, dry_run: bool) -> RetireResult:
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:
assert cli.main([]) == 0
def test_status_marker_line_for_a_valid_marker_uses_real_files(
tmp_path: Path,
) -> None:
paths = world(tmp_path, marker=False)
put_marker(paths.watch_dir, [RUN])
now = datetime.now(UTC)
assert cli.status_lines(paths, FakePhone(), now)[2].startswith("marker: valid") # type: ignore[arg-type]

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

@ -0,0 +1,262 @@
"""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_marker,
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:
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:
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:
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:
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:
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:
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:
_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:
_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:
_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:
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:
(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:
_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:
_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 _processed(tmp_path: Path) -> Path:
processed = tmp_path / "processed"
put_tcx(processed, "RunnerUp_2026-08-12-17-42-11_Running.tcx")
put_tcx(processed, "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx")
put_tcx(processed, "RunnerUp_2026-08-14-00-58-45_Running.tcx")
put_tcx(processed, "app_2026-11-01.tcx", author=APP)
return processed
_LISTED = [
"RunnerUp_2026-08-12-17-42-11_Running.tcx",
"RunnerUp_2026-08-14-00-58-45_Running.tcx",
]
def test_marker_valid_counts_device_twins_and_ignores_app_files(
tmp_path: Path,
) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, _LISTED)
check = rt.check_marker(marker, processed)
assert check.ok
assert "2 runs" in check.reason
def test_marker_may_name_the_device_prefixed_copy(tmp_path: Path) -> None:
processed = _processed(tmp_path)
names = ["RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx", _LISTED[1]]
assert rt.check_marker(put_marker(tmp_path, names), processed).ok
def test_marker_missing_is_not_valid(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.WARNING):
check = rt.check_marker(tmp_path / "m.json", tmp_path)
assert not check.ok
assert "has not imported yet" in check.reason
assert "marker problem" in caplog.text
def test_marker_unreadable_and_bad_json(tmp_path: Path) -> None:
folder_as_marker = tmp_path / "m"
folder_as_marker.mkdir()
assert "unreadable" in rt.check_marker(folder_as_marker, tmp_path).reason
bad = tmp_path / "bad.json"
bad.write_text("{nope")
assert "unreadable" in rt.check_marker(bad, tmp_path).reason
@pytest.mark.parametrize(
("body", "fragment"),
[
("[]", "not a JSON object"),
('{"count": 0, "tcx_names": []}', "no completed_at"),
('{"completed_at": "", "count": 0, "tcx_names": []}', "no completed_at"),
('{"completed_at": "t", "count": 0, "tcx_names": "x"}', "not a list"),
('{"completed_at": "t", "count": 0, "tcx_names": [1]}', "not a list"),
('{"completed_at": "t", "count": 3, "tcx_names": ["a"]}', "count 3"),
('{"completed_at": "t", "count": true, "tcx_names": ["a"]}', "count True"),
('{"completed_at": "t", "count": "1", "tcx_names": ["a"]}', "count '1'"),
('{"completed_at": "t", "count": 1, "tcx_names": ["x/a"]}', "bare file"),
('{"completed_at": "t", "count": 1, "tcx_names": [".a"]}', "bare file"),
],
)
def test_marker_shape_errors(tmp_path: Path, body: str, fragment: str) -> None:
marker = tmp_path / "m.json"
marker.write_text(body)
check = rt.check_marker(marker, tmp_path)
assert not check.ok
assert fragment in check.reason
def test_marker_listing_one_run_twice_is_invalid(tmp_path: Path) -> None:
processed = _processed(tmp_path)
twin = "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx"
marker = put_marker(tmp_path, [_LISTED[0], twin])
assert "more than once" in rt.check_marker(marker, processed).reason
def test_marker_naming_a_run_not_in_processed_is_invalid(tmp_path: Path) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, [*_LISTED, "RunnerUp_2099-01-01-00-00-00_X.tcx"])
reason = rt.check_marker(marker, processed).reason
assert "not in processed/" in reason
def test_marker_missing_a_processed_run_asks_for_a_reimport(tmp_path: Path) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, _LISTED[:1])
reason = rt.check_marker(marker, processed).reason
assert "re-run the app's import" in reason