From bdc405c8c997372c9ce6a8009d079b64481ed84f Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 15 Aug 2026 20:11:28 +0200 Subject: [PATCH] Split _shutdown, _phone_verification and _status_data Six new modules, all composed or re-exported so no caller changes: _shutdown.py (384) -> _shutdown_sick_state.py SickDayStateMixin _wake_alarm.py WakeAlarmMixin _phone_verification.py (380)-> _adb_transport.py AdbTransportMixin _http_workout_fetch.py HttpWorkoutFetchMixin _status_data.py (370) -> _status_types.py the 7 dataclasses _status_projection.py shutdown projection + format_summary_line Moving a name means moving where it is *read*, so 30-odd mock patch targets move with it. Most importantly conftest.py's autouse _block_real_network fixture: ruff correctly dropped the now-unused `socket` import from _phone_verification, which silently removed the guard's target and would have let subnet probes reach the real network. It now blocks socket in both modules that actually own the probes, and both targets are asserted to exist. 976 tests pass at 100% branch coverage; pylint 8.70 -> 8.71; `screen_lock --status` renders the real week unchanged. File-length gate: 41 -> 35. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01EveB813BQ8abAP68VSqZ5Y --- screen_locker/_adb_transport.py | 137 ++++++++++++ screen_locker/_http_workout_fetch.py | 94 +++++++++ screen_locker/_phone_verification.py | 182 +--------------- screen_locker/_shutdown.py | 180 +--------------- screen_locker/_shutdown_sick_state.py | 119 +++++++++++ screen_locker/_status_data.py | 196 +++--------------- screen_locker/_status_projection.py | 110 ++++++++++ screen_locker/_status_types.py | 101 +++++++++ screen_locker/_wake_alarm.py | 93 +++++++++ screen_locker/tests/conftest.py | 5 +- screen_locker/tests/test_adb_and_phone.py | 10 +- .../tests/test_phone_verification_part2.py | 4 +- .../tests/test_phone_verification_part4.py | 4 +- .../tests/test_phone_verification_part6.py | 8 +- .../tests/test_phone_verification_part7.py | 6 +- screen_locker/tests/test_shutdown_part3.py | 12 +- screen_locker/tests/test_shutdown_part4.py | 12 +- screen_locker/tests/test_shutdown_part5.py | 8 +- screen_locker/tests/test_wake_shutdown.py | 16 +- 19 files changed, 731 insertions(+), 566 deletions(-) create mode 100644 screen_locker/_adb_transport.py create mode 100644 screen_locker/_http_workout_fetch.py create mode 100644 screen_locker/_shutdown_sick_state.py create mode 100644 screen_locker/_status_projection.py create mode 100644 screen_locker/_status_types.py create mode 100644 screen_locker/_wake_alarm.py 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