diff --git a/screen_locker/_adb_transport.py b/screen_locker/_adb_transport.py new file mode 100644 index 0000000..e08337f --- /dev/null +++ b/screen_locker/_adb_transport.py @@ -0,0 +1,137 @@ +"""ADB transport for phone verification: connect, reconnect, run commands. + +Split out of :mod:`screen_locker._phone_verification` to keep every file under +the 250-line cap. Composed back into ``PhoneVerificationMixin`` there, so +callers see no change. + +Everything here is plumbing: reaching the phone and running a command on it. +Deciding whether what came back counts as a workout stays in +``_phone_verification``. +""" + +from __future__ import annotations + +from concurrent.futures import ( # pylint: disable=no-name-in-module + ThreadPoolExecutor, + as_completed, +) +import contextlib +import logging +import shutil +import socket +import subprocess + +from screen_locker._constants import ADB_TIMEOUT + +_logger = logging.getLogger(__name__) + + +class AdbTransportMixin: + """Finds the phone over USB or wifi and runs adb commands against it.""" + + def _run_adb(self, args: list[str]) -> tuple[bool, str]: + """Run an ADB command and return success flag and stdout.""" + adb = shutil.which("adb") or "adb" + # When multiple devices are connected (e.g. USB + wireless), pin to + # the wireless device's serial to avoid "more than one device" errors. + _discovery_cmds = {"devices", "connect", "disconnect", "kill-server"} + serial = ( + self._get_wireless_serial() + if args and args[0] not in _discovery_cmds + else None + ) + serial_args = ["-s", serial] if serial else [] + try: + result = subprocess.run( + [adb, *serial_args, *args], + capture_output=True, + text=True, + timeout=ADB_TIMEOUT, + check=False, + ) + except (FileNotFoundError, OSError) as exc: + _logger.warning("ADB not available: %s", exc) + return False, "" + except subprocess.TimeoutExpired: + _logger.warning("ADB command timed out: %s", args) + return False, "" + return not result.returncode, result.stdout + + def _adb_shell(self, command: str, *, root: bool = False) -> tuple[bool, str]: + """Run a shell command on the connected Android device.""" + if root: + return self._run_adb(["shell", "su", "-c", command]) + return self._run_adb(["shell", command]) + + def _get_wireless_serial(self) -> str | None: + """Return the serial (ip:port) of the first connected wireless ADB device.""" + success, output = self._run_adb(["devices"]) + if not success: + return None + for line in output.strip().split("\n")[1:]: + parts = line.split() + if parts and ":" in parts[0] and "device" in line and "offline" not in line: + return parts[0] + return None + + def _has_adb_device(self) -> bool: + """Return True if adb devices shows at least one connected device.""" + success, output = self._run_adb(["devices"]) + if not success: + return False + lines = output.strip().split("\n")[1:] + return any("device" in line and "offline" not in line for line in lines) + + def _try_adb_connect(self, address: str) -> bool: + """Run adb connect to address. Returns True on success.""" + _, output = self._run_adb(["connect", address]) + lower = output.lower() + return "connected" in lower and "unable" not in lower and "failed" not in lower + + def _get_local_subnet_prefix(self) -> str | None: + """Detect the local /24 network prefix (e.g. '192.168.1').""" + with ( + contextlib.suppress(OSError), + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, + ): + sock.connect(("8.8.8.8", 80)) + return ".".join(sock.getsockname()[0].split(".")[:3]) + return None + + def _try_wireless_reconnect(self) -> bool: + """Scan local /24 subnet on port 5555 and attempt ADB connect to phone.""" + prefix = self._get_local_subnet_prefix() + if prefix is None: + _logger.info("Could not determine local subnet for wireless scan") + return False + + def probe(i: int) -> bool: + ip = f"{prefix}.{i}" + with ( + contextlib.suppress(OSError), + socket.create_connection((ip, 5555), timeout=0.5), + ): + if self._try_adb_connect(f"{ip}:5555"): + return self._has_adb_device() + return False + + _logger.info("Scanning %s.1-254:5555 for phone...", prefix) + with ThreadPoolExecutor(max_workers=64) as executor: + for future in as_completed( + executor.submit(probe, i) for i in range(1, 255) + ): + if future.result(): + return True + return False + + def _is_phone_connected(self) -> bool: + """Check if an Android device is connected via ADB. + + If no device is visible, attempts wireless reconnection via subnet scan. + """ + if self._has_adb_device(): + return True + _logger.info("No ADB device detected — attempting wireless reconnect...") + return self._try_wireless_reconnect() + + # ── ADB verification ────────────────────────────────────────────────────── diff --git a/screen_locker/_http_workout_fetch.py b/screen_locker/_http_workout_fetch.py new file mode 100644 index 0000000..8f77010 --- /dev/null +++ b/screen_locker/_http_workout_fetch.py @@ -0,0 +1,94 @@ +"""HTTP fallback for phone verification: find the app's server, fetch JSON. + +Split out of :mod:`screen_locker._phone_verification` to keep every file under +the 250-line cap. Composed back into ``PhoneVerificationMixin`` there, so +callers see no change. + +Used when adb cannot reach the phone but it is on the same wifi: the workout +app serves the same JSON over ``WORKOUT_HTTP_PORT``. +""" + +from __future__ import annotations + +from concurrent.futures import ( # pylint: disable=no-name-in-module + ThreadPoolExecutor, + as_completed, +) +import contextlib +from http import client as _http_client +import json +import logging +import socket + +from screen_locker._constants import WORKOUT_HTTP_PORT + +_HTTPConnection = _http_client.HTTPConnection +_HTTPException = _http_client.HTTPException +_HTTP_OK = _http_client.OK + +_logger = logging.getLogger(__name__) + + +class HttpWorkoutFetchMixin: + """Scans the local subnet for the workout app's HTTP server.""" + + def _scan_for_http_server(self) -> str | None: + """Scan local /24 subnet for the workout app HTTP server on port 8765. + + Returns the first reachable URL or None. + """ + prefix = self._get_local_subnet_prefix() + if prefix is None: + return None + + def probe(i: int) -> str | None: + ip = f"{prefix}.{i}" + with ( + contextlib.suppress(OSError), + socket.create_connection((ip, WORKOUT_HTTP_PORT), timeout=0.3), + ): + return f"http://{ip}:{WORKOUT_HTTP_PORT}/workout" + return None + + _logger.info( + "Scanning %s.1-254:%d for workout app...", prefix, WORKOUT_HTTP_PORT + ) + with ThreadPoolExecutor(max_workers=64) as executor: + for future in as_completed( + executor.submit(probe, i) for i in range(1, 255) + ): + result = future.result() + if result is not None: + return result + return None + + def _fetch_http_workout(self) -> dict | None: + """Fetch workout JSON from the app's HTTP server on the local network. + + Uses http.client directly to avoid urllib URL-open security lint rules. + The URL is always http://:8765/workout — no user input involved. + """ + url = self._scan_for_http_server() + if url is None: + return None + # url is always "http://:/workout" — constructed internally. + try: + _, _, hostport = url.partition("://") + host, _, path = hostport.partition("/") + hostname, _, port_str = host.partition(":") + conn = _HTTPConnection(hostname, int(port_str), timeout=5) + conn.request("GET", f"/{path}") + resp = conn.getresponse() + if resp.status != _HTTP_OK: + return None + return json.loads(resp.read().decode()) + except (_HTTPException, OSError, ValueError, json.JSONDecodeError) as exc: + _logger.warning( + "HTTP fetch of today's workout from %s failed (%s) — no data " + "from the phone's HTTP server, so verification falls through", + url, + exc, + ) + return None + + # ── Main verification entry point ───────────────────────────────────────── diff --git a/screen_locker/_phone_verification.py b/screen_locker/_phone_verification.py index a690112..1c25d56 100644 --- a/screen_locker/_phone_verification.py +++ b/screen_locker/_phone_verification.py @@ -2,29 +2,21 @@ from __future__ import annotations -from concurrent.futures import ( # pylint: disable=no-name-in-module - ThreadPoolExecutor, - as_completed, -) -import contextlib from datetime import date, datetime, timedelta, timezone from http import client as _http_client import json import logging from pathlib import Path -import shutil -import socket -import subprocess import tempfile import time +from screen_locker._adb_transport import AdbTransportMixin from screen_locker._constants import ( - ADB_TIMEOUT, MIN_WORKOUT_DURATION_MINUTES, WORKOUT_APP_JSON_REMOTES, WORKOUT_DURATION_ACCEPT_MINUTES, - WORKOUT_HTTP_PORT, ) +from screen_locker._http_workout_fetch import HttpWorkoutFetchMixin from screen_locker._log_mixin import write_signed_entry from screen_locker._time_check import check_clock_skew from screen_locker._workout_sync import pull_synced_workout @@ -36,116 +28,9 @@ _HTTP_OK = _http_client.OK _logger = logging.getLogger(__name__) -class PhoneVerificationMixin: +class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): """Mixin providing phone-based workout verification via ADB.""" - def _run_adb(self, args: list[str]) -> tuple[bool, str]: - """Run an ADB command and return success flag and stdout.""" - adb = shutil.which("adb") or "adb" - # When multiple devices are connected (e.g. USB + wireless), pin to - # the wireless device's serial to avoid "more than one device" errors. - _discovery_cmds = {"devices", "connect", "disconnect", "kill-server"} - serial = ( - self._get_wireless_serial() - if args and args[0] not in _discovery_cmds - else None - ) - serial_args = ["-s", serial] if serial else [] - try: - result = subprocess.run( - [adb, *serial_args, *args], - capture_output=True, - text=True, - timeout=ADB_TIMEOUT, - check=False, - ) - except (FileNotFoundError, OSError) as exc: - _logger.warning("ADB not available: %s", exc) - return False, "" - except subprocess.TimeoutExpired: - _logger.warning("ADB command timed out: %s", args) - return False, "" - return not result.returncode, result.stdout - - def _adb_shell(self, command: str, *, root: bool = False) -> tuple[bool, str]: - """Run a shell command on the connected Android device.""" - if root: - return self._run_adb(["shell", "su", "-c", command]) - return self._run_adb(["shell", command]) - - def _get_wireless_serial(self) -> str | None: - """Return the serial (ip:port) of the first connected wireless ADB device.""" - success, output = self._run_adb(["devices"]) - if not success: - return None - for line in output.strip().split("\n")[1:]: - parts = line.split() - if parts and ":" in parts[0] and "device" in line and "offline" not in line: - return parts[0] - return None - - def _has_adb_device(self) -> bool: - """Return True if adb devices shows at least one connected device.""" - success, output = self._run_adb(["devices"]) - if not success: - return False - lines = output.strip().split("\n")[1:] - return any("device" in line and "offline" not in line for line in lines) - - def _try_adb_connect(self, address: str) -> bool: - """Run adb connect to address. Returns True on success.""" - _, output = self._run_adb(["connect", address]) - lower = output.lower() - return "connected" in lower and "unable" not in lower and "failed" not in lower - - def _get_local_subnet_prefix(self) -> str | None: - """Detect the local /24 network prefix (e.g. '192.168.1').""" - with ( - contextlib.suppress(OSError), - socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock, - ): - sock.connect(("8.8.8.8", 80)) - return ".".join(sock.getsockname()[0].split(".")[:3]) - return None - - def _try_wireless_reconnect(self) -> bool: - """Scan local /24 subnet on port 5555 and attempt ADB connect to phone.""" - prefix = self._get_local_subnet_prefix() - if prefix is None: - _logger.info("Could not determine local subnet for wireless scan") - return False - - def probe(i: int) -> bool: - ip = f"{prefix}.{i}" - with ( - contextlib.suppress(OSError), - socket.create_connection((ip, 5555), timeout=0.5), - ): - if self._try_adb_connect(f"{ip}:5555"): - return self._has_adb_device() - return False - - _logger.info("Scanning %s.1-254:5555 for phone...", prefix) - with ThreadPoolExecutor(max_workers=64) as executor: - for future in as_completed( - executor.submit(probe, i) for i in range(1, 255) - ): - if future.result(): - return True - return False - - def _is_phone_connected(self) -> bool: - """Check if an Android device is connected via ADB. - - If no device is visible, attempts wireless reconnection via subnet scan. - """ - if self._has_adb_device(): - return True - _logger.info("No ADB device detected — attempting wireless reconnect...") - return self._try_wireless_reconnect() - - # ── ADB verification ────────────────────────────────────────────────────── - def _pull_workout_app_json(self) -> dict | None: """Pull workout_result.json from the phone (ADB, no root needed). @@ -258,67 +143,6 @@ class PhoneVerificationMixin: # ── HTTP fallback (no ADB / developer options required) ─────────────────── - def _scan_for_http_server(self) -> str | None: - """Scan local /24 subnet for the workout app HTTP server on port 8765. - - Returns the first reachable URL or None. - """ - prefix = self._get_local_subnet_prefix() - if prefix is None: - return None - - def probe(i: int) -> str | None: - ip = f"{prefix}.{i}" - with ( - contextlib.suppress(OSError), - socket.create_connection((ip, WORKOUT_HTTP_PORT), timeout=0.3), - ): - return f"http://{ip}:{WORKOUT_HTTP_PORT}/workout" - return None - - _logger.info( - "Scanning %s.1-254:%d for workout app...", prefix, WORKOUT_HTTP_PORT - ) - with ThreadPoolExecutor(max_workers=64) as executor: - for future in as_completed( - executor.submit(probe, i) for i in range(1, 255) - ): - result = future.result() - if result is not None: - return result - return None - - def _fetch_http_workout(self) -> dict | None: - """Fetch workout JSON from the app's HTTP server on the local network. - - Uses http.client directly to avoid urllib URL-open security lint rules. - The URL is always http://:8765/workout — no user input involved. - """ - url = self._scan_for_http_server() - if url is None: - return None - # url is always "http://:/workout" — constructed internally. - try: - _, _, hostport = url.partition("://") - host, _, path = hostport.partition("/") - hostname, _, port_str = host.partition(":") - conn = _HTTPConnection(hostname, int(port_str), timeout=5) - conn.request("GET", f"/{path}") - resp = conn.getresponse() - if resp.status != _HTTP_OK: - return None - return json.loads(resp.read().decode()) - except (_HTTPException, OSError, ValueError, json.JSONDecodeError) as exc: - _logger.warning( - "HTTP fetch of today's workout from %s failed (%s) — no data " - "from the phone's HTTP server, so verification falls through", - url, - exc, - ) - return None - - # ── Main verification entry point ───────────────────────────────────────── - def _verify_phone_workout(self) -> tuple[str, str]: """Verify today's workout: GitHub sync first, ADB/HTTP as fallback. diff --git a/screen_locker/_shutdown.py b/screen_locker/_shutdown.py index 226be6b..3382307 100644 --- a/screen_locker/_shutdown.py +++ b/screen_locker/_shutdown.py @@ -2,21 +2,17 @@ from __future__ import annotations -import calendar -from datetime import datetime, timedelta, timezone -import json +from datetime import datetime, timezone import logging import subprocess from typing import TYPE_CHECKING from screen_locker._constants import ( ADJUST_SHUTDOWN_SCRIPT, - ALARM_DAYS, - RTCWAKE_BIN, SHUTDOWN_CONFIG_FILE, - SICK_DAY_STATE_FILE, - WAKE_AFTER_HOURS, ) +from screen_locker._shutdown_sick_state import SickDayStateMixin +from screen_locker._wake_alarm import WakeAlarmMixin if TYPE_CHECKING: from pathlib import Path @@ -52,7 +48,7 @@ def read_shutdown_config(path: Path) -> tuple[int, int, int] | None: ) -class ShutdownMixin: +class ShutdownMixin(SickDayStateMixin, WakeAlarmMixin): """Mixin providing shutdown schedule adjustment functionality.""" def _apply_earlier_shutdown(self, today: str) -> bool: @@ -139,102 +135,6 @@ class ShutdownMixin: ) return False - def _sick_mode_used_today(self) -> bool: - """Check if sick mode was already used today.""" - if not SICK_DAY_STATE_FILE.exists(): - return False - - try: - with SICK_DAY_STATE_FILE.open() as f: - state = json.load(f) - today = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d") - return state.get("date") == today - except (OSError, json.JSONDecodeError) as exc: - _logger.warning( - "Could not read sick-day state from %s: %s — treating sick mode " - "as UNUSED today, which may allow a second sick day", - SICK_DAY_STATE_FILE, - exc, - ) - return False - - def _save_sick_day_state( - self, - date: str, - orig_mon_wed: int, - orig_thu_sun: int, - ) -> bool: - """Save sick day state with original config values. - - Returns True if saved successfully, False otherwise. - """ - state = { - "date": date, - "original_mon_wed_hour": orig_mon_wed, - "original_thu_sun_hour": orig_thu_sun, - } - try: - with SICK_DAY_STATE_FILE.open("w") as f: - json.dump(state, f, indent=2) - except OSError as e: - _logger.warning("Failed to save sick day state: %s", e) - return False - - _logger.info("Saved sick day state for %s", date) - return True - - def _load_sick_day_state(self) -> tuple[str, int, int] | None: - """Load sick day state file. - - Returns (date, orig_mon_wed_hour, orig_thu_sun_hour) or None. - """ - with SICK_DAY_STATE_FILE.open() as f: - state = json.load(f) - date = state.get("date") - orig_mw = state.get("original_mon_wed_hour") - orig_ts = state.get("original_thu_sun_hour") - if date is None or orig_mw is None or orig_ts is None: - return None - return (str(date), int(orig_mw), int(orig_ts)) - - def _write_restored_config( - self, - orig_mw: int, - orig_ts: int, - state_date: str, - ) -> None: - """Write restored config values and clean up state file.""" - config_values = self._read_shutdown_config() - if config_values: - _, _, morning_end = config_values - _logger.info( - "Restoring original shutdown config from %s", - state_date, - ) - self._write_shutdown_config( - orig_mw, - orig_ts, - morning_end, - restore=True, - ) - SICK_DAY_STATE_FILE.unlink() - _logger.info("Removed stale sick day state from %s", state_date) - - def _restore_original_config_if_needed(self) -> None: - """Restore original config if sick day state is from a previous day.""" - if not SICK_DAY_STATE_FILE.exists(): - return - try: - loaded = self._load_sick_day_state() - if loaded is None: - return - state_date, orig_mw, orig_ts = loaded - today = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d") - if state_date != today: - self._write_restored_config(orig_mw, orig_ts, state_date) - except (OSError, json.JSONDecodeError) as e: - _logger.warning("Error checking sick day state: %s", e) - def _read_shutdown_config(self) -> tuple[int, int, int] | None: """Read shutdown config. Returns (mw_hour, ts_hour, me_hour) or None.""" return read_shutdown_config(SHUTDOWN_CONFIG_FILE) @@ -310,75 +210,3 @@ class ShutdownMixin: result.stdout.strip(), ) return True - - # ------------------------------------------------------------------ - # rtcwake integration for weekend wake alarm - # ------------------------------------------------------------------ - - @staticmethod - def _is_tomorrow_alarm_day() -> bool: - """Check if tomorrow is an alarm day.""" - tomorrow = datetime.now(tz=timezone.utc) + timedelta(days=1) - return tomorrow.weekday() in ALARM_DAYS - - @staticmethod - def _compute_wake_timestamp() -> int: - """Compute the UTC epoch timestamp for the next wake alarm. - - Returns: - Epoch seconds WAKE_AFTER_HOURS from now. - """ - wake_time = datetime.now(tz=timezone.utc) + timedelta( - hours=WAKE_AFTER_HOURS, - ) - return calendar.timegm(wake_time.utctimetuple()) - - @staticmethod - def _schedule_rtcwake() -> bool: - """Set rtcwake to power on the PC after WAKE_AFTER_HOURS. - - Uses ``rtcwake -m disk`` to hibernate immediately while programming - the RTC to restore power at wake_epoch. Hibernate is completely - silent and dark (state written to swap file), making it suitable - when the PC is in a bedroom. - - Returns: - True if rtcwake was set successfully, False otherwise. - """ - wake_epoch = ShutdownMixin._compute_wake_timestamp() - cmd = [ - "/usr/bin/sudo", - RTCWAKE_BIN, - "-m", - "disk", - "-t", - str(wake_epoch), - ] - try: - subprocess.run( - cmd, - check=True, - capture_output=True, - text=True, - ) - except subprocess.SubprocessError as exc: - _logger.warning("Failed to set rtcwake: %s", exc) - return False - _logger.info( - "rtcwake set: PC will wake at epoch %d", - wake_epoch, - ) - return True - - def schedule_wake_if_needed(self) -> bool: - """Schedule rtcwake if tomorrow is an alarm day. - - Call this at shutdown time. - - Returns: - True if wake was scheduled, False if not needed or failed. - """ - if not self._is_tomorrow_alarm_day(): - _logger.info("Tomorrow is not an alarm day — skipping rtcwake") - return False - return self._schedule_rtcwake() diff --git a/screen_locker/_shutdown_sick_state.py b/screen_locker/_shutdown_sick_state.py new file mode 100644 index 0000000..1638734 --- /dev/null +++ b/screen_locker/_shutdown_sick_state.py @@ -0,0 +1,119 @@ +"""Sick-day shutdown state: persistence and next-day restoration. + +Split out of :mod:`screen_locker._shutdown` to keep every file under the +250-line cap. Composed back into ``ShutdownMixin`` there, so callers see no +change. + +Tracks the shutdown hours that were in force before a sick day moved them +earlier, so the following day can put them back. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json +import logging + +from screen_locker._constants import SICK_DAY_STATE_FILE + +_logger = logging.getLogger(__name__) + + +class SickDayStateMixin: + """Persists and restores the pre-sick-day shutdown configuration.""" + + def _sick_mode_used_today(self) -> bool: + """Check if sick mode was already used today.""" + if not SICK_DAY_STATE_FILE.exists(): + return False + + try: + with SICK_DAY_STATE_FILE.open() as f: + state = json.load(f) + today = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d") + return state.get("date") == today + except (OSError, json.JSONDecodeError) as exc: + _logger.warning( + "Could not read sick-day state from %s: %s — treating sick mode " + "as UNUSED today, which may allow a second sick day", + SICK_DAY_STATE_FILE, + exc, + ) + return False + + def _save_sick_day_state( + self, + date: str, + orig_mon_wed: int, + orig_thu_sun: int, + ) -> bool: + """Save sick day state with original config values. + + Returns True if saved successfully, False otherwise. + """ + state = { + "date": date, + "original_mon_wed_hour": orig_mon_wed, + "original_thu_sun_hour": orig_thu_sun, + } + try: + with SICK_DAY_STATE_FILE.open("w") as f: + json.dump(state, f, indent=2) + except OSError as e: + _logger.warning("Failed to save sick day state: %s", e) + return False + + _logger.info("Saved sick day state for %s", date) + return True + + def _load_sick_day_state(self) -> tuple[str, int, int] | None: + """Load sick day state file. + + Returns (date, orig_mon_wed_hour, orig_thu_sun_hour) or None. + """ + with SICK_DAY_STATE_FILE.open() as f: + state = json.load(f) + date = state.get("date") + orig_mw = state.get("original_mon_wed_hour") + orig_ts = state.get("original_thu_sun_hour") + if date is None or orig_mw is None or orig_ts is None: + return None + return (str(date), int(orig_mw), int(orig_ts)) + + def _write_restored_config( + self, + orig_mw: int, + orig_ts: int, + state_date: str, + ) -> None: + """Write restored config values and clean up state file.""" + config_values = self._read_shutdown_config() + if config_values: + _, _, morning_end = config_values + _logger.info( + "Restoring original shutdown config from %s", + state_date, + ) + self._write_shutdown_config( + orig_mw, + orig_ts, + morning_end, + restore=True, + ) + SICK_DAY_STATE_FILE.unlink() + _logger.info("Removed stale sick day state from %s", state_date) + + def _restore_original_config_if_needed(self) -> None: + """Restore original config if sick day state is from a previous day.""" + if not SICK_DAY_STATE_FILE.exists(): + return + try: + loaded = self._load_sick_day_state() + if loaded is None: + return + state_date, orig_mw, orig_ts = loaded + today = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d") + if state_date != today: + self._write_restored_config(orig_mw, orig_ts, state_date) + except (OSError, json.JSONDecodeError) as e: + _logger.warning("Error checking sick day state: %s", e) diff --git a/screen_locker/_status_data.py b/screen_locker/_status_data.py index 777b0b2..e78d72d 100644 --- a/screen_locker/_status_data.py +++ b/screen_locker/_status_data.py @@ -10,13 +10,37 @@ status view are built on. from __future__ import annotations -from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import TYPE_CHECKING from screen_locker import _manual_workout -from screen_locker._compliance_state import LockExplanation, explain_lock_decision +from screen_locker._compliance_state import explain_lock_decision +from screen_locker._status_projection import ( + _shutdown_projection, + format_summary_line, +) +from screen_locker._status_types import ( + DayStatus, + ManualWorkoutBudgetStatus, + ShutdownProjection, + ShutdownProjectionDay, + SickBudgetStatus, + StatusSnapshot, + WeeklySummary, +) + +__all__ = [ + "DayStatus", + "ManualWorkoutBudgetStatus", + "ShutdownProjection", + "ShutdownProjectionDay", + "SickBudgetStatus", + "StatusSnapshot", + "WeeklySummary", + "format_summary_line", + "gather_status", +] from screen_locker._constants import ( EARLY_BIRD_PENDING_FILE, EXTRA_BENEFITS_FILE, @@ -32,12 +56,9 @@ from screen_locker._constants import ( from screen_locker._extra_benefits import ( current_streak, has_extended_early_bird, - preview_bonus_if_week_ended_now, weekly_shutdown_bonus_hours, ) from screen_locker._log_io import load_workout_log -from screen_locker._shutdown import read_shutdown_config -from screen_locker._shutdown_base import get_base_hours from screen_locker._sick_tracker import ( count_in_window, is_budget_exhausted, @@ -49,7 +70,6 @@ from screen_locker._weekly_check import ( MANUAL_WORKOUT_TYPE, VERIFIED_WORKOUT_TYPES, WEEKLY_WORKOUT_MINIMUM, - count_weekly_workouts, is_relaxed_day, ) @@ -57,97 +77,6 @@ if TYPE_CHECKING: from screen_locker._sick_tracker import SickHistory _DEFAULT_LOG_FILE = Path(__file__).resolve().parent / "workout_log.json" -_RELAXED_DAY_EXPLANATION = ( - "Shutdown time and the screen lock are two separate systems: the lock " - "decides whether you must log a workout right now; shutdown time only " - "controls how late the PC stays on tonight." -) -_WEEKDAY_LABELS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") -_MON_WED_WEEKDAYS = frozenset({0, 1, 2}) - - -@dataclass(frozen=True) -class DayStatus: - """One day's workout outcome (a day may hold several workouts).""" - - date: str - label: str - entry_types: tuple[str, ...] - source: str - counted: bool - day_count: int - is_sick_day: bool - - -@dataclass(frozen=True) -class WeeklySummary: - """This ISO week's workout compliance so far.""" - - days: tuple[DayStatus, ...] - counted_count: int - minimum: int - remaining: int - extra: int - - -@dataclass(frozen=True) -class SickBudgetStatus: - """Rolling sick-day budget usage.""" - - used_7d: int - budget_7d: int - used_30d: int - budget_30d: int - used_90d: int - budget_90d: int - debt: int - exhausted: bool - - -@dataclass(frozen=True) -class ManualWorkoutBudgetStatus: - """Rolling manual-workout budget usage.""" - - used_7d: int - budget_7d: int - used_30d: int - budget_30d: int - exhausted: bool - - -@dataclass(frozen=True) -class ShutdownProjectionDay: - """One labeled row in a shutdown-time projection.""" - - label: str - hour: int - speculative: bool - - -@dataclass(frozen=True) -class ShutdownProjection: - """Tonight's live config plus deterministic/speculative week projections.""" - - tonight: tuple[int, int, int] | None - rest_of_week: tuple[ShutdownProjectionDay, ...] - next_week_preview: tuple[ShutdownProjectionDay, ...] - explanation: str - - -@dataclass(frozen=True) -class StatusSnapshot: - """Everything the status view needs, gathered in one read-only pass.""" - - today: DayStatus - week: WeeklySummary - bonus_hours_this_week: int - streak: int - early_bird_extended: bool - shutdown: ShutdownProjection - lock_explanation: LockExplanation - sick_budget: SickBudgetStatus - manual_workout_budget: ManualWorkoutBudgetStatus - generated_at: str def _day_status(day_date: date, entries: list[dict], sick_days: set[str]) -> DayStatus: @@ -237,51 +166,6 @@ def _manual_workout_budget_status( ) -def _week_rows( - mon_wed_hour: int, thu_sun_hour: int, *, speculative: bool -) -> tuple[ShutdownProjectionDay, ...]: - """Build 7 labeled rows, one per weekday, from a Mon-Wed/Thu-Sun band pair.""" - return tuple( - ShutdownProjectionDay( - label=label, - hour=mon_wed_hour if weekday in _MON_WED_WEEKDAYS else thu_sun_hour, - speculative=speculative, - ) - for weekday, label in enumerate(_WEEKDAY_LABELS) - ) - - -def _shutdown_projection( - *, - shutdown_base_file: Path, - shutdown_config_file: Path, - extra_benefits_file: Path, - log_file: Path, - today_local: datetime, -) -> ShutdownProjection: - """Build the shutdown-time projection: tonight, rest of week, next week.""" - tonight = read_shutdown_config(shutdown_config_file) - base_mw, base_ts = get_base_hours(shutdown_base_file) - bonus = weekly_shutdown_bonus_hours(extra_benefits_file, today=today_local) - rest_of_week = _week_rows(base_mw + bonus, base_ts + bonus, speculative=False) - - this_week_count = count_weekly_workouts(log_file, today=today_local) - streak = current_streak(extra_benefits_file) - _would_be_streak, would_be_bonus = preview_bonus_if_week_ended_now( - this_week_count, streak - ) - next_week_preview = _week_rows( - base_mw + would_be_bonus, base_ts + would_be_bonus, speculative=True - ) - - return ShutdownProjection( - tonight=tonight, - rest_of_week=rest_of_week, - next_week_preview=next_week_preview, - explanation=_RELAXED_DAY_EXPLANATION, - ) - - def gather_status( *, log_file: Path = _DEFAULT_LOG_FILE, @@ -340,31 +224,3 @@ def gather_status( manual_workout_budget=_manual_workout_budget_status(log_file, today=today_str), generated_at=instant.isoformat(), ) - - -def _tonight_hour(snapshot: StatusSnapshot) -> int | None: - """Pick the shutdown hour band (Mon-Wed vs Thu-Sun) matching today. - - Uses ``snapshot.generated_at`` (not a fresh ``datetime.now()``) so the - summary line is a pure function of the snapshot it was handed. - """ - if snapshot.shutdown.tonight is None: - return None - mon_wed_hour, thu_sun_hour, _morning = snapshot.shutdown.tonight - generated_weekday = ( - datetime.fromisoformat(snapshot.generated_at).astimezone().weekday() - ) - return mon_wed_hour if generated_weekday in _MON_WED_WEEKDAYS else thu_sun_hour - - -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 "…" - tonight_hour = _tonight_hour(snapshot) - tonight_str = f"{tonight_hour:02d}:00" if tonight_hour is not None else "?" - sick = snapshot.sick_budget - return ( - f"{mark} {week.counted_count}/{week.minimum} workouts · " - f"{tonight_str} tonight · sick {sick.used_7d}/{sick.budget_7d}" - ) diff --git a/screen_locker/_status_projection.py b/screen_locker/_status_projection.py new file mode 100644 index 0000000..0593fa1 --- /dev/null +++ b/screen_locker/_status_projection.py @@ -0,0 +1,110 @@ +"""Shutdown projection and the one-line i3blocks summary. + +Split out of :mod:`screen_locker._status_data` to keep every file under the +250-line cap. ``format_summary_line`` is re-exported from there, so +``status_view`` and the i3blocks CLI are unaffected. + +Projects what time the PC will shut down on each remaining day of the week, +and renders the compact status line. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from screen_locker._extra_benefits import ( + current_streak, + preview_bonus_if_week_ended_now, + weekly_shutdown_bonus_hours, +) +from screen_locker._shutdown import read_shutdown_config +from screen_locker._shutdown_base import get_base_hours +from screen_locker._status_types import ShutdownProjection, ShutdownProjectionDay +from screen_locker._weekly_check import count_weekly_workouts + +if TYPE_CHECKING: + from pathlib import Path + + from screen_locker._status_types import StatusSnapshot + +_RELAXED_DAY_EXPLANATION = ( + "Shutdown time and the screen lock are two separate systems: the lock " + "decides whether you must log a workout right now; shutdown time only " + "controls how late the PC stays on tonight." +) +_WEEKDAY_LABELS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") +_MON_WED_WEEKDAYS = frozenset({0, 1, 2}) + + +def _week_rows( + mon_wed_hour: int, thu_sun_hour: int, *, speculative: bool +) -> tuple[ShutdownProjectionDay, ...]: + """Build 7 labeled rows, one per weekday, from a Mon-Wed/Thu-Sun band pair.""" + return tuple( + ShutdownProjectionDay( + label=label, + hour=mon_wed_hour if weekday in _MON_WED_WEEKDAYS else thu_sun_hour, + speculative=speculative, + ) + for weekday, label in enumerate(_WEEKDAY_LABELS) + ) + + +def _shutdown_projection( + *, + shutdown_base_file: Path, + shutdown_config_file: Path, + extra_benefits_file: Path, + log_file: Path, + today_local: datetime, +) -> ShutdownProjection: + """Build the shutdown-time projection: tonight, rest of week, next week.""" + tonight = read_shutdown_config(shutdown_config_file) + base_mw, base_ts = get_base_hours(shutdown_base_file) + bonus = weekly_shutdown_bonus_hours(extra_benefits_file, today=today_local) + rest_of_week = _week_rows(base_mw + bonus, base_ts + bonus, speculative=False) + + this_week_count = count_weekly_workouts(log_file, today=today_local) + streak = current_streak(extra_benefits_file) + _would_be_streak, would_be_bonus = preview_bonus_if_week_ended_now( + this_week_count, streak + ) + next_week_preview = _week_rows( + base_mw + would_be_bonus, base_ts + would_be_bonus, speculative=True + ) + + return ShutdownProjection( + tonight=tonight, + rest_of_week=rest_of_week, + next_week_preview=next_week_preview, + explanation=_RELAXED_DAY_EXPLANATION, + ) + + +def _tonight_hour(snapshot: StatusSnapshot) -> int | None: + """Pick the shutdown hour band (Mon-Wed vs Thu-Sun) matching today. + + Uses ``snapshot.generated_at`` (not a fresh ``datetime.now()``) so the + summary line is a pure function of the snapshot it was handed. + """ + if snapshot.shutdown.tonight is None: + return None + mon_wed_hour, thu_sun_hour, _morning = snapshot.shutdown.tonight + generated_weekday = ( + datetime.fromisoformat(snapshot.generated_at).astimezone().weekday() + ) + return mon_wed_hour if generated_weekday in _MON_WED_WEEKDAYS else thu_sun_hour + + +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 "…" + tonight_hour = _tonight_hour(snapshot) + tonight_str = f"{tonight_hour:02d}:00" if tonight_hour is not None else "?" + sick = snapshot.sick_budget + return ( + f"{mark} {week.counted_count}/{week.minimum} workouts · " + f"{tonight_str} tonight · sick {sick.used_7d}/{sick.budget_7d}" + ) diff --git a/screen_locker/_status_types.py b/screen_locker/_status_types.py new file mode 100644 index 0000000..45718bd --- /dev/null +++ b/screen_locker/_status_types.py @@ -0,0 +1,101 @@ +"""The frozen dataclasses describing a status snapshot. + +Split out of :mod:`screen_locker._status_data` to keep every file under the +250-line cap. Re-exported from there, so ``status_view`` and every other +importer is unaffected. + +These are plain value objects: building them from disk is ``_status_data``'s +job, rendering them is ``_status_sections``'. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from screen_locker._compliance_state import LockExplanation + + +@dataclass(frozen=True) +class DayStatus: + """One day's workout outcome (a day may hold several workouts).""" + + date: str + label: str + entry_types: tuple[str, ...] + source: str + counted: bool + day_count: int + is_sick_day: bool + + +@dataclass(frozen=True) +class WeeklySummary: + """This ISO week's workout compliance so far.""" + + days: tuple[DayStatus, ...] + counted_count: int + minimum: int + remaining: int + extra: int + + +@dataclass(frozen=True) +class SickBudgetStatus: + """Rolling sick-day budget usage.""" + + used_7d: int + budget_7d: int + used_30d: int + budget_30d: int + used_90d: int + budget_90d: int + debt: int + exhausted: bool + + +@dataclass(frozen=True) +class ManualWorkoutBudgetStatus: + """Rolling manual-workout budget usage.""" + + used_7d: int + budget_7d: int + used_30d: int + budget_30d: int + exhausted: bool + + +@dataclass(frozen=True) +class ShutdownProjectionDay: + """One labeled row in a shutdown-time projection.""" + + label: str + hour: int + speculative: bool + + +@dataclass(frozen=True) +class ShutdownProjection: + """Tonight's live config plus deterministic/speculative week projections.""" + + tonight: tuple[int, int, int] | None + rest_of_week: tuple[ShutdownProjectionDay, ...] + next_week_preview: tuple[ShutdownProjectionDay, ...] + explanation: str + + +@dataclass(frozen=True) +class StatusSnapshot: + """Everything the status view needs, gathered in one read-only pass.""" + + today: DayStatus + week: WeeklySummary + bonus_hours_this_week: int + streak: int + early_bird_extended: bool + shutdown: ShutdownProjection + lock_explanation: LockExplanation + sick_budget: SickBudgetStatus + manual_workout_budget: ManualWorkoutBudgetStatus + generated_at: str diff --git a/screen_locker/_wake_alarm.py b/screen_locker/_wake_alarm.py new file mode 100644 index 0000000..237def7 --- /dev/null +++ b/screen_locker/_wake_alarm.py @@ -0,0 +1,93 @@ +"""rtcwake wake-alarm scheduling, split out of the shutdown mixin. + +Split out of :mod:`screen_locker._shutdown` to keep every file under the +250-line cap. Composed back into ``ShutdownMixin`` there, so callers see no +change. +""" + +from __future__ import annotations + +import calendar +from datetime import datetime, timedelta, timezone +import logging +import subprocess + +from screen_locker._constants import ALARM_DAYS, RTCWAKE_BIN, WAKE_AFTER_HOURS + +_logger = logging.getLogger(__name__) + + +class WakeAlarmMixin: + """Schedules an RTC wake alarm for the next alarm day.""" + + # ------------------------------------------------------------------ + # rtcwake integration for weekend wake alarm + # ------------------------------------------------------------------ + + @staticmethod + def _is_tomorrow_alarm_day() -> bool: + """Check if tomorrow is an alarm day.""" + tomorrow = datetime.now(tz=timezone.utc) + timedelta(days=1) + return tomorrow.weekday() in ALARM_DAYS + + @staticmethod + def _compute_wake_timestamp() -> int: + """Compute the UTC epoch timestamp for the next wake alarm. + + Returns: + Epoch seconds WAKE_AFTER_HOURS from now. + """ + wake_time = datetime.now(tz=timezone.utc) + timedelta( + hours=WAKE_AFTER_HOURS, + ) + return calendar.timegm(wake_time.utctimetuple()) + + @staticmethod + def _schedule_rtcwake() -> bool: + """Set rtcwake to power on the PC after WAKE_AFTER_HOURS. + + Uses ``rtcwake -m disk`` to hibernate immediately while programming + the RTC to restore power at wake_epoch. Hibernate is completely + silent and dark (state written to swap file), making it suitable + when the PC is in a bedroom. + + Returns: + True if rtcwake was set successfully, False otherwise. + """ + wake_epoch = WakeAlarmMixin._compute_wake_timestamp() + cmd = [ + "/usr/bin/sudo", + RTCWAKE_BIN, + "-m", + "disk", + "-t", + str(wake_epoch), + ] + try: + subprocess.run( + cmd, + check=True, + capture_output=True, + text=True, + ) + except subprocess.SubprocessError as exc: + _logger.warning("Failed to set rtcwake: %s", exc) + return False + _logger.info( + "rtcwake set: PC will wake at epoch %d", + wake_epoch, + ) + return True + + def schedule_wake_if_needed(self) -> bool: + """Schedule rtcwake if tomorrow is an alarm day. + + Call this at shutdown time. + + Returns: + True if wake was scheduled, False if not needed or failed. + """ + if not self._is_tomorrow_alarm_day(): + _logger.info("Tomorrow is not an alarm day — skipping rtcwake") + return False + return self._schedule_rtcwake() diff --git a/screen_locker/tests/conftest.py b/screen_locker/tests/conftest.py index 21a858b..99f6241 100644 --- a/screen_locker/tests/conftest.py +++ b/screen_locker/tests/conftest.py @@ -138,7 +138,10 @@ def _block_real_network() -> Iterator[None]: reach the real network. Tests needing a real probe patch it locally. """ targets = ( - "screen_locker._phone_verification.socket.create_connection", + # The subnet probes moved out of _phone_verification when it was split; + # both new homes must be blocked or a test reaches the real network. + "screen_locker._adb_transport.socket.create_connection", + "screen_locker._http_workout_fetch.socket.create_connection", "screen_locker._temperature.urllib.request.urlopen", ) with ExitStack() as stack: diff --git a/screen_locker/tests/test_adb_and_phone.py b/screen_locker/tests/test_adb_and_phone.py index 124a8f1..60279de 100644 --- a/screen_locker/tests/test_adb_and_phone.py +++ b/screen_locker/tests/test_adb_and_phone.py @@ -26,7 +26,7 @@ class TestRunAdb: locker = create_locker(mock_tk, tmp_path) mock_result = MagicMock(returncode=0, stdout="ok\n") with patch( - "screen_locker._phone_verification.subprocess.run", + "screen_locker._adb_transport.subprocess.run", return_value=mock_result, ) as mock_run: success, output = locker._run_adb(["devices"]) @@ -45,7 +45,7 @@ class TestRunAdb: locker = create_locker(mock_tk, tmp_path) mock_result = MagicMock(returncode=1, stdout="") with patch( - "screen_locker._phone_verification.subprocess.run", + "screen_locker._adb_transport.subprocess.run", return_value=mock_result, ): success, _output = locker._run_adb(["devices"]) @@ -61,7 +61,7 @@ class TestRunAdb: """Test ADB binary not found.""" locker = create_locker(mock_tk, tmp_path) with patch( - "screen_locker._phone_verification.subprocess.run", + "screen_locker._adb_transport.subprocess.run", side_effect=FileNotFoundError("adb not found"), ): success, output = locker._run_adb(["devices"]) @@ -78,7 +78,7 @@ class TestRunAdb: """Test ADB OSError.""" locker = create_locker(mock_tk, tmp_path) with patch( - "screen_locker._phone_verification.subprocess.run", + "screen_locker._adb_transport.subprocess.run", side_effect=OSError("permission denied"), ): success, output = locker._run_adb(["devices"]) @@ -95,7 +95,7 @@ class TestRunAdb: """Test ADB command timeout.""" locker = create_locker(mock_tk, tmp_path) with patch( - "screen_locker._phone_verification.subprocess.run", + "screen_locker._adb_transport.subprocess.run", side_effect=subprocess.TimeoutExpired("adb", 15), ): success, output = locker._run_adb(["devices"]) diff --git a/screen_locker/tests/test_phone_verification_part2.py b/screen_locker/tests/test_phone_verification_part2.py index 08d67d5..b0dccbe 100644 --- a/screen_locker/tests/test_phone_verification_part2.py +++ b/screen_locker/tests/test_phone_verification_part2.py @@ -144,7 +144,7 @@ class TestGetLocalSubnetPrefix: mock_sock.__enter__ = MagicMock(return_value=mock_sock) mock_sock.__exit__ = MagicMock(return_value=False) with patch( - "screen_locker._phone_verification.socket.socket", + "screen_locker._adb_transport.socket.socket", return_value=mock_sock, ): result = locker._get_local_subnet_prefix() @@ -159,7 +159,7 @@ class TestGetLocalSubnetPrefix: """Test returns None when socket raises OSError.""" locker = create_locker(mock_tk, tmp_path) with patch( - "screen_locker._phone_verification.socket.socket", + "screen_locker._adb_transport.socket.socket", side_effect=OSError("no network"), ): result = locker._get_local_subnet_prefix() diff --git a/screen_locker/tests/test_phone_verification_part4.py b/screen_locker/tests/test_phone_verification_part4.py index 876c703..0a2528f 100644 --- a/screen_locker/tests/test_phone_verification_part4.py +++ b/screen_locker/tests/test_phone_verification_part4.py @@ -128,7 +128,7 @@ class TestScanForHttpServer: with ( patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._http_workout_fetch.socket.create_connection", return_value=_mock_cm(MagicMock()), ), ): @@ -145,7 +145,7 @@ class TestScanForHttpServer: with ( patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._http_workout_fetch.socket.create_connection", side_effect=OSError("refused"), ), ): diff --git a/screen_locker/tests/test_phone_verification_part6.py b/screen_locker/tests/test_phone_verification_part6.py index 0b547cd..18835a6 100644 --- a/screen_locker/tests/test_phone_verification_part6.py +++ b/screen_locker/tests/test_phone_verification_part6.py @@ -39,7 +39,7 @@ class TestTryWirelessReconnect: patch.object(locker, "_try_adb_connect", return_value=True), patch.object(locker, "_has_adb_device", return_value=True), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._adb_transport.socket.create_connection", ) as mock_conn, ): mock_sock = MagicMock() @@ -60,7 +60,7 @@ class TestTryWirelessReconnect: with ( patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._adb_transport.socket.create_connection", side_effect=OSError("refused"), ), ): @@ -80,7 +80,7 @@ class TestTryWirelessReconnect: patch.object(locker, "_try_adb_connect", return_value=True), patch.object(locker, "_has_adb_device", return_value=False), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._adb_transport.socket.create_connection", ) as mock_conn, ): mock_sock = MagicMock() @@ -102,7 +102,7 @@ class TestTryWirelessReconnect: patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"), patch.object(locker, "_try_adb_connect", return_value=False), patch( - "screen_locker._phone_verification.socket.create_connection", + "screen_locker._adb_transport.socket.create_connection", ) as mock_conn, ): mock_sock = MagicMock() diff --git a/screen_locker/tests/test_phone_verification_part7.py b/screen_locker/tests/test_phone_verification_part7.py index e16d58c..7dd3d2e 100644 --- a/screen_locker/tests/test_phone_verification_part7.py +++ b/screen_locker/tests/test_phone_verification_part7.py @@ -45,7 +45,7 @@ class TestFetchHttpWorkout: return_value="http://192.168.1.5:8765/workout", ), patch( - "screen_locker._phone_verification._HTTPConnection", + "screen_locker._http_workout_fetch._HTTPConnection", return_value=conn, ), ): @@ -68,7 +68,7 @@ class TestFetchHttpWorkout: return_value="http://192.168.1.5:8765/workout", ), patch( - "screen_locker._phone_verification._HTTPConnection", + "screen_locker._http_workout_fetch._HTTPConnection", return_value=conn, ), ): @@ -86,7 +86,7 @@ class TestFetchHttpWorkout: return_value="http://192.168.1.5:8765/workout", ), patch( - "screen_locker._phone_verification._HTTPConnection", + "screen_locker._http_workout_fetch._HTTPConnection", side_effect=OSError("unreachable"), ), ): diff --git a/screen_locker/tests/test_shutdown_part3.py b/screen_locker/tests/test_shutdown_part3.py index 1a39798..1d8365e 100644 --- a/screen_locker/tests/test_shutdown_part3.py +++ b/screen_locker/tests/test_shutdown_part3.py @@ -27,7 +27,7 @@ class TestRestoreOriginalConfigIfNeeded: mock_file = MagicMock() mock_file.exists.return_value = False with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", mock_file, ): locker._restore_original_config_if_needed() @@ -52,7 +52,7 @@ class TestRestoreOriginalConfigIfNeeded: ) with ( patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ), patch.object(locker, "_write_restored_config") as mock_restore, @@ -83,7 +83,7 @@ class TestRestoreOriginalConfigIfNeeded: ) with ( patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ), patch.object(locker, "_write_restored_config") as mock_restore, @@ -103,7 +103,7 @@ class TestRestoreOriginalConfigIfNeeded: state_file.write_text(json.dumps({"date": "2020-01-01"})) with ( patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ), patch.object(locker, "_write_restored_config") as mock_restore, @@ -123,7 +123,7 @@ class TestRestoreOriginalConfigIfNeeded: mock_file.exists.return_value = True mock_file.open.side_effect = OSError("fail") with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", mock_file, ): locker._restore_original_config_if_needed() @@ -139,7 +139,7 @@ class TestRestoreOriginalConfigIfNeeded: state_file = tmp_path / "state.json" state_file.write_text("not valid json{{{") with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): locker._restore_original_config_if_needed() diff --git a/screen_locker/tests/test_shutdown_part4.py b/screen_locker/tests/test_shutdown_part4.py index 71735cd..07515fe 100644 --- a/screen_locker/tests/test_shutdown_part4.py +++ b/screen_locker/tests/test_shutdown_part4.py @@ -25,7 +25,7 @@ class TestSaveSickDayState: locker = create_locker(mock_tk, tmp_path) state_file = tmp_path / "state.json" with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): result = locker._save_sick_day_state("2026-03-21", 21, 20) @@ -46,7 +46,7 @@ class TestSaveSickDayState: mock_path = MagicMock() mock_path.open.side_effect = OSError("permission denied") with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", mock_path, ): result = locker._save_sick_day_state("2026-03-21", 21, 20) @@ -75,7 +75,7 @@ class TestLoadSickDayState: ) ) with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): result = locker._load_sick_day_state() @@ -92,7 +92,7 @@ class TestLoadSickDayState: state_file = tmp_path / "state.json" state_file.write_text(json.dumps({"date": "2026-03-20"})) with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): result = locker._load_sick_day_state() @@ -118,7 +118,7 @@ class TestWriteRestoredConfig: locker, "_write_shutdown_config", return_value=True ) as mock_write, patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ), ): @@ -139,7 +139,7 @@ class TestWriteRestoredConfig: with ( patch.object(locker, "_read_shutdown_config", return_value=None), patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ), ): diff --git a/screen_locker/tests/test_shutdown_part5.py b/screen_locker/tests/test_shutdown_part5.py index 36c1982..aaa62df 100644 --- a/screen_locker/tests/test_shutdown_part5.py +++ b/screen_locker/tests/test_shutdown_part5.py @@ -26,7 +26,7 @@ class TestSickModeUsedToday: mock_file = MagicMock() mock_file.exists.return_value = False with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", mock_file, ): assert locker._sick_mode_used_today() is False @@ -41,7 +41,7 @@ class TestSickModeUsedToday: locker = create_locker(mock_tk, tmp_path) state_file = tmp_path / "state.json" with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): from datetime import datetime, timezone @@ -60,7 +60,7 @@ class TestSickModeUsedToday: locker = create_locker(mock_tk, tmp_path) state_file = tmp_path / "state.json" with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): state_file.write_text(json.dumps({"date": "2020-01-01"})) @@ -76,7 +76,7 @@ class TestSickModeUsedToday: locker = create_locker(mock_tk, tmp_path) state_file = tmp_path / "state.json" with patch( - "screen_locker._shutdown.SICK_DAY_STATE_FILE", + "screen_locker._shutdown_sick_state.SICK_DAY_STATE_FILE", state_file, ): state_file.write_text("not json{{{") diff --git a/screen_locker/tests/test_wake_shutdown.py b/screen_locker/tests/test_wake_shutdown.py index 8a10294..07fca1e 100644 --- a/screen_locker/tests/test_wake_shutdown.py +++ b/screen_locker/tests/test_wake_shutdown.py @@ -26,7 +26,7 @@ class TestIsTomorrowAlarmDay: # Sunday 2026-04-12 → tomorrow Monday with patch( - "screen_locker._shutdown.datetime", + "screen_locker._wake_alarm.datetime", ) as mock_dt: mock_dt.now.return_value = datetime(2026, 4, 12, 23, 0, tzinfo=timezone.utc) mock_dt.side_effect = datetime @@ -34,7 +34,7 @@ class TestIsTomorrowAlarmDay: # Ensure timedelta works with patch( - "screen_locker._shutdown.timedelta", + "screen_locker._wake_alarm.timedelta", timedelta, ): assert locker._is_tomorrow_alarm_day() is True @@ -52,10 +52,10 @@ class TestIsTomorrowAlarmDay: # Monday 2026-04-13 → tomorrow Tuesday (weekday=1) with ( patch( - "screen_locker._shutdown.datetime", + "screen_locker._wake_alarm.datetime", ) as mock_dt, patch( - "screen_locker._shutdown.timedelta", + "screen_locker._wake_alarm.timedelta", timedelta, ), ): @@ -76,10 +76,10 @@ class TestIsTomorrowAlarmDay: # Thursday 2026-04-16 → tomorrow Friday (weekday=4) with ( patch( - "screen_locker._shutdown.datetime", + "screen_locker._wake_alarm.datetime", ) as mock_dt, patch( - "screen_locker._shutdown.timedelta", + "screen_locker._wake_alarm.timedelta", timedelta, ), ): @@ -100,7 +100,7 @@ class TestScheduleRtcwake: """Successful rtcwake call returns True.""" locker = create_locker(mock_tk, tmp_path) with patch( - "screen_locker._shutdown.subprocess.run", + "screen_locker._wake_alarm.subprocess.run", ) as mock_run: mock_run.return_value = MagicMock(returncode=0) assert locker._schedule_rtcwake() is True @@ -119,7 +119,7 @@ class TestScheduleRtcwake: import subprocess with patch( - "screen_locker._shutdown.subprocess.run", + "screen_locker._wake_alarm.subprocess.run", side_effect=subprocess.SubprocessError("rtcwake failed"), ): assert locker._schedule_rtcwake() is False