mirror of
https://github.com/kuhyx/screen-locker.git
synced 2026-07-04 13:23:13 +02:00
Restore consistency with WORKOUT_APP_JSON_REMOTES rename in _constants.py
A prior commit pushed the STRONGLIFTS_DB_REMOTE -> WORKOUT_APP_JSON_REMOTES rename in _constants.py without its consumer, breaking CI with an ImportError. This commits the matching _phone_verification.py rewrite and its reorganized test suite to close that gap. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A7vbgtFfZmfxJtN5DdtJky
This commit is contained in:
parent
28c27d24e8
commit
d50bc49b92
@ -1,4 +1,4 @@
|
||||
"""Phone workout verification mixin using ADB and StrongLifts."""
|
||||
"""Phone workout verification: ADB pull first, HTTP scan as fallback."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@ -7,12 +7,12 @@ from concurrent.futures import ( # pylint: disable=no-name-in-module
|
||||
as_completed,
|
||||
)
|
||||
import contextlib
|
||||
from http import client as _http_client
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import socket
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
@ -20,10 +20,15 @@ import time
|
||||
from screen_locker._constants import (
|
||||
ADB_TIMEOUT,
|
||||
MIN_WORKOUT_DURATION_MINUTES,
|
||||
STRONGLIFTS_DB_REMOTE,
|
||||
WORKOUT_APP_JSON_REMOTES,
|
||||
WORKOUT_HTTP_PORT,
|
||||
)
|
||||
from screen_locker._time_check import check_clock_skew
|
||||
|
||||
_HTTPConnection = _http_client.HTTPConnection
|
||||
_HTTPException = _http_client.HTTPException
|
||||
_HTTP_OK = _http_client.OK
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -58,23 +63,14 @@ class PhoneVerificationMixin:
|
||||
return False, ""
|
||||
return not result.returncode, result.stdout
|
||||
|
||||
def _adb_shell(
|
||||
self,
|
||||
command: str,
|
||||
*,
|
||||
root: bool = False,
|
||||
) -> tuple[bool, str]:
|
||||
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.
|
||||
|
||||
Used to pin ADB commands to the wireless device when multiple devices
|
||||
(e.g. USB cable + wireless debugging) are simultaneously connected.
|
||||
"""
|
||||
"""Return the serial (ip:port) of the first connected wireless ADB device."""
|
||||
success, output = self._run_adb(["devices"])
|
||||
if not success:
|
||||
return None
|
||||
@ -137,219 +133,144 @@ class PhoneVerificationMixin:
|
||||
def _is_phone_connected(self) -> bool:
|
||||
"""Check if an Android device is connected via ADB.
|
||||
|
||||
If no device is visible, attempts wireless reconnection using the
|
||||
stored phone IP/port config. USB-connected devices are detected
|
||||
automatically by adb devices without any extra steps.
|
||||
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()
|
||||
|
||||
def _pull_stronglifts_db(self) -> Path | None:
|
||||
"""Pull StrongLifts database from phone to a local temp file.
|
||||
# ── ADB verification ──────────────────────────────────────────────────────
|
||||
|
||||
Returns:
|
||||
Path to the local copy, or None on failure.
|
||||
def _pull_workout_app_json(self) -> dict | None:
|
||||
"""Pull workout_result.json from the phone (ADB, no root needed).
|
||||
|
||||
The app writes to one of several candidate paths (see sync_service.dart),
|
||||
so we pull every candidate and prefer the one dated today. A stale file
|
||||
can linger at a fallback path from a day the primary write failed, so
|
||||
"first that parses" is not safe — we explicitly favour today's data and
|
||||
only fall back to a stale/older payload if no candidate is from today.
|
||||
"""
|
||||
tmp = Path(tempfile.gettempdir()) / "stronglifts_check.db"
|
||||
success, _ = self._adb_shell(
|
||||
f"cat '{STRONGLIFTS_DB_REMOTE}' > /sdcard/_sl_tmp.db",
|
||||
root=True,
|
||||
)
|
||||
if not success:
|
||||
return None
|
||||
ok, _ = self._run_adb(["pull", "/sdcard/_sl_tmp.db", str(tmp)])
|
||||
tmp = Path(tempfile.gettempdir()) / "workout_result.json"
|
||||
today = time.strftime("%Y-%m-%d")
|
||||
first_parsed: dict | None = None
|
||||
for remote in WORKOUT_APP_JSON_REMOTES:
|
||||
ok, _ = self._run_adb(["pull", remote, str(tmp)])
|
||||
if not ok:
|
||||
return None
|
||||
return tmp
|
||||
|
||||
def _count_today_workouts(self, db_path: Path) -> int:
|
||||
"""Count today's workouts in a local copy of StrongLifts DB.
|
||||
|
||||
Args:
|
||||
db_path: Path to the locally-pulled StrongLifts database.
|
||||
|
||||
Returns:
|
||||
Number of workouts started today (local time).
|
||||
"""
|
||||
try:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT COUNT(*) FROM workouts "
|
||||
"WHERE date(start / 1000, 'unixepoch', 'localtime') "
|
||||
"= date('now', 'localtime')",
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
return int(row[0]) if row else 0
|
||||
finally:
|
||||
conn.close()
|
||||
except (sqlite3.Error, ValueError, TypeError):
|
||||
_logger.warning("Failed to query StrongLifts database")
|
||||
return 0
|
||||
|
||||
def _get_today_workout_duration_minutes(self, db_path: Path) -> float:
|
||||
"""Get the total duration in minutes of today's workouts.
|
||||
|
||||
Args:
|
||||
db_path: Path to the locally-pulled StrongLifts database.
|
||||
|
||||
Returns:
|
||||
Total duration in minutes of all workouts started today.
|
||||
Returns 0.0 on any error or if no workouts found.
|
||||
"""
|
||||
try:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT SUM((finish - start) / 1000.0 / 60.0) "
|
||||
"FROM workouts "
|
||||
"WHERE date(start / 1000, 'unixepoch', 'localtime') "
|
||||
"= date('now', 'localtime') "
|
||||
"AND finish > start",
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
return float(row[0]) if row and row[0] is not None else 0.0
|
||||
finally:
|
||||
conn.close()
|
||||
except (sqlite3.Error, ValueError, TypeError):
|
||||
_logger.warning("Failed to query workout duration")
|
||||
return 0.0
|
||||
|
||||
def _get_today_exercise_count(self, db_path: Path) -> int:
|
||||
"""Count distinct exercises in today's workouts.
|
||||
|
||||
Parses the JSON ``exercises`` column in the ``workouts`` table.
|
||||
Each workout row stores its exercises as a JSON array, not in a
|
||||
separate relational table.
|
||||
|
||||
Args:
|
||||
db_path: Path to the locally-pulled StrongLifts database.
|
||||
|
||||
Returns:
|
||||
Number of distinct exercises across today's workouts.
|
||||
Returns 0 on any error.
|
||||
"""
|
||||
try:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT exercises FROM workouts "
|
||||
"WHERE date(start / 1000, 'unixepoch', 'localtime') "
|
||||
"= date('now', 'localtime')",
|
||||
)
|
||||
exercise_ids: set[str] = set()
|
||||
for (exercises_json,) in cursor:
|
||||
if not exercises_json:
|
||||
continue
|
||||
for ex in json.loads(exercises_json):
|
||||
ex_id = ex.get("id") or ex.get("name", "")
|
||||
if ex_id:
|
||||
exercise_ids.add(ex_id)
|
||||
return len(exercise_ids)
|
||||
finally:
|
||||
conn.close()
|
||||
except (sqlite3.Error, ValueError, TypeError, json.JSONDecodeError):
|
||||
_logger.warning("Failed to query exercise count")
|
||||
return 0
|
||||
|
||||
def _is_workout_finish_recent(self, db_path: Path) -> bool:
|
||||
"""Check if the latest workout's finish time is recent.
|
||||
|
||||
A fresh workout should have finished within the last 24 hours.
|
||||
This prevents using an old pre-prepared database dump while
|
||||
still accepting workouts done earlier the same day.
|
||||
|
||||
Args:
|
||||
db_path: Path to the locally-pulled StrongLifts database.
|
||||
|
||||
Returns:
|
||||
True if the latest finish time is within 24 hours of now.
|
||||
"""
|
||||
max_age_seconds = 24 * 3600 # accept same-day workouts
|
||||
try:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT MAX(finish) FROM workouts "
|
||||
"WHERE date(start / 1000, 'unixepoch', 'localtime') "
|
||||
"= date('now', 'localtime') "
|
||||
"AND finish > start",
|
||||
data = json.loads(tmp.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
continue
|
||||
if data.get("date") == today:
|
||||
return data
|
||||
if first_parsed is None:
|
||||
first_parsed = data
|
||||
return first_parsed
|
||||
|
||||
def _validate_json_data(self, data: dict) -> tuple[str, str]:
|
||||
"""Validate parsed workout JSON. Returns (status, message)."""
|
||||
today = time.strftime("%Y-%m-%d")
|
||||
if data.get("date") != today:
|
||||
return "stale", f"Workout JSON is from {data.get('date')}, not today"
|
||||
if not data.get("exercises"):
|
||||
return "no_exercises", "No exercises found in today's workout JSON"
|
||||
duration_min = data.get("duration_seconds", 0) / 60.0
|
||||
if duration_min < MIN_WORKOUT_DURATION_MINUTES:
|
||||
return (
|
||||
"too_short",
|
||||
f"Workout too short! {duration_min:.0f} min logged, "
|
||||
f"need at least {MIN_WORKOUT_DURATION_MINUTES} min.",
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if not row or row[0] is None:
|
||||
return False
|
||||
finish_epoch = int(row[0]) / 1000.0
|
||||
return (time.time() - finish_epoch) < max_age_seconds
|
||||
finally:
|
||||
conn.close()
|
||||
except (sqlite3.Error, ValueError, TypeError):
|
||||
_logger.warning("Failed to query workout finish time")
|
||||
return False
|
||||
flag = "all succeeded" if data.get("succeeded") else "partial"
|
||||
return "verified", f"Workout verified! ({duration_min:.0f} min, {flag})"
|
||||
|
||||
def _validate_workout_db(
|
||||
self,
|
||||
local_db: Path,
|
||||
) -> tuple[str, str] | None:
|
||||
"""Validate workout database has a recent, real workout.
|
||||
# ── HTTP fallback (no ADB / developer options required) ───────────────────
|
||||
|
||||
Returns:
|
||||
A (status, message) tuple if validation fails, or None if OK.
|
||||
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.
|
||||
"""
|
||||
count = self._count_today_workouts(local_db)
|
||||
if count <= 0:
|
||||
return "not_verified", "No workout found on phone today"
|
||||
if not self._is_workout_finish_recent(local_db):
|
||||
return (
|
||||
"stale",
|
||||
"Workout finish time is too old. Did you actually work out today?",
|
||||
)
|
||||
exercise_count = self._get_today_exercise_count(local_db)
|
||||
if exercise_count < 1:
|
||||
return (
|
||||
"no_exercises",
|
||||
"No exercises found in today's workout. "
|
||||
"Log actual exercises in StrongLifts!",
|
||||
)
|
||||
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):
|
||||
return None
|
||||
|
||||
# ── Main verification entry point ─────────────────────────────────────────
|
||||
|
||||
def _verify_phone_workout(self) -> tuple[str, str]:
|
||||
"""Verify workout was recorded in StrongLifts on the phone.
|
||||
"""Verify today's workout: ADB pull if available, HTTP scan as fallback.
|
||||
|
||||
Returns:
|
||||
Tuple of (status, message) where status is one of:
|
||||
- "verified": Workout confirmed and >= minimum duration.
|
||||
- "too_short": Workout found but shorter than minimum.
|
||||
- "not_verified": Phone connected but no workout found.
|
||||
- "no_phone": No phone connected via ADB.
|
||||
- "error": Could not access StrongLifts database.
|
||||
- "stale": Workout finish time is not recent.
|
||||
- "no_exercises": Workout has no logged exercises.
|
||||
- "clock_tampered": System clock skew exceeds threshold.
|
||||
Returns (status, message). Status values:
|
||||
verified / too_short / not_verified / no_phone /
|
||||
stale / no_exercises / clock_tampered.
|
||||
"""
|
||||
clock_ok, clock_msg = check_clock_skew()
|
||||
if not clock_ok:
|
||||
return "clock_tampered", clock_msg
|
||||
if not self._is_phone_connected():
|
||||
return "no_phone", "No phone connected via ADB"
|
||||
local_db = self._pull_stronglifts_db()
|
||||
if local_db is None:
|
||||
return "error", "StrongLifts database not found on phone"
|
||||
db_error = self._validate_workout_db(local_db)
|
||||
if db_error is not None:
|
||||
return db_error
|
||||
duration = self._get_today_workout_duration_minutes(local_db)
|
||||
if duration < MIN_WORKOUT_DURATION_MINUTES:
|
||||
|
||||
# Prefer ADB when a device is visible, but if the pull yields no usable
|
||||
# JSON, fall through to the HTTP/WiFi scan — the app's in-memory HTTP
|
||||
# server may still hold today's workout even when the file pull fails.
|
||||
adb_connected = self._is_phone_connected()
|
||||
if adb_connected:
|
||||
data = self._pull_workout_app_json()
|
||||
if data is not None:
|
||||
return self._validate_json_data(data)
|
||||
_logger.info("ADB pull found no workout JSON — trying HTTP scan...")
|
||||
else:
|
||||
_logger.info("No ADB device — trying HTTP scan on local network...")
|
||||
|
||||
data = self._fetch_http_workout()
|
||||
if data is not None:
|
||||
return self._validate_json_data(data)
|
||||
if adb_connected:
|
||||
return (
|
||||
"too_short",
|
||||
f"Workout too short! {duration:.0f} min logged, "
|
||||
f"need at least {MIN_WORKOUT_DURATION_MINUTES} min.",
|
||||
)
|
||||
exercise_count = self._get_today_exercise_count(local_db)
|
||||
return (
|
||||
"verified",
|
||||
f"Workout verified! ({self._count_today_workouts(local_db)}"
|
||||
f" session(s), {duration:.0f} min, "
|
||||
f"{exercise_count} exercise(s))",
|
||||
"not_verified",
|
||||
"Workout app JSON not found. Complete a workout in the app first.",
|
||||
)
|
||||
return "no_phone", "Phone not reachable via ADB or HTTP on local network"
|
||||
|
||||
@ -7,7 +7,6 @@ import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker.screen_lock import STRONGLIFTS_DB_REMOTE
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -255,109 +254,3 @@ class TestIsPhoneConnected:
|
||||
)
|
||||
|
||||
assert locker._is_phone_connected() is False
|
||||
|
||||
|
||||
class TestFindHealthConnectDb:
|
||||
"""Tests for _pull_stronglifts_db method."""
|
||||
|
||||
def test_db_pulled_successfully(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test StrongLifts DB pulled from device."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_adb_shell",
|
||||
MagicMock(
|
||||
return_value=(True, ""),
|
||||
),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_run_adb",
|
||||
MagicMock(
|
||||
return_value=(True, ""),
|
||||
),
|
||||
)
|
||||
|
||||
result = locker._pull_stronglifts_db()
|
||||
|
||||
assert result is not None
|
||||
locker._adb_shell.assert_called_once()
|
||||
locker._run_adb.assert_called_once()
|
||||
call_args = locker._run_adb.call_args[0][0]
|
||||
assert call_args[0] == "pull"
|
||||
|
||||
def test_db_cat_fails(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns None when cat command fails."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_adb_shell",
|
||||
MagicMock(
|
||||
return_value=(False, ""),
|
||||
),
|
||||
)
|
||||
|
||||
assert locker._pull_stronglifts_db() is None
|
||||
|
||||
def test_db_pull_fails(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns None when adb pull fails."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_adb_shell",
|
||||
MagicMock(
|
||||
return_value=(True, ""),
|
||||
),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_run_adb",
|
||||
MagicMock(
|
||||
return_value=(False, ""),
|
||||
),
|
||||
)
|
||||
|
||||
assert locker._pull_stronglifts_db() is None
|
||||
|
||||
def test_db_uses_correct_remote_path(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test uses the correct StrongLifts DB remote path."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_adb_shell",
|
||||
MagicMock(
|
||||
return_value=(True, ""),
|
||||
),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_run_adb",
|
||||
MagicMock(
|
||||
return_value=(True, ""),
|
||||
),
|
||||
)
|
||||
|
||||
locker._pull_stronglifts_db()
|
||||
|
||||
shell_cmd = locker._adb_shell.call_args[0][0]
|
||||
assert STRONGLIFTS_DB_REMOTE in shell_cmd
|
||||
|
||||
@ -1,394 +0,0 @@
|
||||
"""Tests for ADB commands, phone connection, and database operations."""
|
||||
# pylint: disable=protected-access,unused-argument
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
class TestGetTodayWorkoutDurationMinutes:
|
||||
"""Tests for _get_today_workout_duration_minutes method."""
|
||||
|
||||
def test_returns_duration_for_today_workout(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns correct duration for a 60-minute workout."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
duration_ms = 60 * 60 * 1000 # 60 minutes
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms + duration_ms),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
result = locker._get_today_workout_duration_minutes(db_file)
|
||||
assert result == pytest.approx(60.0, abs=1.0)
|
||||
|
||||
def test_returns_zero_for_no_workouts(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0.0 when no workouts today."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
yesterday_ms = int((time.time() - 200000) * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", yesterday_ms, yesterday_ms + 3600000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_workout_duration_minutes(db_file)
|
||||
|
||||
def test_sums_multiple_workouts(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test sums durations of multiple workouts today."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
# 30 min + 25 min = 55 min total
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 30 * 60 * 1000),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w2", now_ms + 31 * 60 * 1000, now_ms + 56 * 60 * 1000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
result = locker._get_today_workout_duration_minutes(db_file)
|
||||
assert result == pytest.approx(55.0, abs=1.0)
|
||||
|
||||
def test_ignores_invalid_finish(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test ignores workouts where finish <= start."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
# finish == start (zero duration - should be excluded by WHERE)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_workout_duration_minutes(db_file)
|
||||
|
||||
def test_invalid_db_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0.0 for invalid database file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
bad_file = tmp_path / "not_a_db.db"
|
||||
bad_file.write_text("not a database")
|
||||
|
||||
assert not locker._get_today_workout_duration_minutes(bad_file)
|
||||
|
||||
def test_missing_table_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0.0 when workouts table doesn't exist."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "empty.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute("CREATE TABLE other (id TEXT)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_workout_duration_minutes(db_file)
|
||||
|
||||
|
||||
class TestGetTodayExerciseCount:
|
||||
"""Tests for _get_today_exercise_count method."""
|
||||
|
||||
def test_counts_exercises(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test counts distinct exercises in today's workouts."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER, exercises TEXT)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
exercises_json = json.dumps(
|
||||
[
|
||||
{"id": "squat", "name": "Squat"},
|
||||
{"id": "bench_press", "name": "Bench Press"},
|
||||
{"id": "squat", "name": "Squat"},
|
||||
{"category": "WARMUP"},
|
||||
]
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000, exercises_json),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._get_today_exercise_count(db_file) == 2
|
||||
|
||||
def test_no_exercises_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 when no exercises exist."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER, exercises TEXT)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000, "[]"),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_exercise_count(db_file)
|
||||
|
||||
def test_invalid_db_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 for invalid database file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
bad_file = tmp_path / "bad.db"
|
||||
bad_file.write_text("not a db")
|
||||
|
||||
assert not locker._get_today_exercise_count(bad_file)
|
||||
|
||||
def test_missing_exercises_column_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 when workouts table has no exercises column."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "empty.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_exercise_count(db_file)
|
||||
|
||||
def test_null_exercises_json_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 when exercises JSON is NULL."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "null_ex.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER, exercises TEXT)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000, None),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_exercise_count(db_file)
|
||||
|
||||
def test_malformed_exercises_json_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 when exercises JSON is malformed."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "bad_json.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER, exercises TEXT)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000, "not valid json"),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._get_today_exercise_count(db_file)
|
||||
|
||||
|
||||
class TestIsWorkoutFinishRecent:
|
||||
"""Tests for _is_workout_finish_recent method."""
|
||||
|
||||
def test_recent_workout_returns_true(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns True for workout that finished recently."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
# Anchor to local noon to avoid midnight boundary issues: the SQL
|
||||
# date() filter requires start and now to share the same local date.
|
||||
local_noon = (
|
||||
datetime.datetime.now(tz=datetime.timezone.utc)
|
||||
.astimezone()
|
||||
.replace(hour=12, minute=0, second=0, microsecond=0)
|
||||
)
|
||||
local_noon_ms = int(local_noon.timestamp() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", local_noon_ms, local_noon_ms + 3_600_000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._is_workout_finish_recent(db_file) is True
|
||||
|
||||
def test_old_workout_returns_false(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns False for workout that finished >24 hours ago."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
# Finished 25 hours ago (not "today" in local time either)
|
||||
now_ms = int(time.time() * 1000)
|
||||
old_finish = now_ms - 25 * 3600 * 1000 # beyond 24h window
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", old_finish - 3600000, old_finish),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._is_workout_finish_recent(db_file) is False
|
||||
|
||||
def test_no_workouts_returns_false(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns False when no workouts exist."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._is_workout_finish_recent(db_file) is False
|
||||
|
||||
def test_invalid_db_returns_false(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns False for invalid database file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
bad_file = tmp_path / "bad.db"
|
||||
bad_file.write_text("not a db")
|
||||
|
||||
assert locker._is_workout_finish_recent(bad_file) is False
|
||||
@ -1,125 +0,0 @@
|
||||
"""Tests for _count_today_workouts and related database queries."""
|
||||
# pylint: disable=protected-access,unused-argument
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestCountTodayWorkouts:
|
||||
"""Tests for _count_today_workouts method."""
|
||||
|
||||
def test_workouts_found_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test workouts found today."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
# Insert a workout with today's timestamp (ms)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._count_today_workouts(db_file) == 1
|
||||
|
||||
def test_no_workouts_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test no workouts today."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
# Insert a workout from yesterday (24h+ ago)
|
||||
yesterday_ms = int((time.time() - 200000) * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", yesterday_ms, yesterday_ms + 3600000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._count_today_workouts(db_file)
|
||||
|
||||
def test_invalid_db_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 for invalid database file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
bad_file = tmp_path / "not_a_db.db"
|
||||
bad_file.write_text("not a database")
|
||||
|
||||
assert not locker._count_today_workouts(bad_file)
|
||||
|
||||
def test_missing_table_returns_zero(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test returns 0 when workouts table doesn't exist."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "empty.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute("CREATE TABLE other (id TEXT)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert not locker._count_today_workouts(db_file)
|
||||
|
||||
def test_multiple_workouts_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test counts multiple workouts today correctly."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
db_file = tmp_path / "sl_test.db"
|
||||
conn = sqlite3.connect(str(db_file))
|
||||
conn.execute(
|
||||
"CREATE TABLE workouts "
|
||||
"(id TEXT PRIMARY KEY, start INTEGER, finish INTEGER)",
|
||||
)
|
||||
now_ms = int(time.time() * 1000)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w1", now_ms, now_ms + 3600000),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO workouts VALUES (?, ?, ?)",
|
||||
("w2", now_ms + 100000, now_ms + 3700000),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
assert locker._count_today_workouts(db_file) == 2
|
||||
@ -1,297 +0,0 @@
|
||||
"""Tests for phone workout verification, phone check, and unlock operations."""
|
||||
# pylint: disable=protected-access,unused-argument
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestVerifyPhoneWorkout:
|
||||
"""Tests for _verify_phone_workout method."""
|
||||
|
||||
def test_verified(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test workout verified on phone with sufficient duration."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=tmp_path / "sl.db"),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_count_today_workouts",
|
||||
MagicMock(return_value=2),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_workout_finish_recent",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_get_today_exercise_count",
|
||||
MagicMock(return_value=3),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_get_today_workout_duration_minutes",
|
||||
MagicMock(return_value=65.0),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "verified"
|
||||
assert "2 session" in message
|
||||
assert "65 min" in message
|
||||
assert "3 exercise" in message
|
||||
|
||||
def test_not_verified(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test no workout found on phone."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=tmp_path / "sl.db"),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_count_today_workouts",
|
||||
MagicMock(return_value=0),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "not_verified"
|
||||
assert "No workout" in message
|
||||
|
||||
def test_too_short(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test workout found but too short."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=tmp_path / "sl.db"),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_count_today_workouts",
|
||||
MagicMock(return_value=1),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_workout_finish_recent",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_get_today_exercise_count",
|
||||
MagicMock(return_value=3),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_get_today_workout_duration_minutes",
|
||||
MagicMock(return_value=25.0),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "too_short"
|
||||
assert "25 min" in message
|
||||
assert "50 min" in message
|
||||
|
||||
def test_no_phone(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test no phone connected."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=False),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, _ = locker._verify_phone_workout()
|
||||
|
||||
assert status == "no_phone"
|
||||
|
||||
def test_error_no_db(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test error when StrongLifts DB cannot be pulled."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=None),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "error"
|
||||
assert "database" in message.lower()
|
||||
|
||||
def test_clock_tampered(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test clock_tampered when NTP check fails."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(False, "System clock is 600s ahead"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "clock_tampered"
|
||||
assert "600s" in message
|
||||
|
||||
def test_stale_workout(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test stale status when workout finish is not recent."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=tmp_path / "sl.db"),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_count_today_workouts",
|
||||
MagicMock(return_value=1),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_workout_finish_recent",
|
||||
MagicMock(return_value=False),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "stale"
|
||||
assert "old" in message.lower()
|
||||
|
||||
def test_no_exercises(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test no_exercises when workout has no exercise data."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_phone_connected",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_pull_stronglifts_db",
|
||||
MagicMock(return_value=tmp_path / "sl.db"),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_count_today_workouts",
|
||||
MagicMock(return_value=1),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_is_workout_finish_recent",
|
||||
MagicMock(return_value=True),
|
||||
)
|
||||
object.__setattr__(
|
||||
locker,
|
||||
"_get_today_exercise_count",
|
||||
MagicMock(return_value=0),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, "Clock OK"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
|
||||
assert status == "no_exercises"
|
||||
assert "exercise" in message.lower()
|
||||
254
screen_locker/tests/test_phone_verification_part3.py
Normal file
254
screen_locker/tests/test_phone_verification_part3.py
Normal file
@ -0,0 +1,254 @@
|
||||
"""Tests for multi-path workout-JSON pull and HTTP fall-through (part 3).
|
||||
|
||||
Covers the fix for the path mismatch where the app writes
|
||||
``/sdcard/workout_result.json`` (primary) but the locker only checked the
|
||||
app-external fallback path. The locker now pulls every candidate path, prefers
|
||||
the one dated today, and falls through to the HTTP scan when an ADB pull yields
|
||||
no usable JSON even though a device is connected.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import time
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker._constants import WORKOUT_APP_JSON_REMOTES
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
_PRIMARY, _FALLBACK = WORKOUT_APP_JSON_REMOTES
|
||||
|
||||
|
||||
def _today() -> str:
|
||||
"""Return today's date as the app stamps it (local YYYY-MM-DD)."""
|
||||
return time.strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def _adb_pull_from(path_contents: dict[str, str]) -> MagicMock:
|
||||
"""Build a fake ``_run_adb`` that writes per-remote content to the dest file.
|
||||
|
||||
``path_contents`` maps a remote path to the JSON text the device would
|
||||
return for it. A remote absent from the map simulates a missing file
|
||||
(``adb pull`` failure).
|
||||
"""
|
||||
|
||||
def fake_run_adb(args: list[str]) -> tuple[bool, str]:
|
||||
if not args or args[0] != "pull":
|
||||
return False, ""
|
||||
remote, dest = args[1], args[2]
|
||||
content = path_contents.get(remote)
|
||||
if content is None:
|
||||
return False, ""
|
||||
Path(dest).write_text(content)
|
||||
return True, ""
|
||||
|
||||
return MagicMock(side_effect=fake_run_adb)
|
||||
|
||||
|
||||
class TestPullWorkoutAppJsonMultiPath:
|
||||
"""Tests for _pull_workout_app_json across multiple candidate paths."""
|
||||
|
||||
def test_prefers_todays_data_over_stale_primary(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""A stale primary file must not shadow today's fallback file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
stale = json.dumps({"date": "2000-01-01", "exercises": ["old"]})
|
||||
fresh = json.dumps({"date": _today(), "exercises": ["new"]})
|
||||
fake = _adb_pull_from({_PRIMARY: stale, _FALLBACK: fresh})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is not None
|
||||
assert result["date"] == _today()
|
||||
assert result["exercises"] == ["new"]
|
||||
|
||||
def test_returns_primary_when_dated_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Primary path dated today is returned without needing the fallback."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
fresh = json.dumps({"date": _today(), "exercises": ["primary"]})
|
||||
fake = _adb_pull_from({_PRIMARY: fresh})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is not None
|
||||
assert result["exercises"] == ["primary"]
|
||||
|
||||
def test_falls_back_to_stale_when_none_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""With no today's file, the first parseable payload is returned."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
stale = json.dumps({"date": "2000-01-01", "exercises": ["old"]})
|
||||
fake = _adb_pull_from({_FALLBACK: stale})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is not None
|
||||
assert result["date"] == "2000-01-01"
|
||||
|
||||
def test_keeps_first_stale_when_multiple_non_today(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""With several non-today files, the first parseable one is kept."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
first = json.dumps({"date": "2000-01-01", "exercises": ["primary"]})
|
||||
second = json.dumps({"date": "1999-12-31", "exercises": ["fallback"]})
|
||||
fake = _adb_pull_from({_PRIMARY: first, _FALLBACK: second})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is not None
|
||||
assert result["exercises"] == ["primary"]
|
||||
|
||||
def test_skips_unparseable_payload(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""A corrupt file at one path doesn't block today's file at another."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
fresh = json.dumps({"date": _today(), "exercises": ["new"]})
|
||||
fake = _adb_pull_from({_PRIMARY: "{not valid json", _FALLBACK: fresh})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is not None
|
||||
assert result["date"] == _today()
|
||||
|
||||
def test_returns_none_when_no_candidate_pulls(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Returns None when no candidate path yields a file."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
fake = _adb_pull_from({})
|
||||
with patch.object(locker, "_run_adb", fake):
|
||||
result = locker._pull_workout_app_json()
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestVerifyPhoneWorkoutFallthrough:
|
||||
"""Tests for the ADB→HTTP fall-through in _verify_phone_workout."""
|
||||
|
||||
def test_adb_connected_but_pull_empty_falls_through_to_http(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""A connected device with no pullable JSON still tries the HTTP scan."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
http_data = {
|
||||
"date": _today(),
|
||||
"exercises": ["a"],
|
||||
"duration_seconds": 4000,
|
||||
"succeeded": True,
|
||||
}
|
||||
with (
|
||||
patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, ""),
|
||||
),
|
||||
patch.object(locker, "_is_phone_connected", return_value=True),
|
||||
patch.object(locker, "_pull_workout_app_json", return_value=None),
|
||||
patch.object(locker, "_fetch_http_workout", return_value=http_data),
|
||||
):
|
||||
status, _ = locker._verify_phone_workout()
|
||||
assert status == "verified"
|
||||
|
||||
def test_adb_connected_pull_and_http_empty_is_not_verified(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Connected device, no JSON anywhere → not_verified (not no_phone)."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with (
|
||||
patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, ""),
|
||||
),
|
||||
patch.object(locker, "_is_phone_connected", return_value=True),
|
||||
patch.object(locker, "_pull_workout_app_json", return_value=None),
|
||||
patch.object(locker, "_fetch_http_workout", return_value=None),
|
||||
):
|
||||
status, _ = locker._verify_phone_workout()
|
||||
assert status == "not_verified"
|
||||
|
||||
def test_adb_pull_success_returns_without_http(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""A successful ADB pull validates directly, never touching HTTP."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
data = {
|
||||
"date": _today(),
|
||||
"exercises": ["a"],
|
||||
"duration_seconds": 4000,
|
||||
"succeeded": False,
|
||||
}
|
||||
http = MagicMock()
|
||||
with (
|
||||
patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, ""),
|
||||
),
|
||||
patch.object(locker, "_is_phone_connected", return_value=True),
|
||||
patch.object(locker, "_pull_workout_app_json", return_value=data),
|
||||
patch.object(locker, "_fetch_http_workout", http),
|
||||
):
|
||||
status, _ = locker._verify_phone_workout()
|
||||
assert status == "verified"
|
||||
http.assert_not_called()
|
||||
|
||||
def test_clock_tampered_short_circuits(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""A clock-skew failure returns clock_tampered before any phone access."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(False, "clock skew too large"),
|
||||
):
|
||||
status, message = locker._verify_phone_workout()
|
||||
assert status == "clock_tampered"
|
||||
assert message == "clock skew too large"
|
||||
|
||||
def test_no_device_and_http_empty_is_no_phone(
|
||||
self,
|
||||
mock_tk: MagicMock,
|
||||
mock_sys_exit: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""No device and no HTTP server → no_phone."""
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with (
|
||||
patch(
|
||||
"screen_locker._phone_verification.check_clock_skew",
|
||||
return_value=(True, ""),
|
||||
),
|
||||
patch.object(locker, "_is_phone_connected", return_value=False),
|
||||
patch.object(locker, "_fetch_http_workout", return_value=None),
|
||||
):
|
||||
status, _ = locker._verify_phone_workout()
|
||||
assert status == "no_phone"
|
||||
209
screen_locker/tests/test_phone_verification_part4.py
Normal file
209
screen_locker/tests/test_phone_verification_part4.py
Normal file
@ -0,0 +1,209 @@
|
||||
"""Tests for JSON workout validation and HTTP fallback (part 4).
|
||||
|
||||
Replaces the obsolete StrongLifts-DB-based ``test_phone_check_unlock.py``.
|
||||
Covers ``_validate_json_data`` (all status branches) and the HTTP fallback
|
||||
``_scan_for_http_server`` / ``_fetch_http_workout`` used when ADB is
|
||||
unavailable. Network is fully mocked — no test touches a real socket.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from screen_locker._constants import MIN_WORKOUT_DURATION_MINUTES
|
||||
from screen_locker.tests.conftest import create_locker
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _today() -> str:
|
||||
"""Return today's date as the validator computes it (local YYYY-MM-DD)."""
|
||||
return time.strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def _mock_cm(return_value: MagicMock) -> MagicMock:
|
||||
"""Build a MagicMock usable as a context manager yielding ``return_value``."""
|
||||
cm = MagicMock()
|
||||
cm.__enter__ = MagicMock(return_value=return_value)
|
||||
cm.__exit__ = MagicMock(return_value=False)
|
||||
return cm
|
||||
|
||||
|
||||
class TestValidateJsonData:
|
||||
"""Tests for _validate_json_data across every status branch."""
|
||||
|
||||
def test_stale_when_not_today(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
status, message = locker._validate_json_data(
|
||||
{"date": "2000-01-01", "exercises": ["x"], "duration_seconds": 4000}
|
||||
)
|
||||
assert status == "stale"
|
||||
assert "2000-01-01" in message
|
||||
|
||||
def test_no_exercises_when_empty(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
status, message = locker._validate_json_data(
|
||||
{"date": _today(), "exercises": [], "duration_seconds": 4000}
|
||||
)
|
||||
assert status == "no_exercises"
|
||||
assert "exercise" in message.lower()
|
||||
|
||||
def test_too_short_under_minimum(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
short_seconds = int((MIN_WORKOUT_DURATION_MINUTES - 10) * 60)
|
||||
status, message = locker._validate_json_data(
|
||||
{"date": _today(), "exercises": ["x"], "duration_seconds": short_seconds}
|
||||
)
|
||||
assert status == "too_short"
|
||||
assert f"{MIN_WORKOUT_DURATION_MINUTES}" in message
|
||||
|
||||
def test_verified_all_succeeded(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
status, message = locker._validate_json_data(
|
||||
{
|
||||
"date": _today(),
|
||||
"exercises": ["x"],
|
||||
"duration_seconds": 6000,
|
||||
"succeeded": True,
|
||||
}
|
||||
)
|
||||
assert status == "verified"
|
||||
assert "all succeeded" in message
|
||||
|
||||
def test_verified_partial_when_not_succeeded(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
status, message = locker._validate_json_data(
|
||||
{
|
||||
"date": _today(),
|
||||
"exercises": ["x"],
|
||||
"duration_seconds": 6000,
|
||||
"succeeded": False,
|
||||
}
|
||||
)
|
||||
assert status == "verified"
|
||||
assert "partial" in message
|
||||
|
||||
|
||||
class TestScanForHttpServer:
|
||||
"""Tests for _scan_for_http_server subnet probing."""
|
||||
|
||||
def test_returns_none_without_prefix(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with patch.object(locker, "_get_local_subnet_prefix", return_value=None):
|
||||
assert locker._scan_for_http_server() is None
|
||||
|
||||
def test_returns_url_when_probe_connects(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with (
|
||||
patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"),
|
||||
patch(
|
||||
"screen_locker._phone_verification.socket.create_connection",
|
||||
return_value=_mock_cm(MagicMock()),
|
||||
),
|
||||
):
|
||||
result = locker._scan_for_http_server()
|
||||
assert result is not None
|
||||
assert result.startswith("http://192.168.1.")
|
||||
assert result.endswith(":8765/workout")
|
||||
|
||||
def test_returns_none_when_all_probes_refused(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with (
|
||||
patch.object(locker, "_get_local_subnet_prefix", return_value="192.168.1"),
|
||||
patch(
|
||||
"screen_locker._phone_verification.socket.create_connection",
|
||||
side_effect=OSError("refused"),
|
||||
),
|
||||
):
|
||||
assert locker._scan_for_http_server() is None
|
||||
|
||||
|
||||
class TestFetchHttpWorkout:
|
||||
"""Tests for _fetch_http_workout over the local HTTP server."""
|
||||
|
||||
def test_returns_none_when_scan_finds_nothing(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with patch.object(locker, "_scan_for_http_server", return_value=None):
|
||||
assert locker._fetch_http_workout() is None
|
||||
|
||||
def test_returns_json_on_http_200(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
resp = MagicMock()
|
||||
resp.status = 200
|
||||
resp.read.return_value = b'{"date": "2026-06-12", "exercises": ["a"]}'
|
||||
conn = MagicMock()
|
||||
conn.getresponse.return_value = resp
|
||||
with (
|
||||
patch.object(
|
||||
locker,
|
||||
"_scan_for_http_server",
|
||||
return_value="http://192.168.1.5:8765/workout",
|
||||
),
|
||||
patch(
|
||||
"screen_locker._phone_verification._HTTPConnection",
|
||||
return_value=conn,
|
||||
),
|
||||
):
|
||||
result = locker._fetch_http_workout()
|
||||
assert result == {"date": "2026-06-12", "exercises": ["a"]}
|
||||
|
||||
def test_returns_none_on_non_ok_status(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
resp = MagicMock()
|
||||
resp.status = 404
|
||||
conn = MagicMock()
|
||||
conn.getresponse.return_value = resp
|
||||
with (
|
||||
patch.object(
|
||||
locker,
|
||||
"_scan_for_http_server",
|
||||
return_value="http://192.168.1.5:8765/workout",
|
||||
),
|
||||
patch(
|
||||
"screen_locker._phone_verification._HTTPConnection",
|
||||
return_value=conn,
|
||||
),
|
||||
):
|
||||
assert locker._fetch_http_workout() is None
|
||||
|
||||
def test_returns_none_on_connection_error(
|
||||
self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
|
||||
) -> None:
|
||||
locker = create_locker(mock_tk, tmp_path)
|
||||
with (
|
||||
patch.object(
|
||||
locker,
|
||||
"_scan_for_http_server",
|
||||
return_value="http://192.168.1.5:8765/workout",
|
||||
),
|
||||
patch(
|
||||
"screen_locker._phone_verification._HTTPConnection",
|
||||
side_effect=OSError("unreachable"),
|
||||
),
|
||||
):
|
||||
assert locker._fetch_http_workout() is None
|
||||
Loading…
Reference in New Issue
Block a user