mirror of
https://github.com/kuhyx/screen-locker.git
synced 2026-11-10 20:27:13 +01:00
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EveB813BQ8abAP68VSqZ5Y
This commit is contained in:
parent
43a8b1d0f6
commit
bdc405c8c9
137
screen_locker/_adb_transport.py
Normal file
137
screen_locker/_adb_transport.py
Normal file
@ -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 ──────────────────────────────────────────────────────
|
||||
94
screen_locker/_http_workout_fetch.py
Normal file
94
screen_locker/_http_workout_fetch.py
Normal file
@ -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://<local-ip>:8765/workout — no user input involved.
|
||||
"""
|
||||
url = self._scan_for_http_server()
|
||||
if url is None:
|
||||
return None
|
||||
# url is always "http://<ip>:<port>/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 ─────────────────────────────────────────
|
||||
@ -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://<local-ip>:8765/workout — no user input involved.
|
||||
"""
|
||||
url = self._scan_for_http_server()
|
||||
if url is None:
|
||||
return None
|
||||
# url is always "http://<ip>:<port>/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.
|
||||
|
||||
|
||||
@ -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()
|
||||
|
||||
119
screen_locker/_shutdown_sick_state.py
Normal file
119
screen_locker/_shutdown_sick_state.py
Normal file
@ -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)
|
||||
@ -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}"
|
||||
)
|
||||
|
||||
110
screen_locker/_status_projection.py
Normal file
110
screen_locker/_status_projection.py
Normal file
@ -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}"
|
||||
)
|
||||
101
screen_locker/_status_types.py
Normal file
101
screen_locker/_status_types.py
Normal file
@ -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
|
||||
93
screen_locker/_wake_alarm.py
Normal file
93
screen_locker/_wake_alarm.py
Normal file
@ -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()
|
||||
@ -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:
|
||||
|
||||
@ -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"])
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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"),
|
||||
),
|
||||
):
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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"),
|
||||
),
|
||||
):
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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,
|
||||
),
|
||||
):
|
||||
|
||||
@ -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{{{")
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user