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

Harden RunnerUp retirement after review

Browse Source 
Silent-failure checker, mypy, pylint now clean. Hidden packages count
as installed (pm list -u), pull gets a long timeout, wifi entries are
matched by ro.serialno, and a missing log/drop dir is a warning.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZ9CKZsNxtD5Jr9bYD
This commit is contained in:
Krzysztof Rudnicki 2026-11-10 16:08:40 +01:00
parent 50faee673f
commit 0411116f24
13 changed files with 504 additions and 237 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
screen_locker/_runnerup_retire.py
View File

@ -16,17 +16,17 @@ from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import UTC, datetime, timedelta
from enum import StrEnum
import logging
from pathlib import Path
from typing import TYPE_CHECKING
from screen_locker._runnerup_retire_adb import PACKAGE, AdbError
from screen_locker._runnerup_retire_result import LEVEL, Outcome, RetireResult
from screen_locker._runnerup_retire_state import load_state, save_state
from screen_locker._runnerup_retire_tcx import check_marker, find_anchor
from screen_locker._runnerup_retire_tcx import app_run_ends, check_marker, find_anchor
if TYPE_CHECKING:
from screen_locker._runnerup_retire_adb import Phone
from screen_locker._runnerup_retire_adb import PhoneLike
from screen_locker._runnerup_retire_state import RetireState
_logger = logging.getLogger(__name__)
@ -38,57 +38,6 @@ LOG_FILE = Path(__file__).resolve().parent / "log.json"
MARKER_NAME = ".workout_app_import.json"
class Outcome(StrEnum):
"""How a run ended; the value is what the journal shows."""
UNINSTALLED = "uninstalled"
ALREADY_GONE = "already-gone"
ALREADY_DONE = "already-done"
NO_ANCHOR = "no-anchor"
NOT_DUE = "not-due"
DRY_RUN = "dry-run"
DEVICE_ABSENT = "device-absent"
MARKER_INVALID = "marker-invalid"
ADB_MISSING = "adb-missing"
PHONE_ERROR = "phone-error"
BACKUP_FAILED = "backup-failed"
UNINSTALL_FAILED = "uninstall-failed"
# Level per outcome; the failures also make the unit exit non-zero.
_LEVEL = {
Outcome.DEVICE_ABSENT: logging.WARNING,
Outcome.MARKER_INVALID: logging.WARNING,
Outcome.ADB_MISSING: logging.ERROR,
Outcome.PHONE_ERROR: logging.ERROR,
Outcome.BACKUP_FAILED: logging.ERROR,
Outcome.UNINSTALL_FAILED: logging.ERROR,
}
_NONZERO = frozenset(
{
Outcome.MARKER_INVALID,
Outcome.ADB_MISSING,
Outcome.PHONE_ERROR,
Outcome.BACKUP_FAILED,
Outcome.UNINSTALL_FAILED,
}
)
@dataclass(frozen=True)
class RetireResult:
"""What happened and why, in a sentence a human can act on."""
outcome: Outcome
reason: str
logged: bool = False # the failure was already logged where it was caught
@property
def ok(self) -> bool:
"""False for outcomes the unit should surface as failed."""
return self.outcome not in _NONZERO
@dataclass(frozen=True)
class Paths:
"""Every on-disk location, so tests (and ``--status``) can redirect them."""
@ -128,7 +77,7 @@ def _finish(paths: Paths, state: RetireState, now: datetime, why: str) -> None:
)
def _probe(phone: Phone) -> RetireResult | tuple[str, bool]:
def _probe(phone: PhoneLike) -> RetireResult | tuple[str, bool]:
"""``(serial, installed)`` if the phone answers, else the result explaining why."""
if not phone.available():
return RetireResult(
@ -149,7 +98,7 @@ def _probe(phone: Phone) -> RetireResult | tuple[str, bool]:
def _backup(
paths: Paths, phone: Phone, state: RetireState, serial: str, now: datetime
paths: Paths, phone: PhoneLike, state: RetireState, serial: str, now: datetime
) -> RetireState | RetireResult:
"""Back the phone's exports up once; a later retry reuses a finished backup."""
if state.backup_path and Path(state.backup_path).is_dir():
@ -167,7 +116,7 @@ def _backup(
def _retire(
paths: Paths, phone: Phone, state: RetireState, now: datetime, *, dry_run: bool
paths: Paths, phone: PhoneLike, state: RetireState, now: datetime, *, dry_run: bool
) -> RetireResult:
"""Everything after "it is due and the marker is good": the phone steps."""
probed = _probe(phone)
@ -199,6 +148,21 @@ def _retire(
)
def _no_anchor(paths: Paths) -> RetireResult:
"""Why there is no anchor; a missing source is a warning, not "nothing yet"."""
absent = [str(p) for p in (paths.watch_dir, paths.log_file) if not p.exists()]
if absent:
return RetireResult(
Outcome.SOURCES_MISSING,
f"cannot look for app runs, missing: {', '.join(absent)}",
)
seen = len(app_run_ends(paths.anchor_dirs()))
return RetireResult(
Outcome.NO_ANCHOR,
f"no app-recorded run credited yet ({seen} app TCX on disk); nothing to retire",
)
def _gate(
paths: Paths, state: RetireState, now: datetime, *, dry_run: bool
) -> RetireState | RetireResult:
@ -209,9 +173,7 @@ def _gate(
)
anchor = derive_anchor(paths, state)
if anchor is None:
return RetireResult(
Outcome.NO_ANCHOR, "no app-recorded run credited yet; nothing to retire"
)
return _no_anchor(paths)
if not state.anchor_at:
state = replace(state, anchor_at=anchor.isoformat())
if not dry_run:
@ -230,7 +192,11 @@ def _gate(
def run_retire(
paths: Paths, phone: Phone, now: datetime | None = None, *, dry_run: bool = False
paths: Paths,
phone: PhoneLike,
now: datetime | None = None,
*,
dry_run: bool = False,
) -> RetireResult:
"""One pass: uninstall RunnerUp if (and only if) every condition holds."""
when = now or datetime.now(UTC)
@ -241,7 +207,7 @@ def run_retire(
result = _retire(paths, phone, gated, when, dry_run=dry_run)
if not result.logged:
_logger.log(
_LEVEL.get(result.outcome, logging.INFO),
LEVEL.get(result.outcome, logging.INFO),
"RunnerUp retirement: %s: %s",
result.outcome.value,
result.reason,

62
screen_locker/_runnerup_retire_adb.py
View File

@ -13,7 +13,7 @@ import logging
from pathlib import Path
import shutil
import subprocess
from typing import TYPE_CHECKING, Any, NoReturn
from typing import TYPE_CHECKING, Any, NoReturn, Protocol
if TYPE_CHECKING:
from collections.abc import Callable
@ -24,6 +24,9 @@ PHONE_SERIAL = "23181JEGR08034"
PACKAGE = "org.runnerup.free"
REMOTE_DIR = "/sdcard/Documents/RunnerUp"
_ADB_TIMEOUT_S = 50
# The whole export history (TCX + GPX) over wifi adb; the oneshot unit has no
# start timeout, so a slow pull is allowed to finish.
_PULL_TIMEOUT_S = 900
class AdbError(RuntimeError):
@ -42,6 +45,25 @@ class Device:
reason: str
class PhoneLike(Protocol):
"""What the retirement needs from a phone (tests substitute a fake)."""
def available(self) -> bool:
"""Whether adb can be run at all."""
def find_device(self) -> Device:
"""The attached phone, or why there is none."""
def installed(self, dev: str, *, hidden: bool = True) -> bool:
"""Whether RunnerUp is on the phone."""
def backup(self, dev: str, final: Path) -> tuple[Path, int]:
"""Pull the exports to ``final``."""
def uninstall(self, dev: str) -> None:
"""Uninstall RunnerUp or raise ``AdbError``."""
class Phone:
"""adb calls against one phone, with the process runner injectable."""
@ -61,11 +83,11 @@ class Phone:
"""Whether an ``adb`` binary is on PATH."""
return self._which("adb") is not None
def _adb(self, *args: str) -> str:
def _adb(self, *args: str, timeout: int = _ADB_TIMEOUT_S) -> str:
cmd = ["adb", *args]
try:
done = self._run(
cmd, capture_output=True, text=True, timeout=_ADB_TIMEOUT_S, check=False
cmd, capture_output=True, text=True, timeout=timeout, check=False
)
except (OSError, subprocess.TimeoutExpired) as exc:
msg = f"{' '.join(cmd)} did not run: {exc}"
@ -75,13 +97,30 @@ class Phone:
_fail(f"{' '.join(cmd)} exited {done.returncode}: {detail}")
return str(done.stdout)
def _is_phone(self, serial: str) -> bool:
"""Whether an ``adb devices`` entry is the phone.
USB and mDNS serials embed the phone's serial; a plain ``ip:port``
wifi entry does not, so that one is asked for its ``ro.serialno``.
"""
if self.serial in serial:
return True
if ":" not in serial:
return False
try:
asked = self._adb("-s", serial, "shell", "getprop", "ro.serialno")
except AdbError as exc:
_logger.warning("Could not read the serial of %s (%s)", serial, exc)
return False
return asked.strip() == self.serial
def find_device(self) -> Device:
"""The attached phone, over USB or wifi (a wireless serial embeds it)."""
listing = self._adb("devices")
states: list[str] = []
for line in listing.splitlines()[1:]:
serial, _, state = line.partition("\t")
if self.serial not in serial:
if not self._is_phone(serial):
continue
if state.strip() == "device":
return Device(serial, "attached")
@ -90,9 +129,14 @@ class Phone:
return Device(None, f"phone is {'/'.join(states)}, not ready")
return Device(None, f"phone {self.serial} is not attached")
def installed(self, dev: str) -> bool:
"""Whether the package is on the phone (exact ``package:`` line)."""
out = self._adb("-s", dev, "shell", "pm", "list", "packages", PACKAGE)
def installed(self, dev: str, *, hidden: bool = True) -> bool:
"""Whether the package is on the phone (exact ``package:`` line).
``hidden`` adds ``-u`` so an app that focus mode hides or disables
still counts as present; it must not be mistaken for an uninstall.
"""
flag = ["-u"] if hidden else []
out = self._adb("-s", dev, "shell", "pm", "list", "packages", *flag, PACKAGE)
return f"package:{PACKAGE}" in out.split()
def backup(self, dev: str, final: Path) -> tuple[Path, int]:
@ -108,7 +152,7 @@ class Phone:
staging = final.with_name(final.name + ".partial")
shutil.rmtree(staging, ignore_errors=True)
staging.mkdir(parents=True)
self._adb("-s", dev, "pull", REMOTE_DIR, str(staging))
self._adb("-s", dev, "pull", REMOTE_DIR, str(staging), timeout=_PULL_TIMEOUT_S)
pulled = staging / Path(REMOTE_DIR).name
got = len(list(pulled.rglob("*.tcx"))) if pulled.is_dir() else -1
if got < wanted:
@ -132,6 +176,6 @@ class Phone:
raise AdbError(msg) from exc
if "Success" not in out:
_fail(f"uninstall did not report Success: {out.strip()!r}")
if self.installed(dev):
if self.installed(dev, hidden=False):
_fail(f"{PACKAGE} is still listed after uninstall")
_logger.info("Uninstalled %s from %s", PACKAGE, dev)

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

10
screen_locker/_runnerup_retire_state.py
View File

@ -8,7 +8,7 @@ and never moved: a later, earlier-looking credit must not push the deadline.
from __future__ import annotations
from dataclasses import asdict, dataclass
from dataclasses import asdict, dataclass, fields
from datetime import datetime
import json
import logging
@ -36,10 +36,10 @@ class RetireState:
def load_state(path: Path) -> RetireState:
"""Read the state; a missing file is a fresh start, a bad one is loud."""
if not path.exists():
return RetireState()
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return RetireState()
except (OSError, ValueError) as exc:
_logger.warning(
"Retire state %s is unreadable (%s): starting from scratch, so the "
@ -51,8 +51,8 @@ def load_state(path: Path) -> RetireState:
if not isinstance(raw, dict):
_logger.warning("Retire state %s is not a JSON object: ignored", path)
return RetireState()
fields = RetireState.__dataclass_fields__
clean = {k: v for k, v in raw.items() if k in fields and isinstance(v, str)}
known = {f.name for f in fields(RetireState)}
clean = {k: v for k, v in raw.items() if k in known and isinstance(v, str)}
try:
state = RetireState(**clean)
_ = state.anchor

26
screen_locker/_runnerup_retire_tcx.py
View File

@ -116,16 +116,19 @@ def _credit_ends(log_file: Path) -> list[tuple[float, datetime]]:
return found
def find_anchor(log_file: Path, dirs: Iterable[Path]) -> datetime | None:
"""When the PC first credited an app-recorded run, or None if it never did."""
ends = [
def app_run_ends(dirs: Iterable[Path]) -> list[float]:
"""The end time of every app-authored TCX that has a usable one."""
return [
end
for directory in dirs
for path in list_tcx(directory)
if is_app_tcx(path) and (end := _ended_at(path)) is not None
]
if not ends:
return None
def find_anchor(log_file: Path, dirs: Iterable[Path]) -> datetime | None:
"""When the PC first credited an app-recorded run, or None if it never did."""
ends = app_run_ends(dirs)
credited = [
when
for run_end, when in _credit_ends(log_file)
@ -164,12 +167,15 @@ def _marker_names(marker: object) -> tuple[list[str] | None, str]:
def _read_marker(marker_file: Path) -> tuple[object, str]:
"""The parsed marker JSON, or ``(None, why it could not be read)``."""
try:
return json.loads(marker_file.read_text(encoding="utf-8")), ""
except FileNotFoundError:
if not marker_file.exists():
why = f"{marker_file} is missing: the app has not imported yet"
except (OSError, ValueError) as exc:
why = f"{marker_file} is unreadable ({exc})"
else:
try:
return json.loads(marker_file.read_text(encoding="utf-8")), ""
except (OSError, ValueError) as exc:
why = f"{marker_file} is unreadable ({exc})"
_logger.warning("Import marker problem: %s", why)
return None, why
_logger.warning("Import marker problem: %s", why)
return None, why

10
screen_locker/runnerup_retire.py
View File

@ -14,6 +14,7 @@ import argparse
from datetime import UTC, datetime
import logging
import sys
from typing import TYPE_CHECKING
from screen_locker._runnerup_retire import (
GRACE,
@ -25,12 +26,15 @@ from screen_locker._runnerup_retire_adb import PHONE_SERIAL, AdbError, Phone
from screen_locker._runnerup_retire_state import load_state
from screen_locker._runnerup_retire_tcx import check_marker
if TYPE_CHECKING:
from screen_locker._runnerup_retire_adb import PhoneLike
__all__ = ["main", "status_lines"]
_logger = logging.getLogger(__name__)
def _installed_line(phone: Phone, *, query: bool) -> str:
def _installed_line(phone: PhoneLike, *, query: bool) -> str:
if not query:
return "installed: not queried (add --query-phone)"
try:
@ -44,7 +48,7 @@ def _installed_line(phone: Phone, *, query: bool) -> str:
def status_lines(
paths: Paths, phone: Phone, now: datetime, *, query_phone: bool = False
paths: Paths, phone: PhoneLike, now: datetime, *, query_phone: bool = False
) -> list[str]:
"""The ``--status`` report. Read-only: nothing is created or saved."""
state = load_state(paths.state_file)
@ -69,7 +73,7 @@ def status_lines(
def main(argv: list[str] | None = None) -> int:
"""CLI entry; returns the process exit code."""
parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
parser = argparse.ArgumentParser(description=__doc__.split("\n\n", maxsplit=1)[0])
parser.add_argument(
"--status", action="store_true", help="print state, change nothing"
)

9
screen_locker/tests/_retire_helpers.py
View File

@ -87,23 +87,27 @@ class FakePhone:
device: Device = field(default_factory=lambda: Device("S", "attached"))
present: bool = True
ask_error: AdbError | None = None
backup_error: AdbError | OSError | None = None
backup_error: Exception | None = None
uninstall_error: AdbError | None = None
calls: list[str] = field(default_factory=list)
def available(self) -> bool:
"""Available."""
return self.adb
def find_device(self) -> Device:
"""Find device."""
if self.ask_error:
raise self.ask_error
return self.device
def installed(self, dev: str) -> bool:
def installed(self, dev: str, *, hidden: bool = True) -> bool:
"""Installed."""
self.calls.append(f"installed {dev}")
return self.present
def backup(self, dev: str, final: Path) -> tuple[Path, int]:
"""Backup."""
self.calls.append("backup")
if self.backup_error:
raise self.backup_error
@ -111,6 +115,7 @@ class FakePhone:
return final, 3
def uninstall(self, dev: str) -> None:
"""Uninstall."""
self.calls.append("uninstall")
if self.uninstall_error:
raise self.uninstall_error

51
screen_locker/tests/test_runnerup_retire.py
View File

@ -25,23 +25,46 @@ if TYPE_CHECKING:
from pathlib import Path
def go(paths: Paths, phone: FakePhone, now: datetime = DUE, **kw: bool):
return run_retire(paths, phone, now, **kw) # type: ignore[arg-type]
def go(paths: Paths, phone: FakePhone, now: datetime = DUE, *, dry_run: bool = False):
"""Go."""
return run_retire(paths, phone, now, dry_run=dry_run)
def test_no_anchor_never_touches_anything(tmp_path: Path) -> None:
"""Test no anchor never touches anything."""
paths = world(tmp_path, anchored=False)
phone = FakePhone()
result = go(paths, phone, DUE + timedelta(days=900))
assert result.outcome is Outcome.NO_ANCHOR
assert result.ok
assert phone.calls == []
assert not phone.calls
assert not paths.state_file.exists()
def test_missing_sources_are_a_warning_not_a_quiet_nothing(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test missing sources are a warning not a quiet nothing."""
paths = Paths(tmp_path / "s.json", tmp_path / "nodrop", tmp_path / "nolog.json")
with caplog.at_level(logging.INFO):
result = go(paths, FakePhone())
assert result.outcome is Outcome.SOURCES_MISSING
assert result.ok
assert "nodrop" in result.reason
assert "nolog.json" in result.reason
assert caplog.records[-1].levelno == logging.WARNING
def test_no_anchor_says_how_many_app_runs_it_saw(tmp_path: Path) -> None:
"""Test no anchor says how many app runs it saw."""
paths = world(tmp_path, anchored=False)
assert "(1 app TCX on disk)" in go(paths, FakePhone()).reason
def test_not_due_persists_the_anchor_and_says_when(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test not due persists the anchor and says when."""
paths = world(tmp_path)
with caplog.at_level(logging.INFO):
result = go(paths, FakePhone(), DUE - timedelta(minutes=1))
@ -52,6 +75,7 @@ def test_not_due_persists_the_anchor_and_says_when(
def test_stored_anchor_wins_over_the_log(tmp_path: Path) -> None:
"""Test stored anchor wins over the log."""
paths = world(tmp_path, anchored=False)
save_state(paths.state_file, RetireState(anchor_at=ANCHOR.isoformat()))
assert go(paths, FakePhone(), ANCHOR).outcome is Outcome.NOT_DUE
@ -60,6 +84,7 @@ def test_stored_anchor_wins_over_the_log(tmp_path: Path) -> None:
def test_dry_run_writes_nothing_and_stops_before_the_phone_changes(
tmp_path: Path,
) -> None:
"""Test dry run writes nothing and stops before the phone changes."""
paths = world(tmp_path)
phone = FakePhone()
result = go(paths, phone, dry_run=True)
@ -75,6 +100,7 @@ def test_dry_run_writes_nothing_and_stops_before_the_phone_changes(
def test_invalid_marker_blocks_the_uninstall_loudly(
tmp_path: Path, caplog: pytest.LogCaptureFixture, variant: str
) -> None:
"""Test invalid marker blocks the uninstall loudly."""
paths = world(tmp_path, marker=variant == "mismatch")
if variant == "mismatch":
put_marker(paths.watch_dir, [RUN], count=2)
@ -84,11 +110,12 @@ def test_invalid_marker_blocks_the_uninstall_loudly(
assert result.outcome is Outcome.MARKER_INVALID
assert not result.ok
assert "NOT uninstalling" in result.reason
assert phone.calls == []
assert not phone.calls
assert caplog.records[-1].levelno == logging.WARNING
def test_adb_missing_is_a_failure(tmp_path: Path) -> None:
"""Test adb missing is a failure."""
result = go(world(tmp_path), FakePhone(adb=False))
assert result.outcome is Outcome.ADB_MISSING
assert not result.ok
@ -97,18 +124,20 @@ def test_adb_missing_is_a_failure(tmp_path: Path) -> None:
def test_absent_phone_retries_quietly_but_warns(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test absent phone retries quietly but warns."""
phone = FakePhone(device=Device(None, "phone S is not attached"))
with caplog.at_level(logging.WARNING):
result = go(world(tmp_path), phone)
assert result.outcome is Outcome.DEVICE_ABSENT
assert result.ok
assert "not attached" in caplog.text
assert phone.calls == []
assert not phone.calls
def test_adb_error_while_asking_is_a_logged_failure(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test adb error while asking is a logged failure."""
phone = FakePhone(ask_error=AdbError("server died"))
with caplog.at_level(logging.ERROR):
result = go(world(tmp_path), phone)
@ -118,6 +147,7 @@ def test_adb_error_while_asking_is_a_logged_failure(
def test_already_uninstalled_is_recorded_done(tmp_path: Path) -> None:
"""Test already uninstalled is recorded done."""
paths = world(tmp_path)
result = go(paths, FakePhone(present=False))
assert result.outcome is Outcome.ALREADY_GONE
@ -126,6 +156,7 @@ def test_already_uninstalled_is_recorded_done(tmp_path: Path) -> None:
def test_full_run_backs_up_then_uninstalls_then_never_again(tmp_path: Path) -> None:
"""Test full run backs up then uninstalls then never again."""
paths = world(tmp_path)
phone = FakePhone()
result = go(paths, phone)
@ -133,21 +164,22 @@ def test_full_run_backs_up_then_uninstalls_then_never_again(tmp_path: Path) -> N
assert phone.calls == ["installed S", "backup", "uninstall"]
backup = paths.watch_dir / "runnerup_phone_backup_2026-11-08"
assert backup.is_dir()
assert list(paths.watch_dir.glob("*.tcx")) == [] # nothing top-level
assert not list(paths.watch_dir.glob("*.tcx")) # nothing top-level
state = load_state(paths.state_file)
assert state.anchor == ANCHOR
assert state.backup_path == str(backup)
assert state.done_at
again = FakePhone()
assert go(paths, again).outcome is Outcome.ALREADY_DONE
assert again.calls == []
assert not again.calls
@pytest.mark.parametrize("error", [AdbError("short pull"), OSError("disk full")])
def test_failed_backup_blocks_the_uninstall(
tmp_path: Path, caplog: pytest.LogCaptureFixture, error: Exception
) -> None:
phone = FakePhone(backup_error=error) # type: ignore[arg-type]
"""Test failed backup blocks the uninstall."""
phone = FakePhone(backup_error=error)
with caplog.at_level(logging.ERROR):
result = go(world(tmp_path), phone)
assert result.outcome is Outcome.BACKUP_FAILED
@ -157,6 +189,7 @@ def test_failed_backup_blocks_the_uninstall(
def test_failed_uninstall_keeps_the_backup_for_the_retry(tmp_path: Path) -> None:
"""Test failed uninstall keeps the backup for the retry."""
paths = world(tmp_path)
first = FakePhone(uninstall_error=AdbError("DELETE_FAILED_DEVICE_POLICY_MANAGER"))
result = go(paths, first)
@ -169,6 +202,7 @@ def test_failed_uninstall_keeps_the_backup_for_the_retry(tmp_path: Path) -> None
def test_default_paths_are_the_module_constants() -> None:
"""Test default paths are the module constants."""
paths = Paths.default()
assert paths.state_file == rr.STATE_FILE
assert paths.marker == rr.WATCH_DIR / rr.MARKER_NAME
@ -177,5 +211,6 @@ def test_default_paths_are_the_module_constants() -> None:
def test_run_retire_defaults_to_the_real_clock(tmp_path: Path) -> None:
"""Test run retire defaults to the real clock."""
paths = world(tmp_path)
assert run_retire(paths, FakePhone(), dry_run=True).outcome is Outcome.DRY_RUN

111
screen_locker/tests/test_runnerup_retire_adb.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations
import json
from pathlib import Path
import subprocess
from types import SimpleNamespace
@ -11,7 +10,6 @@ from typing import Any
import pytest
from screen_locker import _runnerup_retire_adb as ra
from screen_locker._runnerup_retire_state import RetireState, load_state, save_state
SERIAL = ra.PHONE_SERIAL
PKG_LINE = f"package:{ra.PACKAGE}\n"
@ -21,10 +19,12 @@ class FakeAdb:
"""A ``subprocess.run`` stand-in: replies by the adb sub-command."""
def __init__(self, replies: dict[str, Any] | None = None) -> None:
"""Init ."""
self.replies = replies or {}
self.calls: list[list[str]] = []
def __call__(self, cmd: list[str], **_: object) -> SimpleNamespace:
"""Call ."""
self.calls.append(cmd)
for key, reply in self.replies.items():
if key in " ".join(cmd):
@ -36,10 +36,12 @@ class FakeAdb:
def phone(fake: FakeAdb, adb: str | None = "/usr/bin/adb") -> ra.Phone:
"""Phone."""
return ra.Phone(run=fake, which=lambda _name: adb)
def test_available_follows_which() -> None:
"""Test available follows which."""
assert phone(FakeAdb()).available() is True
assert phone(FakeAdb(), adb=None).available() is False
@ -48,11 +50,13 @@ def test_available_follows_which() -> None:
"boom", [FileNotFoundError("adb"), subprocess.TimeoutExpired("adb", 1)]
)
def test_adb_failing_to_run_is_an_adb_error(boom: BaseException) -> None:
"""Test adb failing to run is an adb error."""
with pytest.raises(ra.AdbError, match="did not run"):
phone(FakeAdb({"devices": boom})).find_device()
def test_nonzero_exit_carries_the_output() -> None:
"""Test nonzero exit carries the output."""
fake = FakeAdb({"devices": (1, "", "daemon not running")})
with pytest.raises(ra.AdbError, match="daemon not running"):
phone(fake).find_device()
@ -73,12 +77,69 @@ def test_nonzero_exit_carries_the_output() -> None:
],
)
def test_find_device(listing: str, serial: str | None, reason: str) -> None:
"""Test find device."""
device = phone(FakeAdb({"devices": (0, listing, "")})).find_device()
assert device.serial == serial
assert reason in device.reason
def test_wifi_entry_is_asked_for_its_hardware_serial() -> None:
"""Test wifi entry is asked for its hardware serial."""
listing = "List of devices attached\n192.168.1.5:5555\tdevice\n"
fake = FakeAdb({"devices": (0, listing, ""), "getprop": (0, f"{SERIAL}\n", "")})
device = phone(fake).find_device()
assert device.serial == "192.168.1.5:5555"
assert ["adb", "-s", "192.168.1.5:5555", "shell", "getprop", "ro.serialno"] in (
fake.calls
)
def test_wifi_entry_of_another_phone_is_not_ours() -> None:
"""Test wifi entry of another phone is not ours."""
listing = "List of devices attached\n10.0.0.2:5555\tdevice\n"
fake = FakeAdb({"devices": (0, listing, ""), "getprop": (0, "OTHER\n", "")})
assert phone(fake).find_device().serial is None
def test_wifi_entry_that_will_not_answer_is_warned_about(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test wifi entry that will not answer is warned about."""
listing = "List of devices attached\n10.0.0.2:5555\tdevice\n"
fake = FakeAdb({"devices": (0, listing, ""), "getprop": (1, "", "offline")})
with caplog.at_level("WARNING"):
assert phone(fake).find_device().serial is None
assert "Could not read the serial" in caplog.text
def test_presence_probe_sees_hidden_packages_but_the_final_check_does_not() -> None:
"""Test presence probe sees hidden packages but the final check does not."""
fake = FakeAdb({"pm list": (0, PKG_LINE, "")})
phone(fake).installed("dev")
phone(fake).installed("dev", hidden=False)
assert "-u" in fake.calls[0]
assert "-u" not in fake.calls[1]
def test_pull_gets_a_long_timeout_and_other_calls_the_short_one(
tmp_path: Path,
) -> None:
"""Test pull gets a long timeout and other calls the short one."""
timeouts: dict[str, object] = {}
make = _pull_creates(1)
def run(cmd: list[str], **kw: object) -> SimpleNamespace:
"""Run."""
timeouts[cmd[3]] = kw["timeout"]
code, out, err = make(cmd) if cmd[3] == "pull" else (0, "a.tcx\n", "")
return SimpleNamespace(returncode=code, stdout=out, stderr=err)
ra.Phone(run=run, which=lambda _n: "adb").backup("dev", tmp_path / "b")
assert timeouts == {"shell": 50, "pull": 900}
def test_installed_needs_the_exact_package_line() -> None:
"""Test installed needs the exact package line."""
yes = FakeAdb({"pm list": (0, PKG_LINE, "")})
assert phone(yes).installed("dev") is True
assert yes.calls[0][:3] == ["adb", "-s", "dev"]
@ -87,7 +148,10 @@ def test_installed_needs_the_exact_package_line() -> None:
def _pull_creates(count: int, root: str = "RunnerUp"):
"""Pull creates."""
def make(cmd: list[str]) -> tuple[int, str, str]:
"""Make."""
folder = Path(cmd[-1]) / root
folder.mkdir(parents=True)
for i in range(count):
@ -98,6 +162,7 @@ def _pull_creates(count: int, root: str = "RunnerUp"):
def test_backup_stages_then_renames_and_counts(tmp_path: Path) -> None:
"""Test backup stages then renames and counts."""
fake = FakeAdb({"ls": (0, "a.tcx\nb.tcx\nb.gpx\n", ""), "pull": _pull_creates(2)})
final = tmp_path / "backup_2026"
target, count = phone(fake).backup("dev", final)
@ -107,6 +172,7 @@ def test_backup_stages_then_renames_and_counts(tmp_path: Path) -> None:
def test_backup_never_overwrites_an_existing_folder(tmp_path: Path) -> None:
"""Test backup never overwrites an existing folder."""
final = tmp_path / "backup"
final.mkdir()
(tmp_path / "backup.partial").mkdir() # a stale half-pull is cleared
@ -116,6 +182,7 @@ def test_backup_never_overwrites_an_existing_folder(tmp_path: Path) -> None:
def test_backup_short_pull_is_an_error_and_not_renamed(tmp_path: Path) -> None:
"""Test backup short pull is an error and not renamed."""
fake = FakeAdb({"ls": (0, "a.tcx\nb.tcx\n", ""), "pull": _pull_creates(1)})
with pytest.raises(ra.AdbError, match="pulled 1 of 2"):
phone(fake).backup("dev", tmp_path / "b")
@ -123,18 +190,21 @@ def test_backup_short_pull_is_an_error_and_not_renamed(tmp_path: Path) -> None:
def test_backup_pull_that_creates_nothing_is_an_error(tmp_path: Path) -> None:
"""Test backup pull that creates nothing is an error."""
fake = FakeAdb({"ls": (0, "a.tcx\n", "")})
with pytest.raises(ra.AdbError, match="pulled -1 of 1"):
phone(fake).backup("dev", tmp_path / "b")
def test_uninstall_succeeds_only_when_the_package_is_gone() -> None:
"""Test uninstall succeeds only when the package is gone."""
fake = FakeAdb({"uninstall": (0, "Success\n", ""), "pm list": (0, "", "")})
phone(fake).uninstall("dev")
assert ["adb", "-s", "dev", "uninstall", ra.PACKAGE] in fake.calls
def test_uninstall_refusal_keeps_adbs_reason() -> None:
"""Test uninstall refusal keeps adbs reason."""
fail = "Failure [DELETE_FAILED_DEVICE_POLICY_MANAGER]"
fake = FakeAdb({"uninstall": (1, fail, "")})
with pytest.raises(ra.AdbError, match="DELETE_FAILED_DEVICE_POLICY_MANAGER"):
@ -142,47 +212,14 @@ def test_uninstall_refusal_keeps_adbs_reason() -> None:
def test_uninstall_without_success_text_is_an_error() -> None:
"""Test uninstall without success text is an error."""
fake = FakeAdb({"uninstall": (0, "weird\n", "")})
with pytest.raises(ra.AdbError, match="did not report Success"):
phone(fake).uninstall("dev")
def test_uninstall_that_leaves_the_package_is_an_error() -> None:
"""Test uninstall that leaves the package is an error."""
fake = FakeAdb({"uninstall": (0, "Success\n", ""), "pm list": (0, PKG_LINE, "")})
with pytest.raises(ra.AdbError, match="still listed"):
phone(fake).uninstall("dev")
def test_state_roundtrip_and_anchor(tmp_path: Path) -> None:
path = tmp_path / "deep" / "s.json"
assert load_state(path) == RetireState()
save_state(path, RetireState(anchor_at="2026-11-01T10:45:00+00:00"))
assert load_state(path).anchor is not None
assert RetireState().anchor is None
@pytest.mark.parametrize(
"body", ["{nope", "[]", json.dumps({"anchor_at": "not a date"})]
)
def test_bad_state_is_loud_and_ignored(
tmp_path: Path, body: str, caplog: pytest.LogCaptureFixture
) -> None:
path = tmp_path / "s.json"
path.write_text(body)
with caplog.at_level("WARNING"):
assert load_state(path) == RetireState()
assert "Retire state" in caplog.text
def test_unreadable_state_is_loud(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level("WARNING"):
assert load_state(tmp_path) == RetireState()
assert "unreadable" in caplog.text
def test_state_ignores_unknown_and_non_string_fields(tmp_path: Path) -> None:
path = tmp_path / "s.json"
path.write_text(json.dumps({"done_at": 5, "extra": "x", "backup_path": "/b"}))
assert load_state(path) == RetireState(backup_path="/b")

28
screen_locker/tests/test_runnerup_retire_cli.py
View File

@ -19,12 +19,14 @@ if TYPE_CHECKING:
def _tree(root: Path) -> set[str]:
"""Tree."""
return {str(p.relative_to(root)) for p in root.rglob("*")}
def test_status_without_anchor(tmp_path: Path) -> None:
"""Test status without anchor."""
paths = world(tmp_path, anchored=False, marker=False)
lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type]
lines = cli.status_lines(paths, FakePhone(), DUE)
assert lines[0].startswith("anchor: none")
assert lines[1] == "due: never"
assert "INVALID" in lines[2]
@ -35,9 +37,10 @@ def test_status_without_anchor(tmp_path: Path) -> None:
def test_status_is_strictly_read_only_and_marks_a_derived_anchor(
tmp_path: Path,
) -> None:
"""Test status is strictly read only and marks a derived anchor."""
paths = world(tmp_path)
before = _tree(tmp_path)
lines = cli.status_lines(paths, FakePhone(), DUE - timedelta(days=1)) # type: ignore[arg-type]
lines = cli.status_lines(paths, FakePhone(), DUE - timedelta(days=1))
assert _tree(tmp_path) == before
assert not paths.state_file.parent.exists()
assert "derived, not yet stored" in lines[0]
@ -46,12 +49,13 @@ def test_status_is_strictly_read_only_and_marks_a_derived_anchor(
def test_status_with_stored_anchor_and_done(tmp_path: Path) -> None:
"""Test status with stored anchor and done."""
paths = world(tmp_path, anchored=False)
save_state(
paths.state_file,
RetireState(anchor_at=ANCHOR.isoformat(), done_at="2026-11-09T00:00:00+00:00"),
)
lines = cli.status_lines(paths, FakePhone(), DUE) # type: ignore[arg-type]
lines = cli.status_lines(paths, FakePhone(), DUE)
assert "(stored)" in lines[0]
assert lines[3] == "done: 2026-11-09T00:00:00+00:00"
@ -69,15 +73,17 @@ def test_status_with_stored_anchor_and_done(tmp_path: Path) -> None:
],
)
def test_status_query_phone(tmp_path: Path, phone: FakePhone, expected: str) -> None:
"""Test status query phone."""
paths = world(tmp_path)
lines = cli.status_lines(paths, phone, DUE, query_phone=True) # type: ignore[arg-type]
lines = cli.status_lines(paths, phone, DUE, query_phone=True)
assert lines[4] == expected
def test_main_status_prints_and_exits_zero(
capsys: pytest.CaptureFixture[str],
) -> None:
assert cli.main(["--status"]) == 0 # the isolated defaults: an empty world
"""Test main status prints and exits zero."""
assert not cli.main(["--status"]) # the isolated defaults: an empty world
out = capsys.readouterr().out
assert "anchor: none" in out
assert "not queried" in out
@ -86,14 +92,16 @@ def test_main_status_prints_and_exits_zero(
def test_main_status_query_phone_uses_the_serial(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""Test main status query phone uses the serial."""
seen: list[str] = []
def fake_phone(serial: str) -> FakePhone:
"""Fake phone."""
seen.append(serial)
return FakePhone()
monkeypatch.setattr(cli, "Phone", fake_phone)
assert cli.main(["--status", "--query-phone", "--serial", "XYZ"]) == 0
assert not cli.main(["--status", "--query-phone", "--serial", "XYZ"])
assert seen == ["XYZ"]
assert "installed: yes" in capsys.readouterr().out
@ -105,9 +113,11 @@ def test_main_status_query_phone_uses_the_serial(
def test_main_exit_code_follows_the_outcome(
monkeypatch: pytest.MonkeyPatch, outcome: Outcome, code: int
) -> None:
"""Test main exit code follows the outcome."""
calls: list[bool] = []
def fake_run(paths: Paths, phone: object, *, dry_run: bool) -> RetireResult:
"""Fake run."""
calls.append(dry_run)
return RetireResult(outcome, "why")
@ -117,13 +127,15 @@ def test_main_exit_code_follows_the_outcome(
def test_main_with_nothing_to_retire_is_a_clean_noop() -> None:
assert cli.main([]) == 0
"""Test main with nothing to retire is a clean noop."""
assert not cli.main([])
def test_status_marker_line_for_a_valid_marker_uses_real_files(
tmp_path: Path,
) -> None:
"""Test status marker line for a valid marker uses real files."""
paths = world(tmp_path, marker=False)
put_marker(paths.watch_dir, [RUN])
now = datetime.now(UTC)
assert cli.status_lines(paths, FakePhone(), now)[2].startswith("marker: valid") # type: ignore[arg-type]
assert cli.status_lines(paths, FakePhone(), now)[2].startswith("marker: valid")

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

113
screen_locker/tests/test_runnerup_retire_tcx.py
View File

@ -14,7 +14,6 @@ from screen_locker.tests._retire_helpers import (
END_UNIX,
credit,
put_log,
put_marker,
put_tcx,
)
@ -25,6 +24,7 @@ _AT = "2026-11-01T10:45:00+00:00"
def test_run_key_drops_device_prefix_only() -> None:
"""Test run key drops device prefix only."""
plain = "RunnerUp_2026-08-22-23-51-04_Running.tcx"
twin = "RunnerUp_Pixel_6a_2026-08-22-23-51-04_Running.tcx"
assert rt.run_key(twin) == plain
@ -32,6 +32,7 @@ def test_run_key_drops_device_prefix_only() -> None:
def test_list_tcx_skips_dotfiles_other_types_and_missing_dir(tmp_path: Path) -> None:
"""Test list tcx skips dotfiles other types and missing dir."""
assert rt.list_tcx(tmp_path / "nope") == []
put_tcx(tmp_path, "a.tcx")
put_tcx(tmp_path, ".partial.tcx")
@ -52,10 +53,12 @@ def test_list_tcx_skips_dotfiles_other_types_and_missing_dir(tmp_path: Path) ->
def test_is_app_tcx_reads_author_and_creator(
tmp_path: Path, kwargs: dict[str, str], expected: bool
) -> None:
"""Test is app tcx reads author and creator."""
assert rt.is_app_tcx(put_tcx(tmp_path, "r.tcx", **kwargs)) is expected
def test_is_app_tcx_ignores_the_name_in_free_text(tmp_path: Path) -> None:
"""Test is app tcx ignores the name in free text."""
path = tmp_path / "n.tcx"
path.write_text(f"<Notes>{APP}</Notes>")
assert rt.is_app_tcx(path) is False
@ -64,18 +67,21 @@ def test_is_app_tcx_ignores_the_name_in_free_text(tmp_path: Path) -> None:
def test_is_app_tcx_unreadable_is_a_loud_non_app(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test is app tcx unreadable is a loud non app."""
with caplog.at_level(logging.WARNING):
assert rt.is_app_tcx(tmp_path) is False
assert "Cannot read" in caplog.text
def _app_run(tmp_path: Path, name: str = "app.tcx", where: str = "") -> Path:
"""App run."""
return put_tcx(tmp_path / where if where else tmp_path, name, author=APP)
def test_anchor_is_the_credit_time_of_the_entry_matching_the_run_end(
tmp_path: Path,
) -> None:
"""Test anchor is the credit time of the entry matching the run end."""
_app_run(tmp_path, where="processed")
log = tmp_path / "log.json"
put_log(
@ -92,6 +98,7 @@ def test_anchor_is_the_credit_time_of_the_entry_matching_the_run_end(
def test_anchor_accepts_a_stringified_end_and_the_earliest_credit(
tmp_path: Path,
) -> None:
"""Test anchor accepts a stringified end and the earliest credit."""
_app_run(tmp_path)
log = tmp_path / "log.json"
put_log(
@ -106,6 +113,7 @@ def test_anchor_accepts_a_stringified_end_and_the_earliest_credit(
def test_same_day_runnerup_credit_does_not_anchor_the_app_run(tmp_path: Path) -> None:
"""Test same day runnerup credit does not anchor the app run."""
_app_run(tmp_path)
log = tmp_path / "log.json"
put_log(log, {"2026-11-01": [credit(_AT, END_UNIX - 3600)]})
@ -113,6 +121,7 @@ def test_same_day_runnerup_credit_does_not_anchor_the_app_run(tmp_path: Path) ->
def test_no_app_run_means_no_anchor(tmp_path: Path) -> None:
"""Test no app run means no anchor."""
put_tcx(tmp_path, "r.tcx")
log = tmp_path / "log.json"
put_log(log, {"d": [credit(_AT, END_UNIX)]})
@ -122,6 +131,7 @@ def test_no_app_run_means_no_anchor(tmp_path: Path) -> None:
def test_unreadable_or_trackless_app_tcx_cannot_anchor(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test unreadable or trackless app tcx cannot anchor."""
(tmp_path / "bad.tcx").write_text(f"<Author><Name>{APP}</Name></Author>")
log = tmp_path / "log.json"
put_log(log, {"d": [credit(_AT, END_UNIX)]})
@ -133,6 +143,7 @@ def test_unreadable_or_trackless_app_tcx_cannot_anchor(
def test_log_entries_that_cannot_match_are_skipped(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test log entries that cannot match are skipped."""
_app_run(tmp_path)
log = tmp_path / "log.json"
put_log(
@ -160,6 +171,7 @@ def test_log_entries_that_cannot_match_are_skipped(
def test_naive_credit_timestamps_are_read_as_local_time(tmp_path: Path) -> None:
"""Test naive credit timestamps are read as local time."""
_app_run(tmp_path)
log = tmp_path / "log.json"
put_log(log, {"d": [credit("2026-11-01T10:45:00", END_UNIX)]})
@ -168,95 +180,10 @@ def test_naive_credit_timestamps_are_read_as_local_time(tmp_path: Path) -> None:
assert found.tzinfo is not None
def _processed(tmp_path: Path) -> Path:
processed = tmp_path / "processed"
put_tcx(processed, "RunnerUp_2026-08-12-17-42-11_Running.tcx")
put_tcx(processed, "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx")
put_tcx(processed, "RunnerUp_2026-08-14-00-58-45_Running.tcx")
put_tcx(processed, "app_2026-11-01.tcx", author=APP)
return processed
_LISTED = [
"RunnerUp_2026-08-12-17-42-11_Running.tcx",
"RunnerUp_2026-08-14-00-58-45_Running.tcx",
]
def test_marker_valid_counts_device_twins_and_ignores_app_files(
tmp_path: Path,
) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, _LISTED)
check = rt.check_marker(marker, processed)
assert check.ok
assert "2 runs" in check.reason
def test_marker_may_name_the_device_prefixed_copy(tmp_path: Path) -> None:
processed = _processed(tmp_path)
names = ["RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx", _LISTED[1]]
assert rt.check_marker(put_marker(tmp_path, names), processed).ok
def test_marker_missing_is_not_valid(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.WARNING):
check = rt.check_marker(tmp_path / "m.json", tmp_path)
assert not check.ok
assert "has not imported yet" in check.reason
assert "marker problem" in caplog.text
def test_marker_unreadable_and_bad_json(tmp_path: Path) -> None:
folder_as_marker = tmp_path / "m"
folder_as_marker.mkdir()
assert "unreadable" in rt.check_marker(folder_as_marker, tmp_path).reason
bad = tmp_path / "bad.json"
bad.write_text("{nope")
assert "unreadable" in rt.check_marker(bad, tmp_path).reason
@pytest.mark.parametrize(
("body", "fragment"),
[
("[]", "not a JSON object"),
('{"count": 0, "tcx_names": []}', "no completed_at"),
('{"completed_at": "", "count": 0, "tcx_names": []}', "no completed_at"),
('{"completed_at": "t", "count": 0, "tcx_names": "x"}', "not a list"),
('{"completed_at": "t", "count": 0, "tcx_names": [1]}', "not a list"),
('{"completed_at": "t", "count": 3, "tcx_names": ["a"]}', "count 3"),
('{"completed_at": "t", "count": true, "tcx_names": ["a"]}', "count True"),
('{"completed_at": "t", "count": "1", "tcx_names": ["a"]}', "count '1'"),
('{"completed_at": "t", "count": 1, "tcx_names": ["x/a"]}', "bare file"),
('{"completed_at": "t", "count": 1, "tcx_names": [".a"]}', "bare file"),
],
)
def test_marker_shape_errors(tmp_path: Path, body: str, fragment: str) -> None:
marker = tmp_path / "m.json"
marker.write_text(body)
check = rt.check_marker(marker, tmp_path)
assert not check.ok
assert fragment in check.reason
def test_marker_listing_one_run_twice_is_invalid(tmp_path: Path) -> None:
processed = _processed(tmp_path)
twin = "RunnerUp_Pixel_6a_2026-08-12-17-42-11_Running.tcx"
marker = put_marker(tmp_path, [_LISTED[0], twin])
assert "more than once" in rt.check_marker(marker, processed).reason
def test_marker_naming_a_run_not_in_processed_is_invalid(tmp_path: Path) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, [*_LISTED, "RunnerUp_2099-01-01-00-00-00_X.tcx"])
reason = rt.check_marker(marker, processed).reason
assert "not in processed/" in reason
def test_marker_missing_a_processed_run_asks_for_a_reimport(tmp_path: Path) -> None:
processed = _processed(tmp_path)
marker = put_marker(tmp_path, _LISTED[:1])
reason = rt.check_marker(marker, processed).reason
assert "re-run the app's import" in reason
def test_app_run_ends_lists_every_app_run_in_every_folder(tmp_path: Path) -> None:
"""Test app run ends lists every app run in every folder."""
_app_run(tmp_path, "a.tcx")
_app_run(tmp_path, "b.tcx", where="processed")
put_tcx(tmp_path, "r.tcx")
ends = rt.app_run_ends([tmp_path, tmp_path / "processed"])
assert ends == [END_UNIX, END_UNIX]