From f3875302d7cb8c3c0169236985f647da42f1d9e9 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sun, 1 Nov 2026 20:56:44 +0100 Subject: [PATCH] Fix pylint findings in source and scripts Real fixes: split over-long functions (lock-decision chain, manual sync, validation, phone verification, status), hoist a dead lazy import, context-manage Xvfb and temp dirs, add encodings, share the silent-failure argv loop. lock_grab_handoff read lock._recovery, which gatelock's LockWindow does not have (its public property is recovery); a contract test now checks every attribute the handoff reads against the real class. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WzxdWYdwRAtMg7ki6TKxuU --- screen_locker/_adb_transport.py | 2 +- screen_locker/_armed_state.py | 6 +- screen_locker/_compliance_state.py | 290 ++++++++++---------- screen_locker/_credit_notify.py | 2 +- screen_locker/_credit_slot.py | 2 +- screen_locker/_extra_benefits.py | 7 +- screen_locker/_http_workout_fetch.py | 2 +- screen_locker/_late_credit.py | 6 +- screen_locker/_manual_sync.py | 120 +++++--- screen_locker/_manual_sync_draft.py | 2 +- screen_locker/_manual_workout.py | 36 +-- screen_locker/_manual_workout_checks.py | 44 +++ screen_locker/_manual_workout_types.py | 4 +- screen_locker/_phone_verification.py | 105 ++++--- screen_locker/_poke_server.py | 2 +- screen_locker/_session_stream.py | 2 +- screen_locker/_shutdown.py | 3 +- screen_locker/_sick_floor.py | 6 +- screen_locker/_status.py | 108 ++++---- screen_locker/_status_projection.py | 2 +- screen_locker/_status_view_verify.py | 6 +- screen_locker/_stream_cache.py | 2 +- screen_locker/_ui_flows.py | 2 +- screen_locker/_ui_flows_relaxed.py | 2 +- screen_locker/_web_server.py | 2 +- screen_locker/_weekly_check.py | 2 +- screen_locker/_workout_app.py | 6 +- screen_locker/_workout_credit.py | 2 +- screen_locker/_workout_handoff.py | 2 +- screen_locker/_workout_session.py | 6 +- screen_locker/poke_server.py | 2 +- screen_locker/tests/test_workout_handoff.py | 41 ++- scripts/_popup_form_check.py | 13 +- scripts/_preview_sim.py | 4 +- scripts/_silent_cli.py | 40 +++ scripts/_xvfb_reexec.py | 32 ++- scripts/check_requirements_sync.py | 2 +- scripts/check_silent_failures.py | 30 +- scripts/check_silent_failures_dart.py | 28 +- scripts/demo_workout_handoff.py | 18 +- scripts/preview_shutdown.py | 14 +- scripts/refresh_refactor_todo.py | 2 +- scripts/rekey_log_entry.py | 4 +- scripts/split_dart_test.py | 17 +- scripts/verify_workout_credited.py | 2 +- 45 files changed, 583 insertions(+), 449 deletions(-) create mode 100644 screen_locker/_manual_workout_checks.py create mode 100644 scripts/_silent_cli.py diff --git a/screen_locker/_adb_transport.py b/screen_locker/_adb_transport.py index e08337f..0e03ec3 100644 --- a/screen_locker/_adb_transport.py +++ b/screen_locker/_adb_transport.py @@ -11,7 +11,7 @@ Deciding whether what came back counts as a workout stays in from __future__ import annotations -from concurrent.futures import ( # pylint: disable=no-name-in-module +from concurrent.futures import ( ThreadPoolExecutor, as_completed, ) diff --git a/screen_locker/_armed_state.py b/screen_locker/_armed_state.py index 8409c2c..7f3d7a1 100644 --- a/screen_locker/_armed_state.py +++ b/screen_locker/_armed_state.py @@ -174,7 +174,7 @@ def locker_unit_active() -> bool | None: if not systemctl_available(): return None code, out = _run(["systemctl", "--user", "is-active", LOCKER_UNIT]) - if code != 0 and not out.strip(): + if code and not out.strip(): return None return out.strip() == "active" @@ -186,7 +186,7 @@ def _login_target_active() -> bool: True when ``graphical-session.target`` is active. """ code, out = _run(["systemctl", "--user", "is-active", _LOGIN_TARGET]) - return code == 0 and out.strip() == "active" + return not code and out.strip() == "active" def collect_states() -> list[TimerState]: @@ -205,7 +205,7 @@ def collect_states() -> list[TimerState]: states.append( TimerState( name=name, - enabled=code == 0 and raw == "enabled", + enabled=not code and raw == "enabled", enabled_raw=raw, # For a timer this is "systemd has a next trigger for it"; # for the login-run service it is "the target it is wanted by diff --git a/screen_locker/_compliance_state.py b/screen_locker/_compliance_state.py index 3a4e634..ec3be98 100644 --- a/screen_locker/_compliance_state.py +++ b/screen_locker/_compliance_state.py @@ -18,6 +18,7 @@ surfaces the pending opportunity separately via ``AutoUpgradeOpportunity``. from __future__ import annotations +from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING @@ -59,6 +60,135 @@ __all__ = [ ] +@dataclass(frozen=True) +class _Step: + """One predicate of the lock-decision chain, in evaluation order.""" + + name: str + reason_true: str + reason_false: str + terminal_reason: str | None = None + note_when_false: bool = False + + +# Order mirrors ``_check_non_verify_exits`` / ``_check_today_state_exits``. +# ``early_bird_pending_expired`` is recorded but never terminal on its own: an +# expired pending marker still falls through toward a full lock, with the +# upgrade opportunity surfaced separately (see module docstring). +_CHAIN: tuple[_Step, ...] = ( + _Step( + "scheduled_skip", + "Today is in scheduled_skips.json (unconditional manual skip).", + "Today is not a manually scheduled skip day.", + "Manually scheduled skip day — no lock, no workout required.", + ), + _Step( + "early_bird_pending_expired", + "Logged in during the early-bird window and it has since closed " + "with no workout confirmed yet.", + "No expired early-bird pending marker.", + ), + _Step( + "early_bird_window_open", + "Logged in during the early-bird window earlier today; still " + "waiting to see a workout before the window closes.", + "Not pending inside an open early-bird window.", + "Early-bird window still open — lock skipped while waiting for a workout.", + ), + _Step( + "sick_day", + "Today is a sick day in the shared sickdays pool.", + "Today is not a sick day.", + "Sick day today — lock skipped.", + ), + _Step( + "already_logged", + "A validly-signed workout is already logged for today.", + "No workout logged for today yet.", + "Workout already logged today — lock skipped.", + note_when_false=True, + ), + _Step( + "wake_alarm_skip", + "The morning session earned a workout skip for now.", + "No morning-session skip in force.", + "Morning session earned a skip — lock skipped.", + ), + _Step( + "relaxed_day_already_skipped", + "Relaxed day already dismissed via Skip — No Penalty today.", + "No same-day relaxed-day dismissal on record.", + "Relaxed day already skipped today — lock skipped, prompt not shown again.", + ), + _Step( + "relaxed_day", + "Today is a relaxed day (Tue/Wed/Thu).", + "Today is not a relaxed day.", + "Relaxed day (Tue/Wed/Thu) — workout optional, lock skipped.", + ), + _Step( + "weekly_minimum_met", + "Weekly workout minimum already reached.", + "Weekly workout minimum not yet reached.", + "Weekly workout minimum already met — lock skipped.", + ), + _Step( + "evening_lockdown_today", + "Tonight's night lockdown already fired (>= 12:00 today).", + "No evening night lockdown yet today.", + "Night lockdown already happened tonight — the workout lock does not re-arm.", + ), +) + + +def _chain_flags( + *, + log_file: Path, + weekly_minimum_met: bool, + relaxed_day: bool, + wake_skip: bool, + evening_lockdown: bool, + local_dt: datetime, + pending: bool, +) -> dict[str, bool]: + """Evaluate every predicate once; the keys are the ``_CHAIN`` step names.""" + today_str = local_day(local_dt) + # The early-bird window IS the wake-alarm carrot (EarlyBirdMixin). + return { + "scheduled_skip": is_scheduled_skip_today(local_dt.date()), + "early_bird_pending_expired": pending and not wake_skip, + "early_bird_window_open": pending and wake_skip, + "sick_day": is_sick_day_today(local_dt.date()), + "already_logged": has_logged_today(log_file, today=today_str), + "wake_alarm_skip": wake_skip, + "relaxed_day_already_skipped": relaxed_day + and is_relaxed_day_skipped_today(log_file, today=today_str), + "relaxed_day": relaxed_day, + "weekly_minimum_met": weekly_minimum_met, + "evening_lockdown_today": evening_lockdown, + } + + +def _walk_chain( + fired: dict[str, bool], context: StageContext, degraded_note: str +) -> LockExplanation | None: + """Record each step in ``context.trace``; return the first terminal one.""" + for step in _CHAIN: + hit = fired[step.name] + reason = step.reason_true if hit else step.reason_false + if step.note_when_false and not hit: + reason += degraded_note + context.trace.append(PredicateResult(step.name, hit, reason)) + if hit and step.terminal_reason is not None: + return _stage( + fired=False, + stage=step.name, + reason=step.terminal_reason, + context=context, + ) + return None + + def explain_lock_decision( *, log_file: Path, @@ -77,155 +207,25 @@ def explain_lock_decision( reported via ``heat_skip_evaluated=False``, not guessed. """ instant = now if now is not None else datetime.now(tz=UTC) - today_str = local_day(instant) - local_dt = instant.astimezone() - - scheduled = is_scheduled_skip_today(local_dt.date()) - pending = is_early_bird_pending(early_bird_pending_file, today=today_str) - # The early-bird window IS the wake-alarm carrot (EarlyBirdMixin). - window_open = wake_skip - expired = pending and not window_open - early_bird_open_and_pending = pending and window_open - sick_today = is_sick_day_today(local_dt.date()) - logged = has_logged_today(log_file, today=today_str) - relaxed_day_already_skipped = relaxed_day and is_relaxed_day_skipped_today( - log_file, today=today_str + pending = is_early_bird_pending(early_bird_pending_file, today=local_day(instant)) + fired = _chain_flags( + log_file=log_file, + weekly_minimum_met=weekly_minimum_met, + relaxed_day=relaxed_day, + wake_skip=wake_skip, + evening_lockdown=evening_lockdown, + local_dt=instant.astimezone(), + pending=pending, ) - auto_upgrade = describe_auto_upgrade_opportunity( early_bird_pending=pending, - early_bird_window_open=window_open, + early_bird_window_open=wake_skip, ) - - trace: list[PredicateResult] = [] - stage_context = StageContext(trace, auto_upgrade, bool(degraded_sources)) - - def _check( - name: str, - *, - fired: bool, - reason_true: str, - reason_false: str, - terminal_reason: str | None = None, - ) -> LockExplanation | None: - """Record one trace step; return an explanation if it terminates the chain.""" - reason = reason_true if fired else reason_false - trace.append(PredicateResult(name, fired, reason)) - if fired and terminal_reason is not None: - return _stage( - fired=False, - stage=name, - reason=terminal_reason, - context=stage_context, - ) - return None - - result = _check( - "scheduled_skip", - fired=scheduled, - reason_true="Today is in scheduled_skips.json (unconditional manual skip).", - reason_false="Today is not a manually scheduled skip day.", - terminal_reason="Manually scheduled skip day — no lock, no workout required.", - ) - if result is not None: - return result - - # Recorded but never terminal on its own: an expired pending marker still - # falls through toward a full lock, with the upgrade opportunity surfaced - # separately via `auto_upgrade` (see module docstring). - _check( - "early_bird_pending_expired", - fired=expired, - reason_true="Logged in during the early-bird window and it has since closed " - "with no workout confirmed yet.", - reason_false="No expired early-bird pending marker.", - ) - - result = _check( - "early_bird_window_open", - fired=early_bird_open_and_pending, - reason_true="Logged in during the early-bird window earlier today; still " - "waiting to see a workout before the window closes.", - reason_false="Not pending inside an open early-bird window.", - terminal_reason=( - "Early-bird window still open — lock skipped while waiting for a workout." - ), - ) - if result is not None: - return result - - result = _check( - "sick_day", - fired=sick_today, - reason_true="Today is a sick day in the shared sickdays pool.", - reason_false="Today is not a sick day.", - terminal_reason="Sick day today — lock skipped.", - ) - if result is not None: - return result - + context = StageContext([], auto_upgrade, bool(degraded_sources)) degraded_note = describe_degraded_sources(degraded_sources) - result = _check( - "already_logged", - fired=logged, - reason_true="A validly-signed workout is already logged for today.", - reason_false="No workout logged for today yet." + degraded_note, - terminal_reason="Workout already logged today — lock skipped.", - ) - if result is not None: - return result - - result = _check( - "wake_alarm_skip", - fired=wake_skip, - reason_true="The morning session earned a workout skip for now.", - reason_false="No morning-session skip in force.", - terminal_reason="Morning session earned a skip — lock skipped.", - ) - if result is not None: - return result - - result = _check( - "relaxed_day_already_skipped", - fired=relaxed_day_already_skipped, - reason_true="Relaxed day already dismissed via Skip — No Penalty today.", - reason_false="No same-day relaxed-day dismissal on record.", - terminal_reason="Relaxed day already skipped today — lock skipped, " - "prompt not shown again.", - ) - if result is not None: - return result - - result = _check( - "relaxed_day", - fired=relaxed_day, - reason_true="Today is a relaxed day (Tue/Wed/Thu).", - reason_false="Today is not a relaxed day.", - terminal_reason="Relaxed day (Tue/Wed/Thu) — workout optional, lock skipped.", - ) - if result is not None: - return result - - result = _check( - "weekly_minimum_met", - fired=weekly_minimum_met, - reason_true="Weekly workout minimum already reached.", - reason_false="Weekly workout minimum not yet reached.", - terminal_reason="Weekly workout minimum already met — lock skipped.", - ) - if result is not None: - return result - - result = _check( - "evening_lockdown_today", - fired=evening_lockdown, - reason_true="Tonight's night lockdown already fired (>= 12:00 today).", - reason_false="No evening night lockdown yet today.", - terminal_reason="Night lockdown already happened tonight — the workout " - "lock does not re-arm.", - ) - if result is not None: - return result + terminal = _walk_chain(fired, context, degraded_note) + if terminal is not None: + return terminal # Named for the conclusion; "..._pending_heat_check" read as "stuck". return _stage( @@ -237,5 +237,5 @@ def explain_lock_decision( "temperature is checked, and a run may or may not be happening " "right now (see Health)." + degraded_note ), - context=stage_context, + context=context, ) diff --git a/screen_locker/_credit_notify.py b/screen_locker/_credit_notify.py index 42c0db2..8feaa5a 100644 --- a/screen_locker/_credit_notify.py +++ b/screen_locker/_credit_notify.py @@ -103,7 +103,7 @@ def send_notification(title: str, body: str) -> bool: except (OSError, subprocess.TimeoutExpired) as exc: _logger.warning("Desktop notification failed (%s): %s", exc, body) return False - if result.returncode != 0: + if result.returncode: _logger.warning( "notify-send exited %d (%s) — no desktop notification: %s", result.returncode, diff --git a/screen_locker/_credit_slot.py b/screen_locker/_credit_slot.py index d2ffb74..9226782 100644 --- a/screen_locker/_credit_slot.py +++ b/screen_locker/_credit_slot.py @@ -57,7 +57,7 @@ def credit_slot( 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 + return CreditSlot.FURTHER if before else CreditSlot.FIRST def _counts_alone(entry: Mapping[str, Any]) -> bool: diff --git a/screen_locker/_extra_benefits.py b/screen_locker/_extra_benefits.py index 79b66a1..1d7ddc8 100644 --- a/screen_locker/_extra_benefits.py +++ b/screen_locker/_extra_benefits.py @@ -18,7 +18,7 @@ State is persisted in ``extra_benefits_state.json`` next to this file. from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import UTC, date, datetime, timedelta import json import logging from typing import TYPE_CHECKING, Any @@ -62,7 +62,7 @@ def _save_state(state_file: Path, state: dict[str, Any]) -> None: _logger.warning("Failed to save extra benefits state: %s", exc) -def _current_iso_week(now: datetime) -> str: +def _current_iso_week(now: date) -> str: """Return *now*'s ISO week as ``YYYY-Www``.""" year, week, _ = now.isocalendar() return f"{year}-W{week:02d}" @@ -103,8 +103,7 @@ def process_week_transition(log_file: Path, state_file: Path) -> list[str]: eb_weeks: list[str] = list(state.get("extended_early_bird_iso_weeks", [])) rewards: list[str] = [] - prev_year, prev_week, _ = sunday_prev_week.isocalendar() - prev_week_str = f"{prev_year}-W{prev_week:02d}" + prev_week_str = _current_iso_week(sunday_prev_week) if prev_week_count >= _BONUS_THRESHOLD: streak += 1 diff --git a/screen_locker/_http_workout_fetch.py b/screen_locker/_http_workout_fetch.py index 8f77010..1a3646f 100644 --- a/screen_locker/_http_workout_fetch.py +++ b/screen_locker/_http_workout_fetch.py @@ -10,7 +10,7 @@ app serves the same JSON over ``WORKOUT_HTTP_PORT``. from __future__ import annotations -from concurrent.futures import ( # pylint: disable=no-name-in-module +from concurrent.futures import ( ThreadPoolExecutor, as_completed, ) diff --git a/screen_locker/_late_credit.py b/screen_locker/_late_credit.py index 08a77ac..fd1c28f 100644 --- a/screen_locker/_late_credit.py +++ b/screen_locker/_late_credit.py @@ -54,11 +54,13 @@ class LateCreditMixin: if TYPE_CHECKING: - def _read_shutdown_config(self) -> tuple[int, int, int] | None: ... + def _read_shutdown_config(self) -> tuple[int, int, int] | None: + raise NotImplementedError def _write_shutdown_config( self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = ... - ) -> bool: ... + ) -> bool: + raise NotImplementedError def _apply_sick_floor(self) -> bool: ... diff --git a/screen_locker/_manual_sync.py b/screen_locker/_manual_sync.py index 718e910..f8ddd74 100644 --- a/screen_locker/_manual_sync.py +++ b/screen_locker/_manual_sync.py @@ -45,6 +45,9 @@ if TYPE_CHECKING: from collections.abc import Callable, Iterable, Mapping from pathlib import Path + from screen_locker._log_mixin import RecordResult + from screen_locker._manual_workout import ManualWorkoutDraft + OnIngestedCallback = Callable[[dict, "list[dict]"], None] __all__ = ["ingest_manual_records", "is_empty_stub", "reconstruct_draft"] @@ -113,6 +116,76 @@ def _due_date( return date +def _new_valid_draft( + log_file: Path, + record_id: str, + payload: Mapping[str, object], + reference: str, +) -> tuple[str, ManualWorkoutDraft] | None: + """The due date and valid draft of a not-yet-ingested record, else ``None``.""" + date = _due_date(record_id, payload, reference) + if date is None: + return None + if _already_ingested(load_workout_log(log_file), record_id): + return None + draft = _draft_or_report(record_id, payload) + if draft is None: + return None + error = validate_manual_workout(draft) + if error is not None: + _logger.warning("Manual record %s invalid: %s", record_id, error) + return None + return date, draft + + +def _ingest_one( + log_file: Path, + record_id: str, + payload: Mapping[str, object], + reference: str, + pending: Mapping[str, str], +) -> tuple[dict[str, object], RecordResult] | None: + """Validate, budget-check and append one record; ``None`` when skipped.""" + ready = _new_valid_draft(log_file, record_id, payload, reference) + if ready is None: + return None + date, draft = ready + over_budget = budget_error_for_date( + log_file, + date, + [day for rid, day in pending.items() if rid != record_id], + ) + if over_budget is not None: + _logger.info( + "Manual-workout budget exhausted — %s not ingested (%s)", + record_id, + over_budget, + ) + return None + entry = build_entry(draft) + entry[_SYNC_ID_FIELD] = record_id + # The wire id IS the workout's id: the device that logged it minted + # ``manual:T`` once, and the publisher keys on the same + # id. Without this the chokepoint re-derived one from ``date`` + + # start, and when the two dates disagreed (a record whose day key was + # later corrected to the local day) the same workout landed twice + # under two ids -- and then consumed two manual-budget slots. + entry["workout_id"] = record_id + result = write_signed_entry(log_file, date, entry) + if not result.appended: + # This is dedup working, not credit being lost: the workout is + # already in the log under this same workout_id. It was logged at + # warning and re-reported on every 15-minute cycle, which read like + # a recurring fault and buried the lines that did need attention. + _logger.info( + "Manual record %s is already logged under the same workout_id " + "— skipped as a duplicate (its credit is already counted)", + record_id, + ) + return None + return entry, result + + def ingest_manual_records( log_file: Path, records: Iterable[tuple[str, Mapping[str, object]]], @@ -157,51 +230,10 @@ def ingest_manual_records( pending = pending_future_dates(log_file, batch, today=reference) ingested: list[str] = [] for record_id, payload in batch: - date = _due_date(record_id, payload, reference) - if date is None: - continue - if _already_ingested(load_workout_log(log_file), record_id): - continue - draft = _draft_or_report(record_id, payload) - if draft is None: - continue - error = validate_manual_workout(draft) - if error is not None: - _logger.warning("Manual record %s invalid: %s", record_id, error) - continue - over_budget = budget_error_for_date( - log_file, - date, - [day for rid, day in pending.items() if rid != record_id], - ) - if over_budget is not None: - _logger.info( - "Manual-workout budget exhausted — %s not ingested (%s)", - record_id, - over_budget, - ) - continue - entry = build_entry(draft) - entry[_SYNC_ID_FIELD] = record_id - # The wire id IS the workout's id: the device that logged it minted - # ``manual:T`` once, and the publisher keys on the same - # id. Without this the chokepoint re-derived one from ``date`` + - # start, and when the two dates disagreed (a record whose day key was - # later corrected to the local day) the same workout landed twice - # under two ids -- and then consumed two manual-budget slots. - entry["workout_id"] = record_id - result = write_signed_entry(log_file, date, entry) - if not result.appended: - # This is dedup working, not credit being lost: the workout is - # already in the log under this same workout_id. It was logged at - # warning and re-reported on every 15-minute cycle, which read like - # a recurring fault and buried the lines that did need attention. - _logger.info( - "Manual record %s is already logged under the same workout_id " - "— skipped as a duplicate (its credit is already counted)", - record_id, - ) + outcome = _ingest_one(log_file, record_id, payload, reference, pending) + if outcome is None: continue + entry, result = outcome ingested.append(record_id) if on_ingested is not None: on_ingested(entry, result.prior_entries) diff --git a/screen_locker/_manual_sync_draft.py b/screen_locker/_manual_sync_draft.py index 10f588a..6432d3f 100644 --- a/screen_locker/_manual_sync_draft.py +++ b/screen_locker/_manual_sync_draft.py @@ -49,7 +49,7 @@ def is_empty_stub(payload: Mapping[str, object]) -> bool: produced an unbounded stream of identical warnings (four every 15 minutes), which is how the genuinely actionable lines got lost in the noise. """ - return not (_SUBSTANTIVE_FIELDS & set(payload)) + return not _SUBSTANTIVE_FIELDS & set(payload) def _coerce_int(value: object) -> int: diff --git a/screen_locker/_manual_workout.py b/screen_locker/_manual_workout.py index 1ad1071..ba630c9 100644 --- a/screen_locker/_manual_workout.py +++ b/screen_locker/_manual_workout.py @@ -20,12 +20,15 @@ import logging from screen_locker._constants import ( MANUAL_WORKOUT_DESCRIPTION_MIN_CHARS, - MANUAL_WORKOUT_REFLECTION_MIN_CHARS, MANUAL_WORKOUT_RPE_MAX, MANUAL_WORKOUT_RPE_MIN, MIN_WORKOUT_DURATION_MINUTES, WORKOUT_DURATION_ACCEPT_MINUTES, ) +from screen_locker._manual_workout_checks import ( + validate_reflections, + validate_required, +) from screen_locker._manual_workout_types import ( MANUAL_WORKOUT_TYPE, SPORT_CHOICES, @@ -82,19 +85,9 @@ def _duration_minutes(draft: ManualWorkoutDraft) -> float | None: def validate_manual_workout(draft: ManualWorkoutDraft) -> str | None: """Return an error message if the draft is invalid, else None.""" - if draft.sport not in SPORT_CHOICES: - return "Please choose a sport" - - required = { - "Start time": draft.start_time, - "End time": draft.end_time, - "Location name": draft.location_name, - "Transport method": draft.transport_method, - "Cost": draft.cost, - } - for label, value in required.items(): - if not value.strip(): - return f"{label} is required" + error = validate_required(draft) + if error is not None: + return error duration = _duration_minutes(draft) if duration is None: @@ -119,18 +112,7 @@ def validate_manual_workout(draft: ManualWorkoutDraft) -> str | None: if error is not None: return error - for label, value in ( - ("What went well", draft.went_well), - ("What to improve", draft.to_improve), - ("Overall feeling", draft.overall_feeling), - ): - if len(value.strip()) < MANUAL_WORKOUT_REFLECTION_MIN_CHARS: - return ( - f"{label} must be at least {MANUAL_WORKOUT_REFLECTION_MIN_CHARS} " - f"characters (currently {len(value.strip())})" - ) - - return None + return validate_reflections(draft) def _validate_table_tennis(draft: ManualWorkoutDraft) -> str | None: @@ -143,7 +125,7 @@ def _validate_table_tennis(draft: ManualWorkoutDraft) -> str | None: ): if value < 0: return f"{label} cannot be negative" - if draft.matches_won + draft.matches_lost == 0: + if not draft.matches_won + draft.matches_lost: return "Enter at least one match played (won + lost)" if not draft.racket.strip(): return "Racket is required" diff --git a/screen_locker/_manual_workout_checks.py b/screen_locker/_manual_workout_checks.py new file mode 100644 index 0000000..329955f --- /dev/null +++ b/screen_locker/_manual_workout_checks.py @@ -0,0 +1,44 @@ +"""Field-level checks shared by ``validate_manual_workout``. + +Split out of :mod:`screen_locker._manual_workout` to keep every file under the +250-line cap. +""" + +from __future__ import annotations + +from screen_locker._constants import MANUAL_WORKOUT_REFLECTION_MIN_CHARS +from screen_locker._manual_workout_types import SPORT_CHOICES, ManualWorkoutDraft + + +def validate_required(draft: ManualWorkoutDraft) -> str | None: + """Return an error if the sport is unchosen or a required field is blank.""" + if draft.sport not in SPORT_CHOICES: + return "Please choose a sport" + + required = { + "Start time": draft.start_time, + "End time": draft.end_time, + "Location name": draft.location_name, + "Transport method": draft.transport_method, + "Cost": draft.cost, + } + for label, value in required.items(): + if not value.strip(): + return f"{label} is required" + return None + + +def validate_reflections(draft: ManualWorkoutDraft) -> str | None: + """Return an error if a reflection field is shorter than the minimum.""" + for label, value in ( + ("What went well", draft.went_well), + ("What to improve", draft.to_improve), + ("Overall feeling", draft.overall_feeling), + ): + if len(value.strip()) < MANUAL_WORKOUT_REFLECTION_MIN_CHARS: + return ( + f"{label} must be at least {MANUAL_WORKOUT_REFLECTION_MIN_CHARS} " + f"characters (currently {len(value.strip())})" + ) + + return None diff --git a/screen_locker/_manual_workout_types.py b/screen_locker/_manual_workout_types.py index 1b60e69..885ddde 100644 --- a/screen_locker/_manual_workout_types.py +++ b/screen_locker/_manual_workout_types.py @@ -103,8 +103,8 @@ def count_in_window( dated after today are NOT in it. Whether a new entry fits is a different question, answered per workout date by ``_manual_future``. """ - today_str = today or _today_iso() - today_dt = _parse_iso(today_str) + day = today or _today_iso() + today_dt = _parse_iso(day) if today_dt is None: return 0 cutoff = today_dt - timedelta(days=days) diff --git a/screen_locker/_phone_verification.py b/screen_locker/_phone_verification.py index 4d5e364..c769f18 100644 --- a/screen_locker/_phone_verification.py +++ b/screen_locker/_phone_verification.py @@ -28,6 +28,18 @@ _HTTP_OK = _http_client.OK _logger = logging.getLogger(__name__) +def _in_current_week(date_str: str) -> bool: + """True if *date_str* parses and falls in this ISO week up to today.""" + today = datetime.now(tz=UTC).astimezone().date() + week_start = today - timedelta(days=today.weekday()) + try: + entry_date = date.fromisoformat(date_str) + except ValueError: + _logger.warning("StrongLifts JSON has an unparsable date: %r", date_str) + return False + return week_start <= entry_date <= today + + class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): """Mixin providing phone-based workout verification via ADB.""" @@ -119,16 +131,7 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): return 0 data = self._pull_workout_app_json() date_str = data.get("date") if data else None - if not date_str: - return 0 - today = datetime.now(tz=UTC).astimezone().date() - week_start = today - timedelta(days=today.weekday()) - try: - entry_date = date.fromisoformat(date_str) - except ValueError: - _logger.warning("StrongLifts JSON has an unparsable date: %r", date_str) - return 0 - if not week_start <= entry_date <= today: + if not date_str or not _in_current_week(date_str): return 0 status, msg = self._validate_json_content(data) if status != "verified": @@ -145,6 +148,33 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): # ── HTTP fallback (no ADB / developer options required) ─────────────────── + def _pull_sync_result(self) -> tuple[tuple[str, str] | None, str | None]: + """Firebase sync first: the validated result (if any) and sync error. + + Firebase sync is the primary channel -- it works without the phone + being on the same network as the PC. Only a *verified* result + short-circuits ADB/HTTP: a stale/incomplete cloud entry (e.g. the + phone's last push predates today, or a later push failed offline) + must not preempt a fresher ADB pull, so anything less than + "verified" is kept by the caller as a fallback-of-last-resort + candidate. A sync error never blocks unlocking either; `sync_failed` + only surfaces if ADB/HTTP also come up empty. + """ + synced_data, sync_error = pull_synced_workout() + if synced_data is None: + if sync_error is not None: + _logger.info( + "Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error + ) + return None, sync_error + sync_result = self._validate_json_data(synced_data) + if sync_result[0] != "verified": + _logger.info( + "Synced data not fully verified (%s) — trying ADB/HTTP...", + sync_result[0], + ) + return sync_result, sync_error + def _verify_phone_workout(self) -> tuple[str, str]: """Verify today's workout: Firebase sync first, ADB/HTTP as fallback. @@ -156,28 +186,9 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): if not clock_ok: return "clock_tampered", clock_msg - # Firebase sync is the primary channel — it works without the phone - # being on the same network as the PC. Only a *verified* sync result - # short-circuits ADB/HTTP: a stale/incomplete cloud entry (e.g. the - # phone's last push predates today, or a later push failed offline) - # must not preempt a fresher ADB pull, so anything less than - # "verified" is kept as a fallback-of-last-resort candidate instead - # of being returned immediately. A sync error never blocks unlocking - # either; `sync_failed` only surfaces if ADB/HTTP also come up empty. - synced_data, sync_error = pull_synced_workout() - sync_result: tuple[str, str] | None = None - if synced_data is not None: - sync_result = self._validate_json_data(synced_data) - if sync_result[0] == "verified": - return sync_result - _logger.info( - "Synced data not fully verified (%s) — trying ADB/HTTP...", - sync_result[0], - ) - elif sync_error is not None: - _logger.info( - "Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error - ) + sync_result, sync_error = self._pull_sync_result() + if sync_result is not None and sync_result[0] == "verified": + return sync_result # Prefer ADB when a device is visible, but if the pull yields no usable # JSON, fall through to the HTTP/WiFi scan — the app's in-memory HTTP @@ -194,13 +205,23 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): data = self._fetch_http_workout() if data is not None: return self._validate_json_data(data) - if sync_result is not None: - return sync_result - if sync_error is not None: - return "sync_failed", sync_error - if adb_connected: - return ( - "not_verified", - "Workout app JSON not found. Complete a workout in the app first.", - ) - return "no_phone", "Phone not reachable via ADB or HTTP on local network" + return _no_data_result(sync_result, sync_error, adb_connected=adb_connected) + + +def _no_data_result( + sync_result: tuple[str, str] | None, + sync_error: str | None, + *, + adb_connected: bool, +) -> tuple[str, str]: + """The verdict when neither ADB nor HTTP produced a workout JSON.""" + if sync_result is not None: + return sync_result + if sync_error is not None: + return "sync_failed", sync_error + if adb_connected: + return ( + "not_verified", + "Workout app JSON not found. Complete a workout in the app first.", + ) + return "no_phone", "Phone not reachable via ADB or HTTP on local network" diff --git a/screen_locker/_poke_server.py b/screen_locker/_poke_server.py index c4c3eb1..bca4174 100644 --- a/screen_locker/_poke_server.py +++ b/screen_locker/_poke_server.py @@ -135,7 +135,7 @@ class PokeHandler(BaseHTTPRequestHandler): started = time.perf_counter() try: status, reply = self._handle() - except Exception: + except Exception: # pylint: disable=broad-exception-caught # handler boundary: logged, answered 500 _logger.exception("Workout poke from %s crashed", self.client_address[0]) status = HTTPStatus.INTERNAL_SERVER_ERROR reply = _refused( diff --git a/screen_locker/_session_stream.py b/screen_locker/_session_stream.py index a6cb461..87af0a0 100644 --- a/screen_locker/_session_stream.py +++ b/screen_locker/_session_stream.py @@ -123,7 +123,7 @@ class SessionStream: except StreamError as exc: _logger.warning("Firebase session stream failed: %s; %s", exc, _FALLBACK) return False - except Exception: + except Exception: # pylint: disable=broad-exception-caught # daemon boundary: logged, loop reconnects _logger.exception("Firebase session stream crashed; %s", _FALLBACK) return False return True diff --git a/screen_locker/_shutdown.py b/screen_locker/_shutdown.py index 9d9c65a..f6fec64 100644 --- a/screen_locker/_shutdown.py +++ b/screen_locker/_shutdown.py @@ -63,8 +63,7 @@ def read_shutdown_config(path: Path) -> tuple[int, int, int] | None: else: _logger.warning("Shutdown config missing required values") return None - mon_wed, thu_sun, morning_end = values - return mon_wed, thu_sun, morning_end + return values[0], values[1], values[2] class ShutdownMixin(SickFloorMixin, LateCreditMixin): diff --git a/screen_locker/_sick_floor.py b/screen_locker/_sick_floor.py index 17d32a8..0dc09e7 100644 --- a/screen_locker/_sick_floor.py +++ b/screen_locker/_sick_floor.py @@ -48,11 +48,13 @@ class SickFloorMixin: if TYPE_CHECKING: - def _read_shutdown_config(self) -> tuple[int, int, int] | None: ... + def _read_shutdown_config(self) -> tuple[int, int, int] | None: + raise NotImplementedError def _write_shutdown_config( self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = ... - ) -> bool: ... + ) -> bool: + raise NotImplementedError def _apply_sick_floor(self) -> bool: """Raise tonight's shutdown to the sick target if today is sick. diff --git a/screen_locker/_status.py b/screen_locker/_status.py index 24ebfd5..10c18ca 100644 --- a/screen_locker/_status.py +++ b/screen_locker/_status.py @@ -71,55 +71,29 @@ def _print_day_line(d: date, entries: list[dict], sick_days: frozenset[date]) -> return any_counted -def run_status(locker: ScreenLocker) -> None: - """Print weekly workout status, run RunnerUp scan, apply bonus, then exit.""" - today = datetime.now(tz=UTC).astimezone().date() - monday = today - timedelta(days=today.weekday()) - log_file: Path = locker.log_file - log_data = load_workout_log(log_file) - sick_days = workout_standdown_dates() +def _scan_runnerup(locker: ScreenLocker, log_file: Path, before_count: int) -> int: + """Run the RunnerUp week scan, print what it filled, return the new count. - print("=== Weekly Workout Status ===") - - # Per-day breakdown - for i in range(7): - d = monday + timedelta(days=i) - if d > today: - break - _print_day_line(d, log_data.get(d.isoformat(), []), sick_days) - - print() - - # The weekly total, per the anti-gaming rule (each verified workout counts - # individually, manual counts once/day) — NOT a per-day checkmark count, - # which would undercount a day holding two verified workouts. - before_count = count_weekly_workouts(log_file) - - # RunnerUp scan. Each fill is credited by the shared ingestion callback - # (+2h first counted workout of its day, +1h each further one), the same - # rule the timer's sync pass and the daily base reset use. + Each fill is credited by the shared ingestion callback (+2h first counted + workout of its day, +1h each further one), the same rule the timer's sync + pass and the daily base reset use. + """ n_filled = locker._scan_and_fill_week_runnerup( log_file, on_ingested=locker._credit_ingested_workout ) locker.workout_data = {} - if n_filled > 0: - print(f" Auto-filled {n_filled} workout(s) from RunnerUp exports.") - after_count = count_weekly_workouts(log_file) - _, thu_sun, _ = locker._read_shutdown_config() or (None, None, None) - if thu_sun is not None: - print(f" Shutdown tonight: {hhmm(thu_sun)}.") - else: + if n_filled <= 0: print(" No new workouts found via RunnerUp scan.") - after_count = before_count + return before_count + print(f" Auto-filled {n_filled} workout(s) from RunnerUp exports.") + _, thu_sun, _ = locker._read_shutdown_config() or (None, None, None) + if thu_sun is not None: + print(f" Shutdown tonight: {hhmm(thu_sun)}.") + return count_weekly_workouts(log_file) - print() - # Extra benefits summary - streak = current_streak(EXTRA_BENEFITS_FILE) - eb_ext = has_extended_early_bird(EXTRA_BENEFITS_FILE) - eb_str = "Yes — until 09:00" if eb_ext else "No" - - # Heat skips this month +def _heat_skips_this_month(log_data: dict[str, list[dict]]) -> str: + """One-line summary of this month's heat skips: count and the latest.""" this_month = datetime.now(tz=UTC).astimezone().date().strftime("%Y-%m") heat_entries = [ (d, entry) @@ -129,21 +103,17 @@ def run_status(locker: ScreenLocker) -> None: if isinstance(entry, dict) and entry.get("workout_data", {}).get("type") == "heat_skip" ] - if heat_entries: - last_date, last_e = max(heat_entries, key=lambda de: de[0]) - last_temp = last_e.get("workout_data", {}).get("temperature_celsius", "?") - heat_str = f"{len(heat_entries)} (last: {last_date}, {last_temp}°C)" - else: - heat_str = "0" + if not heat_entries: + return "0" + last_date, last_e = max(heat_entries, key=lambda de: de[0]) + last_temp = last_e.get("workout_data", {}).get("temperature_celsius", "?") + return f"{len(heat_entries)} (last: {last_date}, {last_temp}°C)" - print(f" Streak (5+ wks) : {streak}") - print(f" Early-bird extended : {eb_str}") - print(f" Heat skips (month) : {heat_str}") - print() +def _print_minimum(after_count: int) -> None: + """Print how far the weekly count is from (or past) the minimum.""" remaining = max(0, WEEKLY_WORKOUT_MINIMUM - after_count) extra = max(0, after_count - WEEKLY_WORKOUT_MINIMUM) - if remaining > 0: print( f" Need {remaining} more to reach the minimum ({WEEKLY_WORKOUT_MINIMUM})." @@ -156,13 +126,37 @@ def run_status(locker: ScreenLocker) -> None: f" ({WEEKLY_WORKOUT_MINIMUM}/{WEEKLY_WORKOUT_MINIMUM})." ) - # Shutdown config + +def run_status(locker: ScreenLocker) -> None: + """Print weekly workout status, run RunnerUp scan, apply bonus, then exit.""" + today = datetime.now(tz=UTC).astimezone().date() + monday = today - timedelta(days=today.weekday()) + log_file: Path = locker.log_file + log_data = load_workout_log(log_file) + sick_days = workout_standdown_dates() + + print("=== Weekly Workout Status ===") + for i in range(7): + d = monday + timedelta(days=i) + if d > today: + break + _print_day_line(d, log_data.get(d.isoformat(), []), sick_days) + print() + + after_count = _scan_runnerup(locker, log_file, count_weekly_workouts(log_file)) + print() + + eb_ext = has_extended_early_bird(EXTRA_BENEFITS_FILE) + print(f" Streak (5+ wks) : {current_streak(EXTRA_BENEFITS_FILE)}") + print(f" Early-bird extended : {'Yes — until 09:00' if eb_ext else 'No'}") + print(f" Heat skips (month) : {_heat_skips_this_month(log_data)}") + print() + + _print_minimum(after_count) + cfg = locker._read_shutdown_config() if cfg: - _mw, _ts, _morning = cfg - print(f" Shutdown tonight : {hhmm(_mw)}") + print(f" Shutdown tonight : {hhmm(cfg[0])}") - # --status is a read-out, not an enforcement run; make that explicit so the - # journal never reads as though the locker evaluated and declined to lock. record_no_decision("--status") sys.exit(0) diff --git a/screen_locker/_status_projection.py b/screen_locker/_status_projection.py index b174777..971e859 100644 --- a/screen_locker/_status_projection.py +++ b/screen_locker/_status_projection.py @@ -89,7 +89,7 @@ def _tonight_minutes(snapshot: StatusSnapshot) -> int | None: def format_summary_line(snapshot: StatusSnapshot) -> str: """Cheap one-liner for the i3blocks summary (e.g. before/after a click).""" week = snapshot.week - mark = "✓" if week.remaining == 0 else "…" + mark = "…" if week.remaining else "✓" tonight = _tonight_minutes(snapshot) tonight_str = hhmm(tonight) if tonight is not None else "?" sick = " · sick day" if snapshot.today.is_sick_day else "" diff --git a/screen_locker/_status_view_verify.py b/screen_locker/_status_view_verify.py index 793cdb7..144133e 100644 --- a/screen_locker/_status_view_verify.py +++ b/screen_locker/_status_view_verify.py @@ -8,7 +8,7 @@ screen's unlock does. from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import date from typing import TYPE_CHECKING @@ -19,6 +19,7 @@ from screen_locker._day import today_str from screen_locker._earned import extra_minutes, span from screen_locker._status_data import gather_status from screen_locker._weekly_check import WEEKLY_WORKOUT_MINIMUM, count_weekly_workouts +from screen_locker.screen_lock import ScreenLocker if TYPE_CHECKING: from collections.abc import Callable @@ -26,7 +27,6 @@ if TYPE_CHECKING: from pathlib import Path from screen_locker._workout_credit import WorkoutCreditResult - from screen_locker.screen_lock import ScreenLocker @dataclass(frozen=True) @@ -53,8 +53,6 @@ def _make_bare_verifier(log_file: Path) -> ScreenLocker: ``--status`` — just enough state for the verification mixins' methods to run, no Tk lock UI, no ``__init__`` side effects. """ - from screen_locker.screen_lock import ScreenLocker - verifier = object.__new__(ScreenLocker) verifier.log_file = log_file verifier.workout_data = {} diff --git a/screen_locker/_stream_cache.py b/screen_locker/_stream_cache.py index 622c28e..8490ea6 100644 --- a/screen_locker/_stream_cache.py +++ b/screen_locker/_stream_cache.py @@ -128,7 +128,7 @@ class DeviceLogCache: outcome = _poke_credit.credit_sessions( [(rid, payload) for rid, payload, _hlc in pending] ) - except Exception: + except Exception: # pylint: disable=broad-exception-caught # one bad event must not drop the stream # Not a stream problem: keep the connection (a reconnect would # re-download the snapshot and fail the same way). Nothing is # marked seen, so the next event retries. diff --git a/screen_locker/_ui_flows.py b/screen_locker/_ui_flows.py index dc07eff..8ab1a0c 100644 --- a/screen_locker/_ui_flows.py +++ b/screen_locker/_ui_flows.py @@ -2,7 +2,7 @@ from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module +from concurrent.futures import ThreadPoolExecutor from typing import TYPE_CHECKING from screen_locker import _manual_workout diff --git a/screen_locker/_ui_flows_relaxed.py b/screen_locker/_ui_flows_relaxed.py index ad2b19e..7eed9b6 100644 --- a/screen_locker/_ui_flows_relaxed.py +++ b/screen_locker/_ui_flows_relaxed.py @@ -2,7 +2,7 @@ from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module +from concurrent.futures import ThreadPoolExecutor from screen_locker._weekly_check import ( RELAXED_DAY_SKIP_TYPE, diff --git a/screen_locker/_web_server.py b/screen_locker/_web_server.py index 2421d4f..02f6bf5 100644 --- a/screen_locker/_web_server.py +++ b/screen_locker/_web_server.py @@ -89,7 +89,7 @@ def _decision_limit(query: str) -> int: class _Handler(BaseHTTPRequestHandler): """Serve the status payloads and the static frontend bundle (read-only).""" - def log_message(self, fmt: str, *args: object) -> None: + def log_message(self, fmt: str, /, *args: object) -> None: """Route the default request log to ``logging`` at debug level. Args: diff --git a/screen_locker/_weekly_check.py b/screen_locker/_weekly_check.py index 7b66a71..89cef89 100644 --- a/screen_locker/_weekly_check.py +++ b/screen_locker/_weekly_check.py @@ -223,7 +223,7 @@ def count_weekly_workouts( exc, ) continue - if not (week_start <= entry_date <= today_date): + if entry_date < week_start or entry_date > today_date: continue count += count_day_credits(date_str, entries) return count diff --git a/screen_locker/_workout_app.py b/screen_locker/_workout_app.py index 8f8e254..e0474eb 100644 --- a/screen_locker/_workout_app.py +++ b/screen_locker/_workout_app.py @@ -49,9 +49,11 @@ if TYPE_CHECKING: stdout: IO[str] | None returncode: int | None - def terminate(self) -> None: ... + def terminate(self) -> None: + """Ask the process to stop.""" - def wait(self) -> int: ... + def wait(self) -> int: + """Block until the process exits; return its exit code.""" class _Spawner(Protocol): """``subprocess.Popen``, narrowed to the one call shape used here.""" diff --git a/screen_locker/_workout_credit.py b/screen_locker/_workout_credit.py index fda3978..a12aba3 100644 --- a/screen_locker/_workout_credit.py +++ b/screen_locker/_workout_credit.py @@ -183,7 +183,7 @@ class WorkoutCreditMixin: """ try: _credit_notify.notify_workout_credit(before, after, self.log_file) - except Exception: + except Exception: # pylint: disable=broad-exception-caught # notification must never fail the credit _logger.exception( "Workout credit notification failed — the credit itself is " "applied; only the desktop notification is missing" diff --git a/screen_locker/_workout_handoff.py b/screen_locker/_workout_handoff.py index 46f29c0..bebb619 100644 --- a/screen_locker/_workout_handoff.py +++ b/screen_locker/_workout_handoff.py @@ -41,7 +41,7 @@ def lock_grab_handoff(lock: LockWindow) -> GrabHandoff: def release() -> None: # Stop first: a watchdog tick between the release and the app's own # grab would take the screen straight back off it. - lock._recovery.stop() + lock.recovery.stop() # Claim.grab_held false BEFORE the release (fail closed), so # leisure-watch never reads "held" while the workout app has input. # reacquire() -> grab_input() sets it true once the grab lands. diff --git a/screen_locker/_workout_session.py b/screen_locker/_workout_session.py index f0a8e6c..02c2e8c 100644 --- a/screen_locker/_workout_session.py +++ b/screen_locker/_workout_session.py @@ -38,9 +38,11 @@ if TYPE_CHECKING: stdout: IO[str] | None returncode: int | None - def poll(self) -> int | None: ... + def poll(self) -> int | None: + """Return the exit code, or ``None`` while it still runs.""" - def terminate(self) -> None: ... + def terminate(self) -> None: + """Ask the process to stop.""" __all__ = ["WorkoutSession"] diff --git a/screen_locker/poke_server.py b/screen_locker/poke_server.py index b2b7de1..2734245 100644 --- a/screen_locker/poke_server.py +++ b/screen_locker/poke_server.py @@ -69,7 +69,7 @@ def _start_stream(server: PokeServer) -> None: # Same lock as the HTTP handler: a poke and a streamed copy of one # session must never be credited concurrently. SessionStream(server.gate).start() - except Exception: + except Exception: # pylint: disable=broad-exception-caught # daemon boundary: logged, LAN poke must survive _logger.exception( "Could not start the Firebase session stream; off-LAN workouts will " "only be credited by the 15-min workout-sync until the daemon restarts" diff --git a/screen_locker/tests/test_workout_handoff.py b/screen_locker/tests/test_workout_handoff.py index 47cd560..a7b2916 100644 --- a/screen_locker/tests/test_workout_handoff.py +++ b/screen_locker/tests/test_workout_handoff.py @@ -8,6 +8,9 @@ the user Ctrl+Alt+F2 out of the obligation the lock exists to enforce. from __future__ import annotations import tkinter as tk +from unittest.mock import MagicMock + +from gatelock import LockWindow from screen_locker._workout_handoff import lock_grab_handoff, start_workout @@ -28,7 +31,7 @@ class _FakeLock: """A lock window that records the order of grab operations.""" def __init__(self, *, release_raises: bool = False) -> None: - self._recovery = _FakeRecovery() + self.recovery = _FakeRecovery() self.events: list[str] = [] self.closed = False self._release_raises = release_raises @@ -49,7 +52,7 @@ class _FakeLock: def grab_input(self) -> None: self.events.append("grab_input") - self._recovery.running = True + self.recovery.running = True def close(self) -> None: self.closed = True @@ -62,16 +65,16 @@ class TestLockGrabHandoff: """A tick between release and the app's own grab would steal it back.""" lock = _FakeLock() lock_grab_handoff(lock).release() - assert lock._recovery.events == ["stop"] + assert lock.recovery.events == ["stop"] assert lock.events == ["held=False", "grab_release"] - assert not lock._recovery.running + assert not lock.recovery.running def test_reacquire_restores_the_grab_and_the_watchdog(self) -> None: handoff = lock_grab_handoff(lock := _FakeLock()) handoff.release() handoff.reacquire() assert lock.events == ["held=False", "grab_release", "grab_input"] - assert lock._recovery.running + assert lock.recovery.running def test_never_closes_the_lock_window(self) -> None: """close() also restores VT switching -- that is the escape hatch.""" @@ -116,3 +119,31 @@ class TestStartWorkout: _workout_handoff, "launch_workout_app", lambda _h: _Result() ) assert "checking whether it counted" in start_workout(_FakeLock()) + + +class _Recorder: + """A lock stand-in that records every attribute the handoff reads.""" + + def __init__(self) -> None: + self.names: set[str] = set() + + def __getattr__(self, name: str) -> MagicMock: + self.names.add(name) + return MagicMock() + + +class TestAgainstTheRealGatelockApi: + """The fakes above cannot notice a renamed or private gatelock attribute.""" + + def test_the_handoff_only_reads_attributes_a_real_lock_window_has(self) -> None: + """``lock._recovery`` raised AttributeError live while fakes passed.""" + spy = _Recorder() + handoff = lock_grab_handoff(spy) + handoff.release() + handoff.reacquire() + missing = { + name + for name in spy.names + if name != "root" and not hasattr(LockWindow, name) + } + assert not missing diff --git a/scripts/_popup_form_check.py b/scripts/_popup_form_check.py index 7985557..602f712 100644 --- a/scripts/_popup_form_check.py +++ b/scripts/_popup_form_check.py @@ -130,7 +130,7 @@ def grab_holder(root: tk.Tk) -> str: class FormHost(UIWidgetsMixin, ManualWorkoutDialogMixin): """The two hooks the mixin requires, over a real Tk root.""" - def __init__(self, root: tk.Tk) -> None: + def __init__(self, root: tk.Tk, log_file: Path) -> None: """Set only the attributes the form actually reads.""" self.root = root self.demo_mode = False @@ -139,8 +139,7 @@ class FormHost(UIWidgetsMixin, ManualWorkoutDialogMixin): # week's manual workouts are used up, so pointing at the live log made # this check pass or fail on how much the user had exercised that # week rather than on whether the widgets survive a recovery tick. - self._log_dir = TemporaryDirectory() - self.log_file = Path(self._log_dir.name) / "log.json" + self.log_file = log_file self.log_file.write_text("{}") self._colors = LockConfig() self.container = FrameGroup.single(root, bg=self._colors.palette.bg) @@ -178,9 +177,15 @@ def radios(widget: tk.Misc) -> list[tk.Radiobutton]: def check_real_form_survives() -> bool: """Build the production form, pick "Other", hold it through the ticks.""" + with TemporaryDirectory() as scratch: + return _check_form(Path(scratch) / "log.json") + + +def _check_form(log_file: Path) -> bool: + """The body of :func:`check_real_form_survives`, over a throwaway log.""" root = tk.Tk() root.geometry("1600x1200+0+0") - host = FormHost(root) + host = FormHost(root, log_file) host.render() root.update() diff --git a/scripts/_preview_sim.py b/scripts/_preview_sim.py index fad5c7e..34f803f 100644 --- a/scripts/_preview_sim.py +++ b/scripts/_preview_sim.py @@ -33,8 +33,8 @@ def stamp(day: date, hhmm_text: str) -> float: return moment.timestamp() -def _iso(stamp: float) -> str: - return datetime.fromtimestamp(stamp).astimezone().isoformat() +def _iso(moment: float) -> str: + return datetime.fromtimestamp(moment).astimezone().isoformat() def write_walk(folder: Path, day: date, start: str, minutes: int) -> None: diff --git a/scripts/_silent_cli.py b/scripts/_silent_cli.py new file mode 100644 index 0000000..86cc4cf --- /dev/null +++ b/scripts/_silent_cli.py @@ -0,0 +1,40 @@ +"""The argv loop shared by the Python and Dart silent-failure hooks. + +Both hooks read every file named on the command line, report what their +``check_source`` finds, and fail closed on a file they cannot read. Only the +per-file check differs, so it is the parameter. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +import sys +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger = logging.getLogger(__name__) + + +def run_hook(check_source: Callable[[Path, str], list[str]]) -> int: + """Return 1 if ``check_source`` flags any file in ``sys.argv``, else 0.""" + problems: list[str] = [] + for filename in sys.argv[1:]: + path = Path(filename) + try: + source = path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + _logger.warning( + "Could not read %s (%s) — reporting it as a problem rather than " + "skipping it, since its handlers cannot be checked", + path, + exc, + ) + problems.append(f"{path}: could not read ({exc})") + continue + problems.extend(check_source(path, source)) + for problem in problems: + sys.stdout.write(f"{problem}\n") + return 1 if problems else 0 diff --git a/scripts/_xvfb_reexec.py b/scripts/_xvfb_reexec.py index 4f3ffb6..8e384ff 100644 --- a/scripts/_xvfb_reexec.py +++ b/scripts/_xvfb_reexec.py @@ -13,12 +13,17 @@ forever. from __future__ import annotations +import contextlib import logging import os from pathlib import Path import shutil import subprocess import sys +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterator _logger = logging.getLogger(__name__) @@ -63,8 +68,7 @@ def reexec_under_xvfb( repo_root = Path(__file__).resolve().parent.parent # Absolute argv, no shell. target = ["-m", module] if module is not None else [str(script)] - server, display = _start_xvfb(xvfb) - try: + with _xvfb_display(xvfb) as display: if display is None: _logger.error("Xvfb exited before it was ready to accept clients") return 1 @@ -73,26 +77,28 @@ def reexec_under_xvfb( return subprocess.run( [sys.executable, *target], check=False, cwd=repo_root, env=env ).returncode - finally: - server.terminate() - server.wait(timeout=5) -def _start_xvfb(xvfb: str) -> tuple[subprocess.Popen[bytes], str | None]: - """Start Xvfb on a free display; return it and ``:N`` once it is ready. +@contextlib.contextmanager +def _xvfb_display(xvfb: str) -> Iterator[str | None]: + """Run Xvfb on a free display; yield ``:N`` once ready (``None`` if it died). ``-displayfd`` makes Xvfb pick a free display and write its number only once it accepts clients. ``xvfb-run`` instead sleeps a fixed 3 s after starting it, which was most of this pre-commit hook's runtime. """ read_fd, write_fd = os.pipe() - server = subprocess.Popen( + with subprocess.Popen( [xvfb, "-displayfd", str(write_fd), "-nolisten", "tcp", *_SCREEN_ARG], pass_fds=(write_fd,), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - ) - os.close(write_fd) - with os.fdopen(read_fd) as ready: - number = ready.readline().strip() - return server, f":{number}" if number else None + ) as server: + os.close(write_fd) + try: + with os.fdopen(read_fd) as ready: + number = ready.readline().strip() + yield f":{number}" if number else None + finally: + server.terminate() + server.wait(timeout=5) diff --git a/scripts/check_requirements_sync.py b/scripts/check_requirements_sync.py index b768844..a40e9d1 100755 --- a/scripts/check_requirements_sync.py +++ b/scripts/check_requirements_sync.py @@ -88,7 +88,7 @@ def _top_level_imports(module: Path) -> set[str]: if isinstance(node, ast.Import): for alias in node.names: imported.add(alias.name.split(".", 1)[0]) - elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + elif isinstance(node, ast.ImportFrom) and not node.level and node.module: imported.add(node.module.split(".", 1)[0]) return imported diff --git a/scripts/check_silent_failures.py b/scripts/check_silent_failures.py index e7b20d4..9674eb4 100755 --- a/scripts/check_silent_failures.py +++ b/scripts/check_silent_failures.py @@ -24,8 +24,13 @@ from __future__ import annotations import ast import logging -from pathlib import Path import sys +from typing import TYPE_CHECKING + +from scripts._silent_cli import run_hook + +if TYPE_CHECKING: + from pathlib import Path _logger = logging.getLogger(__name__) @@ -61,7 +66,9 @@ def _is_silent_exit(node: ast.AST) -> bool: if not is_exit or len(node.args) != 1: return False arg = node.args[0] - return isinstance(arg, ast.Constant) and arg.value == 0 + return isinstance(arg, ast.Constant) and ( + isinstance(arg.value, int) and not arg.value + ) def _find_silent_exits(tree: ast.AST) -> list[ast.Call]: @@ -177,24 +184,7 @@ def _check_source(path: Path, source: str) -> list[str]: def main() -> int: """Return 1 if any file swallows an exception silently, else 0.""" - problems: list[str] = [] - for filename in sys.argv[1:]: - path = Path(filename) - try: - source = path.read_text(encoding="utf-8", errors="replace") - except OSError as exc: - _logger.warning( - "Could not read %s (%s) — reporting it as a problem rather than " - "skipping it, since its handlers cannot be checked", - path, - exc, - ) - problems.append(f"{path}: could not read ({exc})") - continue - problems.extend(_check_source(path, source)) - for problem in problems: - sys.stdout.write(f"{problem}\n") - return 1 if problems else 0 + return run_hook(_check_source) if __name__ == "__main__": diff --git a/scripts/check_silent_failures_dart.py b/scripts/check_silent_failures_dart.py index 3df9fec..aa503df 100755 --- a/scripts/check_silent_failures_dart.py +++ b/scripts/check_silent_failures_dart.py @@ -30,9 +30,14 @@ not a skip, exactly like the Python hook's SyntaxError branch. from __future__ import annotations import logging -from pathlib import Path import re import sys +from typing import TYPE_CHECKING + +from scripts._silent_cli import run_hook + +if TYPE_CHECKING: + from pathlib import Path _logger = logging.getLogger(__name__) @@ -113,7 +118,7 @@ def _block_after(source: str, start: int) -> tuple[str, int] | None: depth += 1 elif source[idx] == "}": depth -= 1 - if depth == 0: + if not depth: return source[brace : idx + 1], idx return None @@ -187,24 +192,7 @@ def _check_source(path: Path, source: str) -> list[str]: def main() -> int: """Return 1 if any Dart file swallows an error silently, else 0.""" - problems: list[str] = [] - for filename in sys.argv[1:]: - path = Path(filename) - try: - source = path.read_text(encoding="utf-8", errors="replace") - except OSError as exc: - _logger.warning( - "Could not read %s (%s) — reporting it as a problem rather " - "than skipping it, since its handlers cannot be checked", - path, - exc, - ) - problems.append(f"{path}: could not read ({exc})") - continue - problems.extend(_check_source(path, source)) - for problem in problems: - sys.stdout.write(f"{problem}\n") - return 1 if problems else 0 + return run_hook(_check_source) if __name__ == "__main__": diff --git a/scripts/demo_workout_handoff.py b/scripts/demo_workout_handoff.py index 6d206bd..40c5f2d 100644 --- a/scripts/demo_workout_handoff.py +++ b/scripts/demo_workout_handoff.py @@ -29,20 +29,16 @@ also still works -- but the hatch above is the one this demo is testing. One thing a green run does NOT prove: under the demo's local grab gatelock's watchdog early-returns and never re-takes the grab, so the 1000ms steal-back -that makes ``_recovery.stop()`` load-bearing in production is not exercised. +that makes ``recovery.stop()`` load-bearing in production is not exercised. - python3 scripts/demo_workout_handoff.py + python3 -m scripts.demo_workout_handoff # from the repo root """ from __future__ import annotations import logging -from pathlib import Path -import sys import tkinter as tk -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - from gatelock import GrabPolicy, LockConfig, LockWindow from screen_locker._workout_app import ProcessHooks, workout_app_binary @@ -59,6 +55,7 @@ class _DemoHooks: self.session: WorkoutSession | None = None def build_surface(self, parent: tk.Misc, _surface: object) -> None: + """Draw the status line and the release/reacquire buttons.""" tk.Label( parent, text="DEMO LOCK — local grab, VT switching still enabled", @@ -115,11 +112,14 @@ class _DemoHooks: if self.lock is not None: # pragma: no cover self.lock.close() - def teardown_surface(self, _surface: object) -> None: ... + def teardown_surface(self, _surface: object) -> None: + """Nothing to tear down in the demo.""" - def on_focus_ready(self, _surface: object) -> None: ... + def on_focus_ready(self, _surface: object) -> None: + """The demo needs no focus handling.""" - def on_close(self) -> None: ... + def on_close(self) -> None: + """The demo needs no close handling.""" def main() -> int: diff --git a/scripts/preview_shutdown.py b/scripts/preview_shutdown.py index 422caa1..5014e7f 100755 --- a/scripts/preview_shutdown.py +++ b/scripts/preview_shutdown.py @@ -98,7 +98,6 @@ def _simulate(args: argparse.Namespace, tmp: Path) -> None: if args.rest_day: eve = datetime.combine(day - timedelta(days=1), time(12)).astimezone() _say(f" declare: {declare(day, now=eve, path=rest_file).reason}") - rest = is_rest_day(day, rest_file) done = {n for n in (args.done or "").split(",") if n} flat = {e.name: e.name in done for e in _earned.flat_earners()} first = stamp(day, args.first_done) if args.first_done else None @@ -109,7 +108,7 @@ def _simulate(args: argparse.Namespace, tmp: Path) -> None: day, flat, day_credit_count(log, day), - rest_day=rest, + rest_day=is_rest_day(day, rest_file), first_done=min(times, default=None), sick_day=sick_pool(day, sick_day=args.sick_day), ) @@ -130,9 +129,8 @@ def _sequence(args: argparse.Namespace, tmp: Path) -> None: vars(module)["GRACE_STATE_FILE"] = tmp / "grace_floor.json" use_temp_pool(tmp) done = {e.name: False for e in _earned.flat_earners()} - start = derive(DayInputs(day, done)) - sim = SimShutdown(start.minutes) - _say(f"[sequence] {day} start {hhmm(start.minutes)} (nothing done, not sick)") + sim = SimShutdown(derive(DayInputs(day, done)).minutes) + _say(f"[sequence] {day} start {hhmm(sim.config[0])} (nothing done, not sick)") workouts, first, went_late = 0, None, False # The real target reads today's real ledgers; the sim's are ``done``. vars(_sick_floor)["sick_target"] = lambda d, _log: ( @@ -177,12 +175,10 @@ def main() -> int: parser.add_argument("--sequence", help="name@HH:MM,... live passes in order") parser.add_argument("--log", type=Path, default=_LOG, help="log.json (read-only)") args = parser.parse_args() - ladder = getattr(earned_time, "on_ladder", None) _say(f"earned_time from {Path(earned_time.__file__).parent}") if args.day is None: today = date.fromisoformat(today_str()) - on = ladder(today) if ladder else False - _say(f" ladder in force today: {on}") + _say(f" ladder in force today: {earned_time.on_ladder(today)}") live = read_shutdown_config(SHUTDOWN_CONFIG_FILE) _say( f" live config: {tuple(hhmm(m) for m in live) if live else 'unreadable'}" @@ -195,7 +191,7 @@ def main() -> int: _report(f"today {label}", derive(replace(inputs, sick_day=sick))) return 0 day = date.fromisoformat(args.day) - _say(f" ladder in force on {day}: {ladder(day) if ladder else False}") + _say(f" ladder in force on {day}: {earned_time.on_ladder(day)}") with tempfile.TemporaryDirectory(prefix="preview_shutdown_") as raw: if args.sequence: _sequence(args, Path(raw)) diff --git a/scripts/refresh_refactor_todo.py b/scripts/refresh_refactor_todo.py index 3579eea..45dcbca 100755 --- a/scripts/refresh_refactor_todo.py +++ b/scripts/refresh_refactor_todo.py @@ -110,7 +110,7 @@ def commits_last_year(path: Path) -> int: text=True, check=False, ) - if result.returncode != 0: + if result.returncode: _logger.warning( "git log failed for %s (%s) — counting 0 commits, so its ROI " "rank will read lower than it should", diff --git a/scripts/rekey_log_entry.py b/scripts/rekey_log_entry.py index 410950c..73dcbf3 100755 --- a/scripts/rekey_log_entry.py +++ b/scripts/rekey_log_entry.py @@ -71,7 +71,7 @@ def main() -> int: log_path = Path(args.log_file) try: - log = json.loads(log_path.read_text()) + log = json.loads(log_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: _logger.exception("Cannot read %s", log_path) _report(f"ERROR: cannot read {log_path}: {exc}") @@ -83,7 +83,7 @@ def main() -> int: _report(f"ERROR: {exc} — nothing changed") return 1 try: - log_path.write_text(json.dumps(log, indent=2)) + log_path.write_text(json.dumps(log, indent=2), encoding="utf-8") except OSError as exc: _logger.exception("Could not write %s", log_path) _report(f"ERROR: could not write {log_path}: {exc}") diff --git a/scripts/split_dart_test.py b/scripts/split_dart_test.py index 93b5b94..3f41477 100644 --- a/scripts/split_dart_test.py +++ b/scripts/split_dart_test.py @@ -33,7 +33,7 @@ def top_level_groups(lines: list[str]) -> list[int]: def split(path: Path, specs: list[tuple[Path, int]]) -> None: """Write each spec's group range to its own file, trimming the source.""" - lines = path.read_text().splitlines(keepends=True) + lines = path.read_text(encoding="utf-8").splitlines(keepends=True) groups = top_level_groups(lines) main_idx = next(i for i, ln in enumerate(lines) if ln.startswith("void main(")) header = "".join(lines[:main_idx]) @@ -47,16 +47,17 @@ def split(path: Path, specs: list[tuple[Path, int]]) -> None: ) trailer = "".join(lines[main_end + 1 :]) - bounds = [i for _, i in specs] + [len(groups)] - for (out, start), end in zip(specs, bounds[1:], strict=False): - first = groups[start] + ends = [i for _, i in specs[1:]] + [len(groups)] + for (out, start), end in zip(specs, ends, strict=True): last = groups[end] if end < len(groups) else main_end - body = "".join(lines[first:last]).rstrip("\n") - out.write_text(f"{header}void main() {{\n{preamble}{body}\n}}\n{trailer}") + body = "".join(lines[groups[start] : last]).rstrip("\n") + out.write_text( + f"{header}void main() {{\n{preamble}{body}\n}}\n{trailer}", encoding="utf-8" + ) - keep = groups[specs[0][1]] path.write_text( - "".join(lines[:keep]).rstrip("\n") + "\n}\n" + trailer, + "".join(lines[: groups[specs[0][1]]]).rstrip("\n") + "\n}\n" + trailer, + encoding="utf-8", ) diff --git a/scripts/verify_workout_credited.py b/scripts/verify_workout_credited.py index 8c2622c..f0a6673 100755 --- a/scripts/verify_workout_credited.py +++ b/scripts/verify_workout_credited.py @@ -63,7 +63,7 @@ def main() -> int: return 1 try: - log = json.loads(log_path.read_text()) + log = json.loads(log_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: _logger.exception( "Could not read %s — cannot tell whether the workout was "