mirror of
https://github.com/kuhyx/screen-locker.git
synced 2026-11-10 19:27:15 +01:00
Credit: one reward per workout slot, desktop notification, --help, runnerup-watch ignores re-pulls
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018RHwVPyiiQMPh4M3JQwQku
This commit is contained in:
parent
f2b8fdf7ea
commit
aedfdbc7f5
@ -33,6 +33,60 @@ _LOG_MANUAL_FLAG = "--log-manual-workout"
|
||||
_REST_DAY_FLAG = "--declare-rest-day"
|
||||
_BACKFILL_FLAG = "--backfill-workout-ledger"
|
||||
|
||||
# Subcommands that own every argument after them (each has its own argparse,
|
||||
# which also answers its own --help).
|
||||
_TRAILING_ARG_FLAGS = (_LOG_MANUAL_FLAG, _REST_DAY_FLAG, _BACKFILL_FLAG)
|
||||
_MODE_FLAGS = frozenset(
|
||||
{"--status", "--sync-only", "--apply-bonuses", "--production", "--verify-workout"}
|
||||
)
|
||||
_HELP_FLAGS = frozenset({"-h", "--help"})
|
||||
_EXIT_USAGE = 2
|
||||
|
||||
_USAGE = """\
|
||||
usage: python3 -m screen_locker.screen_lock [MODE]
|
||||
|
||||
With no mode: the demo lock screen (Tk). Modes:
|
||||
--production the real lock screen (what workout-locker.service runs)
|
||||
--verify-workout lock screen in verify-only mode
|
||||
--status print this week's workout status and exit
|
||||
--sync-only headless: pull synced workouts, apply credit, exit
|
||||
--apply-bonuses headless: apply newly earned flat bonuses, exit
|
||||
--log-manual-workout ARGS headless manual workout log (see its --help)
|
||||
--declare-rest-day ARGS sign a future rest day (see its --help)
|
||||
--backfill-workout-ledger ARGS
|
||||
backfill the workout ledger (see its --help)
|
||||
-h, --help show this help and exit
|
||||
"""
|
||||
|
||||
|
||||
def _own_args(argv: list[str]) -> list[str]:
|
||||
"""``argv[1:]`` up to the first subcommand that owns the rest."""
|
||||
args = argv[1:]
|
||||
for index, arg in enumerate(args):
|
||||
if arg in _TRAILING_ARG_FLAGS:
|
||||
return args[:index]
|
||||
return args
|
||||
|
||||
|
||||
def _handle_help_and_unknown(argv: list[str]) -> None:
|
||||
"""Exit 0 on ``--help``; exit 2 on an unknown argument.
|
||||
|
||||
Before this, ``--help`` matched no branch and fell through to building
|
||||
the demo lock screen, and a typo'd flag did the same silently.
|
||||
"""
|
||||
own = _own_args(argv)
|
||||
if _HELP_FLAGS & set(own):
|
||||
# The usage on stdout IS the work --help asked for, so exiting 0 here
|
||||
# abandons nothing. SystemExit is raised directly, not via sys.exit:
|
||||
# the suite stubs sys.exit process-wide, and a help request must never
|
||||
# fall through to building the lock screen even then.
|
||||
sys.stdout.write(_USAGE)
|
||||
raise SystemExit(0)
|
||||
unknown = [arg for arg in own if arg not in _MODE_FLAGS]
|
||||
if unknown:
|
||||
sys.stderr.write(f"error: unknown argument(s): {' '.join(unknown)}\n{_USAGE}")
|
||||
sys.exit(_EXIT_USAGE)
|
||||
|
||||
|
||||
def _headless_locker(locker_cls: type[ScreenLocker]) -> ScreenLocker:
|
||||
"""A ScreenLocker with ``__init__`` bypassed, for the no-UI subcommands.
|
||||
@ -49,6 +103,7 @@ def _headless_locker(locker_cls: type[ScreenLocker]) -> ScreenLocker:
|
||||
|
||||
def main(locker_cls: type[ScreenLocker], argv: list[str]) -> None:
|
||||
"""Dispatch on ``argv`` and run the requested mode."""
|
||||
_handle_help_and_unknown(argv)
|
||||
# Configure logging for EVERY mode, not just --sync-only. This used to sit
|
||||
# inside the --sync-only branch, so `--production` -- the mode systemd
|
||||
# actually runs -- had no handler and fell back to lastResort, which drops
|
||||
|
||||
125
screen_locker/_credit_notify.py
Normal file
125
screen_locker/_credit_notify.py
Normal file
@ -0,0 +1,125 @@
|
||||
"""Desktop notification the moment a workout earns its credit.
|
||||
|
||||
The reward is the point: a workout should be *felt* as paid the second the PC
|
||||
credits it, not discovered at the next look at the status window. Every path
|
||||
that credits a workout ends in
|
||||
:meth:`~screen_locker._workout_credit.WorkoutCreditMixin._apply_credit_for_written_entry`,
|
||||
which calls :func:`notify_workout_credit` once per newly earned credit slot.
|
||||
|
||||
The gaming figure is ``earned_time.resolve`` over screen-locker's own answers
|
||||
(:func:`screen_locker._shutdown_target.gather` -- the same sum the daily reset
|
||||
writes), so no budget math lives here. steam-backlog-enforcer resolves the
|
||||
same registry with its own transports and holds a day's high-water mark, so
|
||||
its figure can sit above this one, never below on a normal day.
|
||||
|
||||
A notification is a courtesy, never part of crediting: any failure (no DBus,
|
||||
no notify-send, a hang) is logged at warning and swallowed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from screen_locker._day import today_str
|
||||
from screen_locker._earned import hhmm, span
|
||||
from screen_locker._grace_floor import tonight_minutes
|
||||
from screen_locker._shutdown_target import gather
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
_logger: Final = logging.getLogger(__name__)
|
||||
|
||||
TITLE: Final = "Workout credited \N{FLEXED BICEPS}"
|
||||
_NOTIFY_TIMEOUT_SECONDS: Final = 5
|
||||
_MINUTES_PER_HOUR: Final = 60.0
|
||||
|
||||
|
||||
def gaming_minutes_today(log_file: Path | None) -> int | None:
|
||||
"""Today's gaming budget per the earned_time registry, or ``None``.
|
||||
|
||||
``None`` (logged) when the inputs cannot be read; the notification then
|
||||
shows the shutdown time alone.
|
||||
"""
|
||||
try:
|
||||
target = gather(date.fromisoformat(today_str()), log_file)
|
||||
except (OSError, ValueError, KeyError) as exc:
|
||||
_logger.warning(
|
||||
"Could not compute today's gaming budget for the credit "
|
||||
"notification (%s) — showing the shutdown time only",
|
||||
exc,
|
||||
)
|
||||
return None
|
||||
return int(target.resolution.gaming_minutes)
|
||||
|
||||
|
||||
def credit_body(
|
||||
before: tuple[int, int, int] | None,
|
||||
after: tuple[int, int, int] | None,
|
||||
gaming_minutes: int | None,
|
||||
) -> str:
|
||||
"""``Shutdown 23:00 (+2h) · gaming 7.0h`` -- whichever parts are known.
|
||||
|
||||
``before``/``after`` are the shutdown config around the credit; the delta
|
||||
is tonight's actual move, so a capped or rest-day credit shows ``+0h``.
|
||||
"""
|
||||
parts: list[str] = []
|
||||
if after is None:
|
||||
_logger.warning(
|
||||
"Shutdown config unreadable after a workout credit — the "
|
||||
"notification shows no shutdown time"
|
||||
)
|
||||
else:
|
||||
day = date.fromisoformat(today_str())
|
||||
tonight = tonight_minutes(after, day)
|
||||
text = f"Shutdown {hhmm(tonight)}"
|
||||
if before is not None:
|
||||
moved = max(0, tonight - tonight_minutes(before, day))
|
||||
text += f" (+{span(moved)})"
|
||||
parts.append(text)
|
||||
if gaming_minutes is not None:
|
||||
parts.append(f"gaming {gaming_minutes / _MINUTES_PER_HOUR:.1f}h")
|
||||
return " \N{MIDDLE DOT} ".join(parts) or "Credit applied (details unavailable)"
|
||||
|
||||
|
||||
def send_notification(title: str, body: str) -> bool:
|
||||
"""Show a desktop notification via ``notify-send``; False (logged) on failure."""
|
||||
binary = shutil.which("notify-send")
|
||||
if binary is None:
|
||||
_logger.warning("notify-send not found — no desktop notification: %s", body)
|
||||
return False
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[binary, "--app-name=screen-locker", title, body],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=_NOTIFY_TIMEOUT_SECONDS,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
_logger.warning("Desktop notification failed (%s): %s", exc, body)
|
||||
return False
|
||||
if result.returncode != 0:
|
||||
_logger.warning(
|
||||
"notify-send exited %d (%s) — no desktop notification: %s",
|
||||
result.returncode,
|
||||
result.stderr.strip() or "no stderr; is a DBus session reachable?",
|
||||
body,
|
||||
)
|
||||
return False
|
||||
_logger.info("Desktop notification: %s — %s", title, body)
|
||||
return True
|
||||
|
||||
|
||||
def notify_workout_credit(
|
||||
before: tuple[int, int, int] | None,
|
||||
after: tuple[int, int, int] | None,
|
||||
log_file: Path | None,
|
||||
) -> bool:
|
||||
"""Notify a newly earned workout credit. Never raises."""
|
||||
body = credit_body(before, after, gaming_minutes_today(log_file))
|
||||
return send_notification(TITLE, body)
|
||||
86
screen_locker/_credit_slot.py
Normal file
86
screen_locker/_credit_slot.py
Normal file
@ -0,0 +1,86 @@
|
||||
"""Which credit a newly written workout earns, decided by slot like the reset.
|
||||
|
||||
The live credit path used to ask "was any counted workout already logged
|
||||
today?" -- so the two StrongLifts ingestion paths of ONE session
|
||||
(``pc_workout_verified`` from the session sync, ``phone_verified`` from the
|
||||
early-bird auto-upgrade) earned the first-unit +2h and then a further-unit +1h
|
||||
(2026-10-09; also 07-17 and 07-27 in ``duplicate_credits.json``). The daily
|
||||
reset counts the same day with :func:`~screen_locker._weekly_check.count_day_credits`,
|
||||
which collapses them onto one slot. Deciding the live credit with that same
|
||||
counter makes the two agree by construction.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import ExitStack, contextmanager
|
||||
import enum
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from screen_locker._bonus_lock import bonus_lock
|
||||
from screen_locker._weekly_check import count_day_credits
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Mapping
|
||||
from pathlib import Path
|
||||
|
||||
_logger: Final = logging.getLogger(__name__)
|
||||
|
||||
# count_day_credits only uses the date to build slot keys, and both counts
|
||||
# below are taken over the same day, so any fixed label is equivalent.
|
||||
_SAME_DAY: Final = "day"
|
||||
|
||||
|
||||
class CreditSlot(enum.Enum):
|
||||
"""What the newest entry of a day earns."""
|
||||
|
||||
NONE = "earns no credit (not a counted workout type)"
|
||||
FIRST = "first credit slot of the day"
|
||||
FURTHER = "a new, distinct credit slot after the first"
|
||||
OCCUPIED = "a slot an earlier entry of the day already credited"
|
||||
|
||||
|
||||
def credit_slot(
|
||||
prior_entries: list[dict[str, Any]], workout_data: Mapping[str, Any]
|
||||
) -> CreditSlot:
|
||||
"""Classify ``workout_data`` against the day's entries written before it.
|
||||
|
||||
Args:
|
||||
prior_entries: The day's log entries before the new one, as returned
|
||||
by the write chokepoint (snapshotted under the log lock).
|
||||
workout_data: The new entry's ``workout_data``.
|
||||
"""
|
||||
new_entry = {"workout_data": dict(workout_data)}
|
||||
day = [e for e in prior_entries if isinstance(e, dict)]
|
||||
before = count_day_credits(_SAME_DAY, day)
|
||||
with_new = count_day_credits(_SAME_DAY, [*day, new_entry])
|
||||
if with_new == before:
|
||||
# No new slot: either it earns nothing at all, or its slot is taken.
|
||||
return CreditSlot.OCCUPIED if _counts_alone(new_entry) else CreditSlot.NONE
|
||||
return CreditSlot.FIRST if before == 0 else CreditSlot.FURTHER
|
||||
|
||||
|
||||
def _counts_alone(entry: Mapping[str, Any]) -> bool:
|
||||
"""Whether ``entry`` would earn a credit on an otherwise empty day."""
|
||||
return count_day_credits(_SAME_DAY, [entry]) == 1
|
||||
|
||||
|
||||
@contextmanager
|
||||
def locked_shutdown_state(state_file: Path) -> Iterator[None]:
|
||||
"""Hold :func:`bonus_lock` on ``state_file``; on failure, warn and go on.
|
||||
|
||||
The reset and the flat-bonus pass hold the same lock, so a credit can no
|
||||
longer lose its minutes to their read-add-write. Never nest inside them:
|
||||
a second flock on a new fd in the same process deadlocks.
|
||||
"""
|
||||
with ExitStack() as stack:
|
||||
try:
|
||||
stack.enter_context(bonus_lock(state_file))
|
||||
except OSError as exc:
|
||||
_logger.warning(
|
||||
"Could not lock %s (%s) — applying the workout credit without "
|
||||
"it; a reset running at the same moment could overwrite it",
|
||||
state_file,
|
||||
exc,
|
||||
)
|
||||
yield
|
||||
@ -146,7 +146,8 @@ def plan_absorb(minutes: int, state_file: Path | None = None) -> tuple[int, int]
|
||||
return max(0, minutes - lift), max(0, lift - minutes)
|
||||
|
||||
|
||||
def _tonight(config: tuple[int, int, int], day: date) -> int:
|
||||
def tonight_minutes(config: tuple[int, int, int], day: date) -> int:
|
||||
"""``day``'s shutdown from a (mon_wed, thu_sun, morning_end) config."""
|
||||
mon_wed, thu_sun, _ = config
|
||||
return mon_wed if day.weekday() in _MON_WED else thu_sun
|
||||
|
||||
@ -184,7 +185,7 @@ class GraceFloorMixin:
|
||||
if config is None:
|
||||
_logger.warning("Grace floor: shutdown config unreadable; not applied")
|
||||
return False
|
||||
current = _tonight(config, day)
|
||||
current = tonight_minutes(config, day)
|
||||
if current >= floor:
|
||||
return False
|
||||
mon_wed, thu_sun, morning = config
|
||||
|
||||
@ -16,7 +16,9 @@ from typing import TYPE_CHECKING
|
||||
|
||||
import earned_time
|
||||
|
||||
from screen_locker import _sick_tracker
|
||||
from screen_locker import _credit_notify, _sick_tracker
|
||||
from screen_locker._constants import SHUTDOWN_BASE_FILE
|
||||
from screen_locker._credit_slot import CreditSlot, credit_slot, locked_shutdown_state
|
||||
from screen_locker._day import today_str
|
||||
from screen_locker._earned import extra_minutes
|
||||
from screen_locker._rest_day import is_rest_day
|
||||
@ -123,10 +125,27 @@ class WorkoutCreditMixin:
|
||||
verified ones; the manual-workout rate budget is the only limiter.
|
||||
"""
|
||||
weekly_count = count_weekly_workouts(self.log_file)
|
||||
first_counted_today = not any(
|
||||
entry.get("workout_data", {}).get("type") in COUNTED_WORKOUT_TYPES
|
||||
for entry in prior_entries
|
||||
# Slot-based, exactly like the daily reset (credit_key): the two
|
||||
# StrongLifts ingestion paths of ONE session share a slot, so the
|
||||
# second copy earns nothing. Deciding on "any counted type before
|
||||
# it" instead paid one session +2h and then +1h (2026-10-09).
|
||||
slot = credit_slot(prior_entries, self.workout_data)
|
||||
if slot is CreditSlot.OCCUPIED:
|
||||
_logger.warning(
|
||||
"Workout %s (%s) shares its credit slot with an entry already "
|
||||
"credited today — no shutdown time, no debt change; the daily "
|
||||
"reset counts the two as one workout too",
|
||||
self.workout_data.get("type", ""),
|
||||
self.workout_data.get("source", ""),
|
||||
)
|
||||
return WorkoutCreditResult(
|
||||
shutdown_adjusted=False,
|
||||
new_debt=None,
|
||||
extra_bonus_delta=0,
|
||||
weekly_count=weekly_count,
|
||||
already_counted_today=True,
|
||||
)
|
||||
first_counted_today = slot is CreditSlot.FIRST
|
||||
|
||||
shutdown_adjusted = False
|
||||
extra_bonus_delta = 0
|
||||
@ -138,20 +157,26 @@ class WorkoutCreditMixin:
|
||||
if rest_paid:
|
||||
_logger.info("Rest day: its first-unit bonus is already in tonight's time")
|
||||
extra = extra_minutes(earned_time.WORKOUT, day)
|
||||
# Same lock as the daily reset and the flat-bonus pass: all three
|
||||
# read-add-write the shutdown config and the grace lift.
|
||||
with locked_shutdown_state(SHUTDOWN_BASE_FILE):
|
||||
before = self._read_shutdown_config()
|
||||
if first_counted_today and not rest_paid:
|
||||
shutdown_adjusted = self._try_adjust_shutdown_for_workout()
|
||||
elif (
|
||||
not first_counted_today
|
||||
slot is CreditSlot.FURTHER
|
||||
and extra
|
||||
and self.workout_data.get("type") in COUNTED_WORKOUT_TYPES
|
||||
and before
|
||||
and self._adjust_shutdown_time_by(extra)
|
||||
):
|
||||
old_cfg = self._read_shutdown_config()
|
||||
if old_cfg and self._adjust_shutdown_time_by(extra):
|
||||
new_cfg = self._read_shutdown_config()
|
||||
if new_cfg:
|
||||
extra_bonus_delta = new_cfg[1] - old_cfg[1]
|
||||
extra_bonus_delta = new_cfg[1] - before[1]
|
||||
after = self._read_shutdown_config()
|
||||
|
||||
new_debt = self._clear_debt_on_verified_workout()
|
||||
if slot is not CreditSlot.NONE:
|
||||
self._notify_credit(before, after)
|
||||
|
||||
return WorkoutCreditResult(
|
||||
shutdown_adjusted=shutdown_adjusted,
|
||||
@ -161,6 +186,26 @@ class WorkoutCreditMixin:
|
||||
already_counted_today=False,
|
||||
)
|
||||
|
||||
def _notify_credit(
|
||||
self,
|
||||
before: tuple[int, int, int] | None,
|
||||
after: tuple[int, int, int] | None,
|
||||
) -> None:
|
||||
"""Pop the "Workout credited" notification; never fails the credit.
|
||||
|
||||
Runs after the credit is fully applied. Anything it raises -- a
|
||||
malformed ledger inside the budget sum, a DBus oddity -- would
|
||||
otherwise escape into the ingest loop and leave the rest of that
|
||||
batch uncredited, so every exception is logged and dropped here.
|
||||
"""
|
||||
try:
|
||||
_credit_notify.notify_workout_credit(before, after, self.log_file)
|
||||
except Exception:
|
||||
_logger.exception(
|
||||
"Workout credit notification failed — the credit itself is "
|
||||
"applied; only the desktop notification is missing"
|
||||
)
|
||||
|
||||
def _apply_workout_credit(self) -> WorkoutCreditResult:
|
||||
"""Append ``workout_data`` and apply its reward, scaled to the day.
|
||||
|
||||
|
||||
@ -85,3 +85,26 @@ def _no_real_runnerup_webdav_dir(tmp_path: Path) -> Iterator[None]:
|
||||
absent = (tmp_path / "no-runnerup-webdav",)
|
||||
with patch("screen_locker._runnerup_verification.RUNNERUP_WEBDAV_DIRS", absent):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _no_desktop_notifications(tmp_path: Path) -> Iterator[None]:
|
||||
"""Keep the credit path off the real desktop and the real lock file.
|
||||
|
||||
Every workout credit ends in ``_credit_notify.send_notification``, which
|
||||
runs ``notify-send`` -- without this the suite would pop hundreds of
|
||||
"Workout credited" bubbles on the developer's screen, and its budget
|
||||
sum would read ledgers and log fail-closed warnings into caplog -- so
|
||||
``notify_workout_credit`` itself is stubbed. The credit also takes
|
||||
``bonus_lock`` on ``_workout_credit.SHUTDOWN_BASE_FILE``, bound at import
|
||||
time, so that is redirected too. Tests of the notifier itself capture the
|
||||
real function at import time.
|
||||
"""
|
||||
with (
|
||||
patch("screen_locker._credit_notify.notify_workout_credit", return_value=True),
|
||||
patch(
|
||||
"screen_locker._workout_credit.SHUTDOWN_BASE_FILE",
|
||||
tmp_path / "credit_shutdown_base.json",
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
@ -69,9 +69,9 @@ if TYPE_CHECKING:
|
||||
from collections.abc import Generator, Iterator
|
||||
|
||||
|
||||
# Autouse fixtures that keep this suite off real user state live here; see
|
||||
# that module for why the two seams are different.
|
||||
pytest_plugins = ("screen_locker.tests._real_state_fixtures",)
|
||||
# Real-state isolation; sync fixtures listed too: `-o addopts=""` drops `-p`.
|
||||
_SYNC = "screen_locker.tests.test_sync_fixtures"
|
||||
pytest_plugins = ("screen_locker.tests._real_state_fixtures", _SYNC)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
||||
73
screen_locker/tests/test_cli_help.py
Normal file
73
screen_locker/tests/test_cli_help.py
Normal file
@ -0,0 +1,73 @@
|
||||
"""``--help`` answers and exits; an unknown flag is a usage error.
|
||||
|
||||
Before this, both fell through to building the demo lock screen. Split from
|
||||
``test_cli.py`` for the 250-line cap. ``sys.exit`` is a no-op suite-wide (see
|
||||
conftest), so each test installs one that really raises ``SystemExit``.
|
||||
"""
|
||||
# pylint: disable=protected-access
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, NoReturn, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from screen_locker import _cli
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from screen_locker.screen_lock import ScreenLocker
|
||||
|
||||
|
||||
def _really_exit(code: int = 0) -> NoReturn:
|
||||
raise SystemExit(code)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _exiting_sys_exit(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(_cli.sys, "exit", _really_exit)
|
||||
|
||||
|
||||
class _NeverBuilt:
|
||||
"""A locker class whose construction would mean the guard failed."""
|
||||
|
||||
def __init__(self, *_: object, **__: object) -> None:
|
||||
msg = "the lock screen must not be built"
|
||||
raise AssertionError(msg)
|
||||
|
||||
|
||||
def _main(argv: list[str]) -> None:
|
||||
locker_cls = cast("type[ScreenLocker]", _NeverBuilt)
|
||||
_cli.main(locker_cls, ["screen_lock.py", *argv])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flag", ["--help", "-h"])
|
||||
def test_help_prints_usage_and_exits_zero(
|
||||
flag: str, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
"""Help goes to stdout and the process exits 0 before any UI."""
|
||||
with pytest.raises(SystemExit) as exited:
|
||||
_main(["--production", flag])
|
||||
assert exited.value.code == 0
|
||||
out = capsys.readouterr().out
|
||||
assert out.startswith("usage: python3 -m screen_locker.screen_lock")
|
||||
|
||||
|
||||
def test_unknown_flag_is_exit_two_with_usage(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""A typo is named on stderr with the usage, and exits 2."""
|
||||
with pytest.raises(SystemExit) as exited:
|
||||
_main(["--prodution"])
|
||||
assert exited.value.code == 2
|
||||
err = capsys.readouterr().err
|
||||
assert "unknown argument(s): --prodution" in err
|
||||
assert "usage:" in err
|
||||
|
||||
|
||||
def test_subcommand_args_are_not_judged_here() -> None:
|
||||
"""Everything after a trailing-arg subcommand belongs to its own parser."""
|
||||
assert (
|
||||
_cli._own_args(["screen_lock.py", "--log-manual-workout", "--help", "--weird"])
|
||||
== []
|
||||
)
|
||||
assert _cli._own_args(["screen_lock.py", "--status"]) == ["--status"]
|
||||
187
screen_locker/tests/test_credit_notify.py
Normal file
187
screen_locker/tests/test_credit_notify.py
Normal file
@ -0,0 +1,187 @@
|
||||
"""The "Workout credited" desktop notification and every way it can fail.
|
||||
|
||||
The autouse ``_no_desktop_notifications`` fixture stubs
|
||||
``notify_workout_credit`` for the whole suite; the real one is captured here
|
||||
at import time, before any fixture runs.
|
||||
"""
|
||||
# pylint: disable=protected-access
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from screen_locker import _credit_notify
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
_REAL_NOTIFY = _credit_notify.notify_workout_credit
|
||||
_FRIDAY = "2026-10-09" # Thu-Sun: tonight is the second config field
|
||||
_MOD = "screen_locker._credit_notify"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _fixed_day() -> Iterator[None]:
|
||||
with patch(f"{_MOD}.today_str", return_value=_FRIDAY):
|
||||
yield
|
||||
|
||||
|
||||
def _warned(caplog: pytest.LogCaptureFixture, needle: str) -> bool:
|
||||
return any(
|
||||
r.levelno >= logging.WARNING and needle in r.getMessage()
|
||||
for r in caplog.records
|
||||
)
|
||||
|
||||
|
||||
class TestGamingMinutes:
|
||||
"""The budget figure, or None when its inputs cannot be read."""
|
||||
|
||||
def test_returns_the_registry_figure(self) -> None:
|
||||
"""gather's resolution is shown as whole minutes."""
|
||||
target = SimpleNamespace(resolution=SimpleNamespace(gaming_minutes=420.0))
|
||||
with patch(f"{_MOD}.gather", return_value=target):
|
||||
assert _credit_notify.gaming_minutes_today(None) == 420
|
||||
|
||||
def test_unreadable_inputs_warn_and_return_none(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""A broken ledger costs the figure, not the notification."""
|
||||
with patch(f"{_MOD}.gather", side_effect=OSError("ledger gone")):
|
||||
assert _credit_notify.gaming_minutes_today(None) is None
|
||||
assert _warned(caplog, "ledger gone")
|
||||
|
||||
|
||||
class TestCreditBody:
|
||||
"""Whichever parts are known end up in the body."""
|
||||
|
||||
def test_full_body(self) -> None:
|
||||
"""Shutdown time, tonight's move and the gaming budget."""
|
||||
body = _credit_notify.credit_body((0, 1200, 0), (0, 1320, 0), 420)
|
||||
assert body == "Shutdown 22:00 (+2h) \N{MIDDLE DOT} gaming 7.0h"
|
||||
|
||||
def test_a_capped_credit_shows_plus_zero(self) -> None:
|
||||
"""No backwards move is ever shown: the delta floors at zero."""
|
||||
body = _credit_notify.credit_body((0, 1380, 0), (0, 1320, 0), None)
|
||||
assert body == "Shutdown 22:00 (+0h)"
|
||||
|
||||
def test_unknown_before_drops_the_delta(self) -> None:
|
||||
"""Without the old config there is no move to show."""
|
||||
assert _credit_notify.credit_body(None, (0, 1320, 0), None) == "Shutdown 22:00"
|
||||
|
||||
def test_unreadable_after_warns_and_shows_gaming_only(
|
||||
self, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""An unreadable config is said out loud, not silently omitted."""
|
||||
assert _credit_notify.credit_body(None, None, 90) == "gaming 1.5h"
|
||||
assert _warned(caplog, "unreadable")
|
||||
|
||||
def test_nothing_known_still_says_something(self) -> None:
|
||||
"""An empty body would read as a glitch; say the credit happened."""
|
||||
body = _credit_notify.credit_body(None, None, None)
|
||||
assert body == "Credit applied (details unavailable)"
|
||||
|
||||
|
||||
class TestSendNotification:
|
||||
"""notify-send's every failure is a warning and a False."""
|
||||
|
||||
def test_missing_binary(self, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""No notify-send on PATH."""
|
||||
with patch(f"{_MOD}.shutil.which", return_value=None):
|
||||
assert not _credit_notify.send_notification("t", "b")
|
||||
assert _warned(caplog, "notify-send not found")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error", [OSError("exec"), subprocess.TimeoutExpired("notify-send", 5)]
|
||||
)
|
||||
def test_spawn_failure_or_hang(
|
||||
self, error: Exception, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""An exec error or a DBus hang past the timeout."""
|
||||
with (
|
||||
patch(f"{_MOD}.shutil.which", return_value="/usr/bin/notify-send"),
|
||||
patch(f"{_MOD}.subprocess.run", side_effect=error),
|
||||
):
|
||||
assert not _credit_notify.send_notification("t", "b")
|
||||
assert _warned(caplog, "Desktop notification failed")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("stderr", "shown"),
|
||||
[("no bus", "no bus"), (" ", "is a DBus session reachable")],
|
||||
)
|
||||
def test_nonzero_exit(
|
||||
self, stderr: str, shown: str, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""A failing notify-send names its stderr, or the likely cause."""
|
||||
done = subprocess.CompletedProcess([], 1, stdout="", stderr=stderr)
|
||||
with (
|
||||
patch(f"{_MOD}.shutil.which", return_value="/usr/bin/notify-send"),
|
||||
patch(f"{_MOD}.subprocess.run", return_value=done),
|
||||
):
|
||||
assert not _credit_notify.send_notification("t", "b")
|
||||
assert _warned(caplog, shown)
|
||||
|
||||
def test_success(self) -> None:
|
||||
"""Exit 0 is the only True."""
|
||||
done = subprocess.CompletedProcess([], 0, stdout="", stderr="")
|
||||
with (
|
||||
patch(f"{_MOD}.shutil.which", return_value="/usr/bin/notify-send"),
|
||||
patch(f"{_MOD}.subprocess.run", return_value=done) as run,
|
||||
):
|
||||
assert _credit_notify.send_notification("Title", "Body")
|
||||
assert run.call_args.args[0] == [
|
||||
"/usr/bin/notify-send",
|
||||
"--app-name=screen-locker",
|
||||
"Title",
|
||||
"Body",
|
||||
]
|
||||
|
||||
|
||||
def test_notify_workout_credit_sends_the_composed_body() -> None:
|
||||
"""The real entry point composes the body and sends it once."""
|
||||
with (
|
||||
patch(f"{_MOD}.gaming_minutes_today", return_value=60),
|
||||
patch(f"{_MOD}.send_notification", return_value=True) as send,
|
||||
):
|
||||
assert _REAL_NOTIFY((0, 1200, 0), (0, 1260, 0), None)
|
||||
send.assert_called_once_with(
|
||||
_credit_notify.TITLE, "Shutdown 21:00 (+1h) \N{MIDDLE DOT} gaming 1.0h"
|
||||
)
|
||||
|
||||
|
||||
class TestCreditSideOfTheNotifier:
|
||||
"""The credit mixin's guards around the notifier."""
|
||||
|
||||
def test_a_raising_notifier_never_fails_the_credit(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Anything the notifier raises is logged at ERROR and dropped."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with patch(f"{_MOD}.notify_workout_credit", side_effect=KeyError("x")):
|
||||
locker._notify_credit(None, None)
|
||||
assert any(
|
||||
r.levelno == logging.ERROR and "notification failed" in r.getMessage()
|
||||
for r in caplog.records
|
||||
)
|
||||
|
||||
def test_uncounted_type_never_moves_the_shutdown(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
"""The first-unit push refuses a type that earns no credit."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
locker.workout_data = {"type": "early_bird"}
|
||||
adjust = MagicMock(return_value=True)
|
||||
object.__setattr__(locker, "_adjust_shutdown_time_later", adjust)
|
||||
assert locker._try_adjust_shutdown_for_workout() is False
|
||||
adjust.assert_not_called()
|
||||
165
screen_locker/tests/test_credit_slot.py
Normal file
165
screen_locker/tests/test_credit_slot.py
Normal file
@ -0,0 +1,165 @@
|
||||
"""Slot-based live workout credit: one slot pays once, a new slot pays again.
|
||||
|
||||
The two StrongLifts ingestion paths of ONE session used to earn +2h and then
|
||||
+1h (2026-10-09). These tests drive the real shutdown arithmetic over a fake
|
||||
config so the minute figures, not just the branch taken, are pinned.
|
||||
"""
|
||||
# pylint: disable=protected-access
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker import _credit_slot
|
||||
from screen_locker._credit_slot import CreditSlot, credit_slot, locked_shutdown_state
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from screen_locker._workout_credit import WorkoutCreditResult
|
||||
from screen_locker.screen_lock import ScreenLocker
|
||||
|
||||
_START = (1200, 1200, 300) # 20:00 / 20:00 / 05:00 -- well under the ceiling
|
||||
|
||||
|
||||
def _entry(wtype: str) -> dict[str, dict[str, str]]:
|
||||
return {"workout_data": {"type": wtype}}
|
||||
|
||||
|
||||
class TestCreditSlot:
|
||||
"""credit_slot classifies the new entry against the day before it."""
|
||||
|
||||
def test_first_counted_entry_of_the_day(self) -> None:
|
||||
"""An empty day: the first counted workout opens the first slot."""
|
||||
assert credit_slot([], {"type": "phone_verified"}) is CreditSlot.FIRST
|
||||
|
||||
def test_second_stronglifts_path_shares_the_slot(self) -> None:
|
||||
"""pc_workout_verified then phone_verified: one session, one slot."""
|
||||
prior = [_entry("pc_workout_verified")]
|
||||
assert credit_slot(prior, {"type": "phone_verified"}) is CreditSlot.OCCUPIED
|
||||
|
||||
def test_a_run_after_stronglifts_is_a_further_slot(self) -> None:
|
||||
"""A RunnerUp run is a separate session from the StrongLifts one."""
|
||||
prior = [_entry("phone_verified")]
|
||||
assert credit_slot(prior, {"type": "runnerup_verified"}) is CreditSlot.FURTHER
|
||||
|
||||
def test_uncounted_type_earns_nothing(self) -> None:
|
||||
"""early_bird is not a counted workout type at all."""
|
||||
assert credit_slot([], {"type": "early_bird"}) is CreditSlot.NONE
|
||||
|
||||
def test_non_dict_prior_entries_are_ignored(self) -> None:
|
||||
"""A corrupt prior entry neither crashes nor occupies a slot."""
|
||||
prior: list = ["garbage", None]
|
||||
assert credit_slot(prior, {"type": "phone_verified"}) is CreditSlot.FIRST
|
||||
|
||||
|
||||
class TestLockedShutdownState:
|
||||
"""The credit takes bonus_lock, and degrades loudly when it cannot."""
|
||||
|
||||
def test_lock_failure_warns_and_still_runs_the_body(
|
||||
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""An unlockable state file must not drop the credit -- only warn."""
|
||||
ran: list[bool] = []
|
||||
with (
|
||||
patch.object(_credit_slot, "bonus_lock", side_effect=OSError("ro fs")),
|
||||
caplog.at_level(logging.WARNING, logger=_credit_slot.__name__),
|
||||
locked_shutdown_state(tmp_path / "base.json"),
|
||||
):
|
||||
ran.append(True)
|
||||
assert ran == [True]
|
||||
assert any(
|
||||
r.levelno == logging.WARNING and "ro fs" in r.getMessage()
|
||||
for r in caplog.records
|
||||
)
|
||||
|
||||
def test_lock_success_runs_the_body(self, tmp_path: Path) -> None:
|
||||
"""The happy path holds the real lock around the body."""
|
||||
ran: list[bool] = []
|
||||
with locked_shutdown_state(tmp_path / "base.json"):
|
||||
ran.append(True)
|
||||
assert ran == [True]
|
||||
|
||||
|
||||
def _wired_locker(mock_tk: MagicMock, tmp_path: Path, wtype: str) -> ScreenLocker:
|
||||
"""A locker over an in-memory shutdown config, real adjust arithmetic."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
locker.workout_data = {"type": wtype}
|
||||
config = {"value": _START}
|
||||
|
||||
def write(mw: int, ts: int, morning: int, *, restore: bool) -> bool:
|
||||
assert restore
|
||||
config["value"] = (mw, ts, morning)
|
||||
return True
|
||||
|
||||
object.__setattr__(
|
||||
locker, "_read_shutdown_config", MagicMock(side_effect=lambda: config["value"])
|
||||
)
|
||||
object.__setattr__(locker, "_write_shutdown_config", MagicMock(side_effect=write))
|
||||
object.__setattr__(
|
||||
locker, "_clear_debt_on_verified_workout", MagicMock(return_value=None)
|
||||
)
|
||||
return locker
|
||||
|
||||
|
||||
def _credit(locker: ScreenLocker, prior: list) -> tuple[WorkoutCreditResult, MagicMock]:
|
||||
with (
|
||||
patch("screen_locker._shutdown.plan_absorb", side_effect=lambda m: (m, 0)),
|
||||
patch("screen_locker._shutdown.save_lift"),
|
||||
patch("screen_locker._workout_credit.is_rest_day", return_value=False),
|
||||
patch("screen_locker._workout_credit.count_weekly_workouts", return_value=3),
|
||||
patch(
|
||||
"screen_locker._workout_credit._credit_notify.notify_workout_credit"
|
||||
) as notify,
|
||||
):
|
||||
result = locker._apply_credit_for_written_entry(prior)
|
||||
return result, notify
|
||||
|
||||
|
||||
class TestSlotCreditMinutes:
|
||||
"""The minutes each slot class actually moves the shutdown."""
|
||||
|
||||
def test_first_slot_earns_two_hours(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
"""First counted workout of the day: shutdown +120 min, notified."""
|
||||
locker = _wired_locker(mock_tk, tmp_path, "phone_verified")
|
||||
result, notify = _credit(locker, [])
|
||||
assert result.shutdown_adjusted is True
|
||||
assert locker._read_shutdown_config() == (1320, 1320, 300)
|
||||
notify.assert_called_once_with(_START, (1320, 1320, 300), locker.log_file)
|
||||
|
||||
def test_distinct_slot_earns_one_hour(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
"""A run after StrongLifts: +60 min, reported as the extra delta."""
|
||||
locker = _wired_locker(mock_tk, tmp_path, "runnerup_verified")
|
||||
result, notify = _credit(locker, [_entry("phone_verified")])
|
||||
assert result.extra_bonus_delta == 60
|
||||
assert locker._read_shutdown_config() == (1260, 1260, 300)
|
||||
notify.assert_called_once()
|
||||
|
||||
def test_same_slot_earns_nothing_and_warns(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""The second StrongLifts path: 0 min, a WARNING, no notification."""
|
||||
locker = _wired_locker(mock_tk, tmp_path, "phone_verified")
|
||||
with caplog.at_level(logging.WARNING, logger="screen_locker._workout_credit"):
|
||||
result, notify = _credit(locker, [_entry("pc_workout_verified")])
|
||||
assert result.already_counted_today is True
|
||||
assert result.extra_bonus_delta == 0
|
||||
assert locker._read_shutdown_config() == _START # untouched
|
||||
notify.assert_not_called()
|
||||
assert any(
|
||||
r.levelno == logging.WARNING and "shares its credit slot" in r.getMessage()
|
||||
for r in caplog.records
|
||||
)
|
||||
@ -111,7 +111,9 @@ class TestApplyWorkoutCredit:
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_read_shutdown_config",
|
||||
MagicMock(side_effect=[(1260, 1260, 300), (1320, 1320, 300)]),
|
||||
MagicMock(
|
||||
side_effect=[(1260, 1260, 300), (1320, 1320, 300), (1320, 1320, 300)]
|
||||
),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker, "_adjust_shutdown_time_by", MagicMock(return_value=True)
|
||||
@ -135,12 +137,12 @@ class TestApplyWorkoutCredit:
|
||||
) -> None:
|
||||
"""An additional same-day VERIFIED workout earns the +1h extra bonus.
|
||||
|
||||
prior_entries already holds a counted workout, so this isn't the first
|
||||
of the day; because the new type is verified it takes the +1h branch,
|
||||
and extra_bonus_delta reflects the shutdown-config minute delta.
|
||||
prior_entries already holds a StrongLifts session; a RunnerUp run is a
|
||||
distinct credit slot, so it takes the +1h branch, and
|
||||
extra_bonus_delta reflects the shutdown-config minute delta.
|
||||
"""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
locker.workout_data = {"type": "phone_verified"}
|
||||
locker.workout_data = {"type": "runnerup_verified"}
|
||||
prior = [{"workout_data": {"type": "phone_verified"}}]
|
||||
object.__setattr__(
|
||||
locker,
|
||||
@ -153,7 +155,9 @@ class TestApplyWorkoutCredit:
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_read_shutdown_config",
|
||||
MagicMock(side_effect=[(1260, 1260, 300), (1320, 1320, 300)]),
|
||||
MagicMock(
|
||||
side_effect=[(1260, 1260, 300), (1320, 1320, 300), (1320, 1320, 300)]
|
||||
),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker, "_adjust_shutdown_time_by", MagicMock(return_value=True)
|
||||
|
||||
@ -8,7 +8,7 @@ from __future__ import annotations
|
||||
|
||||
import pathlib
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker._log_mixin import RecordResult
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
@ -71,7 +71,7 @@ class TestApplyWorkoutCredit:
|
||||
"""A workout whose type isn't in COUNTED_WORKOUT_TYPES (e.g.
|
||||
``early_bird``) and isn't the first counted entry of the day →
|
||||
neither the base push nor the +1h branch fires; the shutdown config
|
||||
is never even consulted."""
|
||||
is never adjusted and no credit notification goes out."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
locker.workout_data = {"type": "early_bird"}
|
||||
prior = [{"workout_data": {"type": "phone_verified"}}]
|
||||
@ -88,13 +88,16 @@ class TestApplyWorkoutCredit:
|
||||
locker, "_clear_debt_on_verified_workout", MagicMock(return_value=None)
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._workout_credit._credit_notify.notify_workout_credit"
|
||||
) as notify:
|
||||
result = locker._apply_credit_for_written_entry(prior)
|
||||
|
||||
assert result.shutdown_adjusted is False
|
||||
assert result.extra_bonus_delta == 0
|
||||
locker._try_adjust_shutdown_for_workout.assert_not_called()
|
||||
locker._read_shutdown_config.assert_not_called()
|
||||
locker._adjust_shutdown_time_by.assert_not_called()
|
||||
notify.assert_not_called()
|
||||
|
||||
def test_skips_save_for_sick_day_type(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
|
||||
101
screen_locker/tests/test_runnerup_watch_filter.py
Normal file
101
screen_locker/tests/test_runnerup_watch_filter.py
Normal file
@ -0,0 +1,101 @@
|
||||
"""scripts/runnerup_watch.sh starts workout-sync only for a FRESH upload.
|
||||
|
||||
endurain-import's adb fallback re-pulls every export it already moved into
|
||||
``processed/`` and renames each into place, so one import fired ~27 events
|
||||
and as many workout-sync starts (~370 on 2026-10-09). The script runs as a
|
||||
black box here: ``inotifywait`` and ``systemctl`` are shims on PATH, the
|
||||
first replaying a fixed event list, the second recording what was started.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "runnerup_watch.sh"
|
||||
_UNIT = "workout-sync.service"
|
||||
# conftest replaces subprocess.run process-wide for every test; this module
|
||||
# needs the real one, captured at import time before any fixture runs.
|
||||
_REAL_RUN = subprocess.run
|
||||
|
||||
|
||||
def _shim(path: Path, body: str) -> None:
|
||||
path.write_text(f"#!/bin/bash\n{body}\n", encoding="utf-8")
|
||||
path.chmod(0o755)
|
||||
|
||||
|
||||
def _run(
|
||||
tmp_path: Path, events: list[str], *, processed: list[str]
|
||||
) -> tuple[list[str], str, int]:
|
||||
"""Replay ``events`` through the watcher; return systemctl calls, stderr, rc."""
|
||||
watch = tmp_path / "RunnerUp"
|
||||
(watch / "processed").mkdir(parents=True)
|
||||
for name in processed:
|
||||
(watch / "processed" / name).write_text("<tcx/>", encoding="utf-8")
|
||||
bin_dir = tmp_path / "bin"
|
||||
bin_dir.mkdir()
|
||||
events_file = tmp_path / "events"
|
||||
events_file.write_text("".join(f"{e}\n" for e in events), encoding="utf-8")
|
||||
calls = tmp_path / "systemctl.calls"
|
||||
_shim(bin_dir / "inotifywait", f'cat "{events_file}"')
|
||||
_shim(
|
||||
bin_dir / "systemctl",
|
||||
f'echo "$*" >>"{calls}"\n[[ $2 == is-active ]] && echo inactive\nexit 0',
|
||||
)
|
||||
env = {
|
||||
**os.environ,
|
||||
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
||||
"RUNNERUP_WATCH_DIR": str(watch),
|
||||
"RUNNERUP_SYNC_UNIT": _UNIT,
|
||||
}
|
||||
done = _REAL_RUN(
|
||||
["bash", str(_SCRIPT)],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
check=False,
|
||||
)
|
||||
lines = calls.read_text(encoding="utf-8").splitlines() if calls.exists() else []
|
||||
return lines, done.stderr, done.returncode
|
||||
|
||||
|
||||
def _starts(calls: list[str]) -> list[str]:
|
||||
return [c for c in calls if c.startswith("--user start")]
|
||||
|
||||
|
||||
def test_a_fresh_upload_starts_the_sync(tmp_path: Path) -> None:
|
||||
"""A new top-level TCX is credited at once."""
|
||||
calls, err, rc = _run(tmp_path, ["2026-10-09-run.tcx"], processed=[])
|
||||
assert _starts(calls) == [f"--user start {_UNIT}"]
|
||||
assert "RunnerUp upload finished: 2026-10-09-run.tcx" in err
|
||||
# inotifywait ending is a failure systemd must restart.
|
||||
assert rc == 1
|
||||
|
||||
|
||||
def test_a_re_pulled_burst_starts_nothing_and_logs_once(tmp_path: Path) -> None:
|
||||
"""Names already in processed/ are skipped, with one line per burst."""
|
||||
old = [f"2026-10-0{d}-run.tcx" for d in range(1, 6)]
|
||||
calls, err, _ = _run(tmp_path, old, processed=old)
|
||||
assert _starts(calls) == []
|
||||
assert err.count("Ignoring re-pulled") == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", [".2026-10-09-run.tcx", "notes.txt", "x.tcx.part"])
|
||||
def test_partials_and_non_tcx_are_ignored(tmp_path: Path, name: str) -> None:
|
||||
"""Dotfiles are partial transfers; anything not *.tcx is not a run."""
|
||||
calls, _, _ = _run(tmp_path, [name], processed=[])
|
||||
assert _starts(calls) == []
|
||||
|
||||
|
||||
def test_fresh_upload_inside_a_re_pull_burst_still_counts(tmp_path: Path) -> None:
|
||||
"""The filter must not swallow the one real upload among the re-pulls."""
|
||||
calls, _, _ = _run(
|
||||
tmp_path,
|
||||
["old-a.tcx", "new.tcx", "old-b.tcx"],
|
||||
processed=["old-a.tcx", "old-b.tcx"],
|
||||
)
|
||||
assert _starts(calls) == [f"--user start {_UNIT}"]
|
||||
@ -52,7 +52,7 @@ class TestUnlockScreenExtras:
|
||||
# Simulate before=22:00, after=23:00 → delta=60
|
||||
old_cfg = (1320, 1320, 300)
|
||||
new_cfg = (1380, 1380, 300)
|
||||
locker._read_shutdown_config.side_effect = [old_cfg, new_cfg]
|
||||
locker._read_shutdown_config.side_effect = [old_cfg, new_cfg, new_cfg]
|
||||
|
||||
text_calls: list[str] = []
|
||||
|
||||
@ -114,7 +114,7 @@ class TestUnlockScreenExtras:
|
||||
"""Re-reading the shutdown config fails after the +1h → delta stays 0."""
|
||||
locker = setup_unlock(mock_tk, tmp_path, seed_today_type="manual_workout")
|
||||
# old_cfg readable, new_cfg unreadable → no delta can be computed.
|
||||
locker._read_shutdown_config.side_effect = [(1320, 1320, 300), None]
|
||||
locker._read_shutdown_config.side_effect = [(1320, 1320, 300), None, None]
|
||||
|
||||
text_calls: list[str] = []
|
||||
object.__setattr__(locker, "_text", lambda msg, **kw: text_calls.append(msg))
|
||||
|
||||
@ -15,6 +15,13 @@
|
||||
# timer run; when one was already in flight it started before this file was
|
||||
# complete, so the start merged into it and a second start is needed.
|
||||
#
|
||||
# Only a FRESH top-level *.tcx counts. endurain-import's adb fallback re-pulls
|
||||
# every phone export it does not see at the top level -- including the ones it
|
||||
# already moved into processed/ -- and renames each into place, so every
|
||||
# 15-min import used to fire ~27 moved_to events and as many workout-sync
|
||||
# starts (~370 on 2026-10-09). A name already in processed/ was handled
|
||||
# before, and dotfiles are partial transfers; both are skipped with zero forks.
|
||||
#
|
||||
# Run by runnerup-watch.service (Restart=always). Event-driven: no polling.
|
||||
# ============================================================================
|
||||
|
||||
@ -22,6 +29,7 @@ set -euo pipefail
|
||||
|
||||
readonly WATCH_DIR="${RUNNERUP_WATCH_DIR:-$HOME/data/cloud/RunnerUp}"
|
||||
readonly SYNC_UNIT="${RUNNERUP_SYNC_UNIT:-workout-sync.service}"
|
||||
readonly PROCESSED_DIR="$WATCH_DIR/processed"
|
||||
|
||||
log() {
|
||||
printf '%s\n' "$*" >&2
|
||||
@ -47,6 +55,15 @@ ingest() {
|
||||
fi
|
||||
}
|
||||
|
||||
# is_fresh_upload NAME -- true for a new top-level TCX, false for anything
|
||||
# endurain-import already handled (its copy is in processed/) or a partial.
|
||||
# Builtins only: this runs once per inotify event.
|
||||
is_fresh_upload() {
|
||||
local name="$1"
|
||||
[[ $name == *.tcx && $name != .* ]] || return 1
|
||||
[[ ! -e $PROCESSED_DIR/$name ]]
|
||||
}
|
||||
|
||||
validate_requirements() {
|
||||
if ! command -v inotifywait >/dev/null 2>&1; then
|
||||
log "ERROR: inotifywait missing (pacman -S inotify-tools) — RunnerUp" \
|
||||
@ -63,9 +80,19 @@ validate_requirements() {
|
||||
main() {
|
||||
validate_requirements
|
||||
log "Watching $WATCH_DIR for finished RunnerUp TCX uploads"
|
||||
local name
|
||||
local name now last_skip_log=0
|
||||
while IFS= read -r name; do
|
||||
if ! is_fresh_upload "$name"; then
|
||||
[[ $name == *.tcx ]] || continue
|
||||
# One line per burst, not per file: the re-pull lands ~27 at once.
|
||||
printf -v now '%(%s)T' -1
|
||||
if ((now - last_skip_log > 60)); then
|
||||
log "Ignoring re-pulled $name (already in processed/; no" \
|
||||
"workout-sync) — further re-pulls this minute not logged"
|
||||
last_skip_log=$now
|
||||
fi
|
||||
continue
|
||||
fi
|
||||
ingest "$name"
|
||||
done < <(inotifywait -m -q -e close_write -e moved_to --format '%f' "$WATCH_DIR")
|
||||
log "ERROR: inotifywait on $WATCH_DIR exited — restarting via systemd"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user