diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 5839c3e..cd844f6 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -11,7 +11,7 @@ jobs:
runs-on: ${{ matrix.os }}
strategy:
matrix:
- os: [ubuntu-latest, macos-latest]
+ os: [ubuntu-latest, macos-latest, windows-latest]
python-version: ["3.12", "3.13"]
steps:
- uses: actions/checkout@v5
diff --git a/README.md b/README.md
index c609968..4017227 100644
--- a/README.md
+++ b/README.md
@@ -7,7 +7,7 @@
Syncs body composition from a Eufy smart scale to Garmin Connect and Strava.
-> macOS only. Needs Python 3.12+ and a terminal.
+> macOS and Windows. Needs Python 3.12+ and a terminal.
## Sync targets
@@ -48,6 +48,14 @@ source $HOME/.local/bin/env
uv tool install eufy-sync
```
+**On Windows:** open PowerShell (press Start, type "powershell", hit Enter) and install uv the same way:
+
+```powershell
+powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
+```
+
+Open a fresh terminal so the `uv` command is found, then run `uv tool install eufy-sync`.
+
Then open a new terminal window, so it picks up the newly installed command, and run:
```bash
@@ -87,12 +95,14 @@ eufy-sync checks weekly and tells you when a new version is out. To update:
eufy-sync --update
```
-## Automatic sync (macOS)
+## Automatic sync
On first run you can opt into syncing every 4 hours. If you do, a macOS Launch Agent runs it in the background: weigh yourself, open your laptop later, and it syncs on its own. Logs go to `~/.garmin-sync/sync.log`, and you get a notification if something fails. Turn it off with `eufy-sync --uninstall-agent`.
If [terminal-notifier](https://github.com/julienXX/terminal-notifier) is installed (`brew install terminal-notifier`), clicking a failure notification opens Terminal with the fix command already running. Without it, notifications still appear; the click just does nothing useful.
+On Windows the same option registers a Scheduled Task that runs every 4 hours with no visible window. When a run fails, a toast notification names the command to fix it.
+
## Headless Linux (server or VPS)
eufy-sync runs on Linux too, and a server is a good home for it: no laptop that has to be awake. Without a system keychain, credentials fall back to a file with `600` permissions.
@@ -150,7 +160,7 @@ On each run it pulls your Eufy history and checks a local SQLite DB for what eac
## Security
-Passwords and OAuth tokens live in a single item in your macOS Keychain, not plaintext files. The config in `~/.garmin-sync/` holds only email addresses and Strava app credentials, at `600` permissions. Credentials go over HTTPS to Eufy, Garmin, and Strava only, and are never logged or sent anywhere else. The one other outbound call is a weekly version check to pypi.org, with no credentials. On a host without a keychain (headless Linux), or if you choose `--use-file-store`, credentials fall back to a single `600` file at `~/.garmin-sync/credentials.json`.
+Passwords and OAuth tokens live in a single item in your macOS Keychain, not plaintext files. The config in `~/.garmin-sync/` holds only email addresses and Strava app credentials, at `600` permissions. Credentials go over HTTPS to Eufy, Garmin, and Strava only, and are never logged or sent anywhere else. The one other outbound call is a weekly version check to pypi.org, with no credentials. On a host without a keychain (headless Linux), or if you choose `--use-file-store`, credentials fall back to a single `600` file at `~/.garmin-sync/credentials.json`. On Windows the credential vault lives in Windows Credential Manager, and the file fallback relies on your user profile's permissions since Windows does not honor POSIX file modes.
### Credential storage
diff --git a/eufy_sync/__init__.py b/eufy_sync/__init__.py
index 47ae70b..a4a6ee8 100644
--- a/eufy_sync/__init__.py
+++ b/eufy_sync/__init__.py
@@ -1,6 +1,6 @@
"""Sync Eufy smart scale body composition data to Garmin Connect and Strava."""
-__version__ = "1.7.22"
+__version__ = "1.8.0"
# Public API for programmatic use
from eufy_sync.garmin_auth import GarminAuth
diff --git a/eufy_sync/cli/app.py b/eufy_sync/cli/app.py
index c1b0589..17f2c92 100644
--- a/eufy_sync/cli/app.py
+++ b/eufy_sync/cli/app.py
@@ -4,10 +4,21 @@
import sys
from pathlib import Path
+from eufy_sync import platform_support
from eufy_sync.cli import doctor, maintenance, profiles, setup, shared, status, updater
def main() -> None:
+ try:
+ _main()
+ except KeyboardInterrupt:
+ # Ctrl+C at any prompt (MFA code, passwords, confirmations) should
+ # read as a cancel, not dump a traceback.
+ print("\nCancelled.")
+ sys.exit(130)
+
+
+def _main() -> None:
import argparse
from eufy_sync import __version__
@@ -29,11 +40,11 @@ def main() -> None:
help="Show recent sync history, last N entries (default: 14)")
parser.add_argument("--backfill-days", type=int, default=None, help="Sync last N days")
parser.add_argument("--dry-run", action="store_true", help="Preview without uploading")
- parser.add_argument("--headless", action="store_true", help="Never prompt; fail with a reauth message if login is needed (for Launch Agent)")
+ parser.add_argument("--headless", action="store_true", help="Never prompt; fail with a reauth message if login is needed (for scheduled runs)")
parser.add_argument("--verbose", "-v", action="store_true", help="Show detailed sync logs")
- parser.add_argument("--install-agent", action="store_true", help="Set up automatic sync (macOS Launch Agent)")
- parser.add_argument("--uninstall-agent", action="store_true", help="Remove the automatic sync Launch Agent")
- parser.add_argument("--uninstall", action="store_true", help="Remove all data, tokens, and Launch Agent")
+ parser.add_argument("--install-agent", action="store_true", help="Set up automatic sync every 4 hours")
+ parser.add_argument("--uninstall-agent", action="store_true", help="Remove automatic sync")
+ parser.add_argument("--uninstall", action="store_true", help="Remove all data, tokens, and automatic sync")
parser.add_argument("--use-file-store", action="store_true", help="Store credentials in a 0o600 file instead of the keychain (no keychain prompts)")
parser.add_argument("--use-keychain", action="store_true", help="Move credentials back into the system keychain")
parser.add_argument("--config", type=Path, default=None, help="Config path (default: ~/.garmin-sync/config.yaml)")
@@ -43,6 +54,12 @@ def main() -> None:
config_path = args.config or shared.DEFAULT_CONFIG
db_path = args.db or shared.DEFAULT_DB
+ # Configure logging before any command dispatch. Every path that can reach
+ # a Garmin login (setup, --update-password, --reauth, sync) needs the
+ # garminconnect logger quieted, or its per-strategy 429 warnings print
+ # through logging's last-resort handler and look like errors.
+ shared._configure_logging(args.verbose)
+
# Handle doctor - must report even on a fresh install, never launch the
# first-run wizard, and never traceback.
if args.doctor:
@@ -105,7 +122,6 @@ def main() -> None:
# Handle reauth
if args.reauth is not None:
- shared._configure_logging(args.verbose)
target = None if args.reauth == "all" else args.reauth
maintenance._reauth(config_path, force=True, target=target)
return
@@ -123,7 +139,7 @@ def main() -> None:
if first_run and args.headless:
msg = "No config found. Run eufy-sync in a terminal to set up."
print(msg)
- shared._notify("eufy-sync", msg)
+ platform_support.notify("eufy-sync", msg)
sys.exit(1)
try:
@@ -143,7 +159,7 @@ def main() -> None:
except Exception as e:
msg = f"eufy-sync could not start: {e}"
print(msg)
- shared._notify("eufy-sync failed", str(e)[:200])
+ platform_support.notify("eufy-sync failed", str(e)[:200])
sys.exit(1)
has_garmin = any(u.garmin for u in config.users)
@@ -178,7 +194,6 @@ def main() -> None:
from eufy_sync.state import SyncState
from eufy_sync.eufy_client import AmbiguousProfileError
- shared._configure_logging(args.verbose)
logger = logging.getLogger("eufy_sync")
updater._check_for_updates()
@@ -192,7 +207,7 @@ def main() -> None:
except Exception as e:
msg = f"eufy-sync could not start: {e}"
print(msg)
- shared._notify("eufy-sync failed", str(e)[:200])
+ platform_support.notify("eufy-sync failed", str(e)[:200])
sys.exit(1)
try:
@@ -245,13 +260,13 @@ def main() -> None:
multiple_profiles = any("multiple Eufy profiles" in err for _, err in failures)
all_transient = all(failure_notify.is_transient_network_error(err) for _, err in failures)
if reauth_needed:
- shared._notify("eufy-sync: re-login needed", "Run: eufy-sync --reauth garmin", command="eufy-sync --reauth garmin")
+ platform_support.notify("eufy-sync: re-login needed", "Run: eufy-sync --reauth garmin", command="eufy-sync --reauth garmin")
failure_notify.clear_network_failures()
elif eufy_password:
- shared._notify("eufy-sync: Eufy login failed", "Run: eufy-sync --update-password", command="eufy-sync --update-password")
+ platform_support.notify("eufy-sync: Eufy login failed", "Run: eufy-sync --update-password", command="eufy-sync --update-password")
failure_notify.clear_network_failures()
elif multiple_profiles:
- shared._notify("eufy-sync: choose your profile", "Run: eufy-sync --select-profile", command="eufy-sync --select-profile")
+ platform_support.notify("eufy-sync: choose your profile", "Run: eufy-sync --select-profile", command="eufy-sync --select-profile")
failure_notify.clear_network_failures()
elif all_transient and args.headless and not args.dry_run:
# A scheduled run that only hit network trouble. Stay quiet - the
@@ -259,14 +274,14 @@ def main() -> None:
# points at a real outage worth one heads-up.
count, hours = failure_notify.record_network_failure()
if failure_notify.should_escalate(count):
- shared._notify(
+ platform_support.notify(
"eufy-sync: network still down",
f"No network for ~{round(hours)}h ({count} runs). "
"Measurements are waiting and will sync when it is back.",
)
else:
fail_msg = "; ".join(f"{name}: {err[:80]}" for name, err in failures)
- shared._notify("eufy-sync failed", fail_msg)
+ platform_support.notify("eufy-sync failed", fail_msg)
failure_notify.clear_network_failures()
logger.error("Sync failed for: %s", "; ".join(f"{n}: {e[:80]}" for n, e in failures))
elif not args.dry_run:
@@ -285,7 +300,7 @@ def main() -> None:
if total > 0:
target_label = " and ".join(n.capitalize() for n in total_counts if total_counts[n] > 0)
- shared._notify("eufy-sync", f"Synced {total} measurement{'s' if total != 1 else ''} to {target_label}")
+ platform_support.notify("eufy-sync", f"Synced {total} measurement{'s' if total != 1 else ''} to {target_label}")
if first_run:
if failures:
diff --git a/eufy_sync/cli/doctor.py b/eufy_sync/cli/doctor.py
index fb7c4f3..a418668 100644
--- a/eufy_sync/cli/doctor.py
+++ b/eufy_sync/cli/doctor.py
@@ -8,14 +8,12 @@
"""
from __future__ import annotations
-import platform
-import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from eufy_sync import credentials
-from eufy_sync.cli import shared
+from eufy_sync import platform_support
from eufy_sync.cli import updater
from eufy_sync.config import load_config
from eufy_sync.credentials import _keyring_available, active_store_label
@@ -94,9 +92,10 @@ def report(status: str, label: str, detail: str, fix: str | None = None) -> None
# 7. eufy cloud
_check_eufy_cloud(report, eufy_client)
- # 8. launch agent (macOS only)
- if platform.system() == "Darwin":
- _check_launch_agent(report)
+ # 8. scheduled sync agent (where the platform manages one)
+ agent = platform_support.agent_status()
+ if agent is not None:
+ report(agent["status"], agent["label"], agent["detail"], agent["fix"])
# 9. state db
_check_state_db(report, db_path, user)
@@ -231,37 +230,6 @@ def _check_eufy_cloud(report, eufy_client) -> None:
pass
-def _check_launch_agent(report) -> None:
- try:
- if not shared.LAUNCH_AGENT_PATH.exists():
- report("WARN", "launch agent", "not installed", "eufy-sync --install-agent")
- return
-
- content = shared.LAUNCH_AGENT_PATH.read_text()
- wrapper_name = shared.LAUNCH_WRAPPER_NAME
- if wrapper_name not in content:
- report(
- "WARN", "launch agent",
- "outdated registration (re-announces on every update)",
- "eufy-sync --install-agent",
- )
- return
-
- result = subprocess.run(
- ["launchctl", "list"],
- capture_output=True,
- text=True,
- timeout=5,
- )
- if shared.LAUNCH_AGENT_LABEL not in (result.stdout or ""):
- report("WARN", "launch agent", "installed but not loaded", "eufy-sync --install-agent")
- return
-
- report("PASS", "launch agent", "loaded, runs every 4h")
- except Exception as e:
- report("WARN", "launch agent", f"could not check ({e})", "eufy-sync --install-agent")
-
-
def _check_state_db(report, db_path: Path, user) -> None:
try:
state = SyncState(db_path)
diff --git a/eufy_sync/cli/maintenance.py b/eufy_sync/cli/maintenance.py
index db8eee9..b257188 100644
--- a/eufy_sync/cli/maintenance.py
+++ b/eufy_sync/cli/maintenance.py
@@ -1,15 +1,14 @@
-"""Password updates, re-auth, and the macOS Launch Agent lifecycle."""
+"""Password updates, re-auth, and the scheduled-sync agent lifecycle."""
from __future__ import annotations
import getpass
-import platform
import shutil
-import subprocess
import sys
from pathlib import Path
import yaml
+from eufy_sync import platform_support
from eufy_sync.cli import shared
@@ -136,127 +135,19 @@ def _reauth(config_path: Path, config: dict | None = None, force: bool = False,
print("Done - Strava tokens saved.")
-def _write_run_script(binary_path: str) -> Path:
- """Write the stable wrapper script the Launch Agent runs.
-
- macOS re-announces "can run in the background" whenever a registered
- background item's executable changes identity, and pipx/uv replace the
- binary on every update. The agent therefore points at this script, whose
- bytes never change across updates, so the announcement fires once, not
- once per release. Skipping the rewrite when content is unchanged is what
- keeps the file's identity stable. The filename is what macOS shows in that
- announcement, so it is a recognizable "eufy-sync-agent", not an opaque one.
- """
- script_path = shared.DATA_DIR / shared.LAUNCH_WRAPPER_NAME
- content = f'#!/bin/sh\nexec "{binary_path}" --headless\n'
- if script_path.exists() and script_path.read_text() == content:
- return script_path
- script_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
- script_path.write_text(content)
- script_path.chmod(0o755)
- return script_path
-
-
-def _generate_plist(program_path: str) -> str:
- """Generate a Launch Agent plist that runs the given program every 4 hours."""
- log_path = str(shared.LOG_FILE)
- return f"""
-
-
-
- Label
- {shared.LAUNCH_AGENT_LABEL}
-
- ProgramArguments
-
- {program_path}
-
-
- StartInterval
- 14400
-
- RunAtLoad
-
-
- StandardOutPath
- {log_path}
- StandardErrorPath
- {log_path}
-
-
-"""
-
-
def _install_launch_agent() -> None:
- """Install the macOS Launch Agent for automatic sync."""
- if platform.system() != "Darwin":
- print("Auto-sync is only supported on macOS.")
- return
-
- binary = shutil.which("eufy-sync")
- if not binary:
- print("Warning: could not find eufy-sync on PATH. Skipping auto-sync setup.")
- return
-
- already_installed = shared.LAUNCH_AGENT_PATH.exists()
-
- # Ensure the log directory exists with restricted permissions
- shared.DATA_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
-
- wrapper = _write_run_script(binary)
-
- # Drop the pre-1.7.17 wrapper name so it does not linger as an orphan next
- # to the new one.
- legacy_wrapper = shared.DATA_DIR / shared.LEGACY_LAUNCH_WRAPPER_NAME
- if legacy_wrapper.exists():
- legacy_wrapper.unlink()
-
- shared.LAUNCH_AGENT_PATH.parent.mkdir(parents=True, exist_ok=True)
- shared.LAUNCH_AGENT_PATH.write_text(_generate_plist(str(wrapper)))
-
- # Unload first in case an old version is loaded
- subprocess.run(
- ["launchctl", "unload", str(shared.LAUNCH_AGENT_PATH)],
- capture_output=True,
- )
- subprocess.run(
- ["launchctl", "load", str(shared.LAUNCH_AGENT_PATH)],
- capture_output=True,
- )
-
- if already_installed:
- print(f"Launch Agent already installed (reloaded). Logs: {shared.LOG_FILE}")
- else:
- print(f"Automatic sync installed. Logs: {shared.LOG_FILE}")
+ """Install the scheduled-sync agent for the current platform."""
+ platform_support.install_agent()
def _offer_launch_agent() -> None:
- """Offer to install a macOS Launch Agent after first-run setup."""
- if platform.system() != "Darwin":
- return
- if not sys.stdin.isatty():
- return
-
- print("")
- answer = input("Set up automatic sync every 4 hours? [y/N] ").strip()
- if not answer.lower().startswith("y"):
- return
-
- _install_launch_agent()
+ """Offer to install the scheduled-sync agent after first-run setup."""
+ platform_support.offer_agent()
def _uninstall_launch_agent() -> None:
- """Remove the macOS Launch Agent."""
- if not shared.LAUNCH_AGENT_PATH.exists():
- print("No Launch Agent installed.")
- return
-
- subprocess.run(
- ["launchctl", "unload", str(shared.LAUNCH_AGENT_PATH)],
- capture_output=True,
- )
- shared.LAUNCH_AGENT_PATH.unlink()
- print("Launch Agent removed. Auto-sync disabled.")
+ """Remove the scheduled-sync agent for the current platform."""
+ platform_support.uninstall_agent()
def _uninstall(data_dir: Path, config_path: Path | None = None, db_path: Path | None = None) -> None:
@@ -274,8 +165,8 @@ def _uninstall(data_dir: Path, config_path: Path | None = None, db_path: Path |
print(f" - All saved credentials and tokens in {data_dir}/")
print(f" - Keychain entries for eufy-sync")
print(f" - Sync history database")
- if shared.LAUNCH_AGENT_PATH.exists():
- print(f" - Automatic sync Launch Agent")
+ if platform_support.agent_installed():
+ print(f" - Automatic sync")
print("")
answer = input("Are you sure? [y/N] ").strip()
@@ -295,10 +186,9 @@ def _uninstall(data_dir: Path, config_path: Path | None = None, db_path: Path |
keep_answer = input("Keep sync history? Prevents duplicates if you reinstall later. [Y/n] ").strip()
keep_db = not keep_answer.lower().startswith("n")
- # Stop and remove Launch Agent
- if shared.LAUNCH_AGENT_PATH.exists():
- subprocess.run(["launchctl", "unload", str(shared.LAUNCH_AGENT_PATH)], capture_output=True)
- shared.LAUNCH_AGENT_PATH.unlink()
+ # Stop and remove the scheduled-sync agent, where the platform manages one.
+ if platform_support.agent_installed():
+ platform_support.purge_agent()
# Clear keychain entries for every user named in the config
user_names = ["default"]
diff --git a/eufy_sync/cli/shared.py b/eufy_sync/cli/shared.py
index 72edb1a..9f8bbba 100644
--- a/eufy_sync/cli/shared.py
+++ b/eufy_sync/cli/shared.py
@@ -1,11 +1,7 @@
-"""Path/agent constants and small helpers shared across the cli package."""
+"""Path constants and small helpers shared across the cli package."""
from __future__ import annotations
-import json
import os
-import shlex
-import shutil
-import subprocess
from pathlib import Path
import yaml
@@ -14,70 +10,10 @@
DEFAULT_CONFIG = DATA_DIR / "config.yaml"
DEFAULT_DB = DATA_DIR / "state.db"
LOG_FILE = DATA_DIR / "sync.log"
-LAUNCH_AGENT_LABEL = "com.sturimcode.eufy-garmin-sync"
-LAUNCH_AGENT_PATH = Path.home() / "Library" / "LaunchAgents" / f"{LAUNCH_AGENT_LABEL}.plist"
-
-# The Launch Agent points at this wrapper script, not the pipx/uv binary,
-# so the registered executable's bytes stay stable across updates (macOS
-# only re-announces "can run in the background" when they change). macOS
-# labels the background-item notification by this filename, so it is named
-# recognizably rather than something opaque. LEGACY_LAUNCH_WRAPPER_NAME is
-# the pre-1.7.17 name, cleaned up on the next --install-agent.
-LAUNCH_WRAPPER_NAME = "eufy-sync-agent"
-LEGACY_LAUNCH_WRAPPER_NAME = "run-sync.sh"
UPDATE_CHECK_INTERVAL = 604800 # check once per week
-def _find_terminal_notifier() -> str | None:
- """Locate terminal-notifier (optional Homebrew tool). Checked beyond
- PATH because launchd runs with a minimal PATH that excludes Homebrew."""
- found = shutil.which("terminal-notifier")
- if found:
- return found
- for candidate in ("/opt/homebrew/bin/terminal-notifier", "/usr/local/bin/terminal-notifier"):
- if os.path.exists(candidate):
- return candidate
- return None
-
-
-def _notify(title: str, message: str, command: str | None = None) -> None:
- """Send a macOS notification. Fails silently on other platforms.
-
- When a fix command is given and terminal-notifier is installed,
- clicking the notification opens a Terminal window running the command,
- so the user lands in the interactive prompts instead of Script Editor
- (plain osascript notifications belong to Script Editor and clicking
- them just launches it).
- """
- try:
- if command:
- notifier = _find_terminal_notifier()
- if notifier:
- do_script = f'tell application "Terminal" to do script "{command}"'
- activate = 'tell application "Terminal" to activate'
- subprocess.run(
- [
- notifier,
- "-title", title,
- "-message", message,
- "-execute", f"osascript -e {shlex.quote(do_script)} -e {shlex.quote(activate)}",
- ],
- capture_output=True,
- timeout=5,
- )
- return
- safe_title = json.dumps(title)
- safe_msg = json.dumps(message)
- subprocess.run(
- ["osascript", "-e", f'display notification {safe_msg} with title {safe_title}'],
- capture_output=True,
- timeout=5,
- )
- except Exception:
- pass
-
-
def _write_config(path: Path, config: dict) -> None:
"""Write config with restricted permissions."""
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
diff --git a/eufy_sync/cli/updater.py b/eufy_sync/cli/updater.py
index 5d35ad3..bde8d07 100644
--- a/eufy_sync/cli/updater.py
+++ b/eufy_sync/cli/updater.py
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
+import platform
import re
import shutil
import subprocess
@@ -9,6 +10,7 @@
import time
import urllib.request
+from eufy_sync import platform_support
from eufy_sync.cli import shared
@@ -65,12 +67,19 @@ def _parse(v: str) -> tuple:
if sys.stdin.isatty():
print(f"Update available: v{latest} (you have v{__version__}). Run: eufy-sync --update")
else:
- shared._notify("eufy-sync", f"Update available: v{latest}. Run: eufy-sync --update", command="eufy-sync --update")
+ platform_support.notify("eufy-sync", f"Update available: v{latest}. Run: eufy-sync --update", command="eufy-sync --update")
except Exception:
pass # never let update check break a sync
+def _quote_cmd(cmd: list[str]) -> str:
+ """Join an argv into one command string, double-quoting any element that
+ contains a space so a path like C:\\Program Files\\... survives as a single
+ argument when the string is handed to cmd.exe."""
+ return " ".join(f'"{part}"' if " " in part else part for part in cmd)
+
+
def _self_update() -> None:
"""Upgrade eufy-sync in place to the latest PyPI release.
@@ -90,7 +99,9 @@ def _self_update() -> None:
print(f"Unexpected version from PyPI ({latest!r}); update manually with pipx.")
return
- if "/uv/tools/" in sys.executable and shutil.which("uv"):
+ # Normalize backslashes to forward slashes so the marker matches on
+ # Windows uv installs (C:\...\uv\tools\...) as well as POSIX ones.
+ if "/uv/tools/" in sys.executable.replace("\\", "/") and shutil.which("uv"):
# Installed with `uv tool install`. Its venvs carry no pip, and pipx
# would create a second copy, so update through uv itself.
cmd = ["uv", "tool", "install", "--force", f"eufy-sync=={latest}"]
@@ -99,9 +110,26 @@ def _self_update() -> None:
else:
cmd = [sys.executable, "-m", "pip", "install", "--upgrade", f"eufy-sync=={latest}"]
+ if platform.system() == "Windows":
+ # A running eufy-sync.exe holds a lock on itself, so it cannot be
+ # overwritten in place. Hand the update to a fresh console that waits
+ # for this process to exit, runs the upgrade, and stays open (pause)
+ # so any failure is visible instead of vanishing with the window.
+ flags = getattr(subprocess, "CREATE_NEW_CONSOLE", 0) | getattr(
+ subprocess, "CREATE_NEW_PROCESS_GROUP", 0
+ )
+ inner = "timeout /t 2 /nobreak >nul & " + _quote_cmd(cmd) + " & echo. & pause"
+ subprocess.Popen(
+ ["cmd", "/c", "start", "eufy-sync update", "cmd", "/c", inner],
+ creationflags=flags,
+ )
+ print(f"Updating eufy-sync from v{__version__} to v{latest} in a new window...")
+ print("(Windows cannot replace a running program, so the update finishes there.)")
+ return
+
print(f"Updating eufy-sync from v{__version__} to v{latest}...")
result = subprocess.run(cmd)
if result.returncode == 0:
print(f"Updated to v{latest}.")
else:
- print("Update failed. Run manually: " + " ".join(cmd))
+ print("Update failed. Run manually: " + _quote_cmd(cmd))
diff --git a/eufy_sync/credentials.py b/eufy_sync/credentials.py
index f3a47ef..2f3c0a7 100644
--- a/eufy_sync/credentials.py
+++ b/eufy_sync/credentials.py
@@ -42,6 +42,14 @@
SERVICE_NAME = "eufy-garmin-sync"
VAULT_ACCOUNT = "vault"
+# Windows Credential Manager caps one entry at ~2,560 bytes stored as UTF-16,
+# roughly 1,280 characters. A vault larger than CHUNK_LIMIT characters is split
+# across numbered "vault:i" entries so set_password never fails on Windows.
+# MAX_CHUNKS bounds how far a save probes for stale leftover chunks to delete,
+# so a corrupt store can never make that scan run away.
+CHUNK_LIMIT = 1200
+MAX_CHUNKS = 40
+
CRED_FILE = Path.home() / ".garmin-sync" / "credentials.json"
@@ -136,15 +144,70 @@ def _load_vault_from_keychain() -> dict:
if raw is None:
return _empty_vault()
try:
- return _normalize_vault(json.loads(raw))
- except (json.JSONDecodeError, TypeError):
+ parsed = json.loads(raw)
+ # An oversized vault is stored as a header pointing at numbered chunks;
+ # reassemble the payload before normalizing. A missing chunk means the
+ # header outlived its payload, which is as unusable as malformed JSON.
+ if isinstance(parsed, dict) and "__chunks__" in parsed:
+ count = parsed["__chunks__"]
+ pieces = []
+ for i in range(1, count + 1):
+ try:
+ piece = keyring.get_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:{i}")
+ except Exception as e:
+ # A keyring failure mid-reassembly (locked, access denied) is
+ # the same unreadable-keychain condition as the initial read,
+ # not a missing chunk. Translate it so a partial read can
+ # never be saved back over the real vault. A chunk that
+ # returns None (below) is a genuinely missing chunk and keeps
+ # its malformed-vault handling.
+ raise RuntimeError(
+ "The system keychain could not be read (it may be locked or "
+ "access was denied). Unlock it and retry, or run: "
+ "eufy-sync --use-file-store"
+ ) from e
+ if piece is None:
+ raise ValueError("missing vault chunk")
+ pieces.append(piece)
+ parsed = json.loads("".join(pieces))
+ return _normalize_vault(parsed)
+ except (ValueError, TypeError):
logger.warning("Keychain vault contained malformed JSON; treating as empty")
return _empty_vault()
+def _delete_stale_chunks(start: int) -> None:
+ # A previous save may have used more chunks than this one. Delete numbered
+ # entries from `start` upward until the first gap, so a later read can never
+ # reassemble a stale tail. Bounded by MAX_CHUNKS.
+ import keyring
+ for i in range(start, MAX_CHUNKS + 1):
+ account = f"{VAULT_ACCOUNT}:{i}"
+ if keyring.get_password(SERVICE_NAME, account) is None:
+ break
+ try:
+ keyring.delete_password(SERVICE_NAME, account)
+ except Exception:
+ pass
+
+
def _save_vault_to_keychain(vault: dict) -> None:
import keyring
- keyring.set_password(SERVICE_NAME, VAULT_ACCOUNT, json.dumps(vault))
+ payload = json.dumps(vault)
+ if len(payload) <= CHUNK_LIMIT:
+ keyring.set_password(SERVICE_NAME, VAULT_ACCOUNT, payload)
+ _delete_stale_chunks(1)
+ return
+ chunks = [payload[i:i + CHUNK_LIMIT] for i in range(0, len(payload), CHUNK_LIMIT)]
+ # Chunks first, header last: a reader that races the write sees either the
+ # old vault or a complete new one, never a header pointing at a chunk that
+ # has not been written yet.
+ for i, chunk in enumerate(chunks, start=1):
+ keyring.set_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:{i}", chunk)
+ keyring.set_password(
+ SERVICE_NAME, VAULT_ACCOUNT, json.dumps({"__chunks__": len(chunks)})
+ )
+ _delete_stale_chunks(len(chunks) + 1)
def _load_vault_from_file() -> dict:
diff --git a/eufy_sync/garmin_auth.py b/eufy_sync/garmin_auth.py
index 19891f4..86bc189 100644
--- a/eufy_sync/garmin_auth.py
+++ b/eufy_sync/garmin_auth.py
@@ -15,7 +15,7 @@
import httpx
-from garminconnect import Garmin
+from garminconnect import Garmin, GarminConnectAuthenticationError
logger = logging.getLogger(__name__)
@@ -206,8 +206,21 @@ def _ensure_chromium() -> None:
)
+class GarminLoginCancelled(Exception):
+ """The user backed out of the interactive login (empty MFA code or Ctrl+C)."""
+
+
def _mfa_prompt() -> str:
- return input("Garmin MFA code (check your email): ").strip()
+ print("Garmin emailed a security code to your account address.")
+ print("No email after a minute? The stored password is likely wrong; press Enter to cancel.")
+ try:
+ code = input("Garmin MFA code: ").strip()
+ except (KeyboardInterrupt, EOFError):
+ print("")
+ raise GarminLoginCancelled("cancelled at the MFA prompt") from None
+ if not code:
+ raise GarminLoginCancelled("no MFA code entered")
+ return code
class GarminAuth:
@@ -238,7 +251,7 @@ def login(self, interactive: bool = True) -> Garmin:
if not interactive:
from eufy_sync.sync import PermanentSyncError
- raise PermanentSyncError("Garmin login needed; run: eufy-sync --reauth")
+ raise PermanentSyncError("Garmin login needed; run: eufy-sync --reauth garmin")
self._fresh_login(garmin)
self._save_token(garmin)
@@ -255,10 +268,24 @@ def force_reauth(self) -> Garmin:
return garmin
def _fresh_login(self, garmin: Garmin) -> None:
- """Try the browser-free login; fall back to the browser on any failure."""
+ """Try the browser-free login; fall back to the browser on any failure.
+
+ Two failures skip the fallback: a cancelled MFA prompt (the user chose
+ to stop) and definitively rejected credentials (the browser would
+ autofill the same bad password and fail minutes later)."""
+ from eufy_sync.sync import PermanentSyncError
try:
garmin.login()
return
+ except GarminLoginCancelled as e:
+ raise PermanentSyncError(
+ "Garmin login cancelled. If the MFA email never arrived, the "
+ "stored password is likely wrong; run: eufy-sync --update-password"
+ ) from e
+ except GarminConnectAuthenticationError as e:
+ raise PermanentSyncError(
+ "Garmin rejected the email or password. Run: eufy-sync --update-password"
+ ) from e
except Exception as e:
logger.warning(
"Browser-free Garmin login failed (%s); opening browser fallback", e
diff --git a/eufy_sync/platform_support/__init__.py b/eufy_sync/platform_support/__init__.py
new file mode 100644
index 0000000..b95ad29
--- /dev/null
+++ b/eufy_sync/platform_support/__init__.py
@@ -0,0 +1,62 @@
+"""OS-specific plumbing behind one small interface.
+
+Notifications and the scheduled-sync agent (install/uninstall/status/offer)
+differ per platform. The active implementation is chosen once from
+platform.system(): Darwin uses the macOS Launch Agent and osascript
+notifications; Windows uses a Task Scheduler entry; every other platform uses
+a no-op generic layer.
+"""
+from __future__ import annotations
+
+import platform
+
+_active = None
+
+
+def _select():
+ """Pick the implementation module for the current OS (uncached)."""
+ system = platform.system()
+ if system == "Darwin":
+ from eufy_sync.platform_support import macos as impl
+ elif system == "Windows":
+ from eufy_sync.platform_support import windows as impl
+ else:
+ from eufy_sync.platform_support import generic as impl
+ return impl
+
+
+def _impl():
+ """Return the active implementation module, selecting it once and caching
+ the choice. Tests override the selection by patching this function."""
+ global _active
+ if _active is None:
+ _active = _select()
+ return _active
+
+
+def notify(title: str, message: str, command: str | None = None) -> None:
+ _impl().notify(title, message, command)
+
+
+def install_agent() -> None:
+ _impl().install_agent()
+
+
+def uninstall_agent() -> None:
+ _impl().uninstall_agent()
+
+
+def offer_agent() -> None:
+ _impl().offer_agent()
+
+
+def agent_status() -> dict | None:
+ return _impl().agent_status()
+
+
+def agent_installed() -> bool:
+ return _impl().agent_installed()
+
+
+def purge_agent() -> None:
+ _impl().purge_agent()
diff --git a/eufy_sync/platform_support/generic.py b/eufy_sync/platform_support/generic.py
new file mode 100644
index 0000000..a0182e5
--- /dev/null
+++ b/eufy_sync/platform_support/generic.py
@@ -0,0 +1,40 @@
+"""Generic no-op layer for platforms without a managed scheduler.
+
+Covers every platform without a native implementation. Notifications are
+silently dropped; the scheduled-sync agent is not managed here, so --doctor
+skips the agent check and --uninstall skips the agent removal.
+"""
+from __future__ import annotations
+
+_UNMANAGED = (
+ "Auto-sync is not managed on this platform. See the Headless Linux "
+ "section of the README for a systemd timer recipe."
+)
+
+
+def notify(title: str, message: str, command: str | None = None) -> None:
+ pass
+
+
+def install_agent() -> None:
+ print(_UNMANAGED)
+
+
+def uninstall_agent() -> None:
+ print(_UNMANAGED)
+
+
+def offer_agent() -> None:
+ pass
+
+
+def agent_status() -> dict | None:
+ return None
+
+
+def agent_installed() -> bool:
+ return False
+
+
+def purge_agent() -> None:
+ pass
diff --git a/eufy_sync/platform_support/macos.py b/eufy_sync/platform_support/macos.py
new file mode 100644
index 0000000..e8e2450
--- /dev/null
+++ b/eufy_sync/platform_support/macos.py
@@ -0,0 +1,268 @@
+"""macOS implementation: osascript/terminal-notifier notifications and the
+Launch Agent that runs the scheduled sync.
+
+This is a move of the code that lived in cli/shared.py, cli/maintenance.py,
+and cli/doctor.py; the behavior is unchanged. The public entry points
+(notify, install_agent, uninstall_agent, offer_agent, agent_status,
+agent_installed, purge_agent) are what the platform_support package dispatches
+to; the leading-underscore helpers are the original functions, moved intact.
+"""
+from __future__ import annotations
+
+import json
+import os
+import platform
+import shlex
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+from eufy_sync.cli import shared
+
+LAUNCH_AGENT_LABEL = "com.sturimcode.eufy-garmin-sync"
+LAUNCH_AGENT_PATH = Path.home() / "Library" / "LaunchAgents" / f"{LAUNCH_AGENT_LABEL}.plist"
+
+# The Launch Agent points at this wrapper script, not the pipx/uv binary,
+# so the registered executable's bytes stay stable across updates (macOS
+# only re-announces "can run in the background" when they change). macOS
+# labels the background-item notification by this filename, so it is named
+# recognizably rather than something opaque. LEGACY_LAUNCH_WRAPPER_NAME is
+# the pre-1.7.17 name, cleaned up on the next --install-agent.
+LAUNCH_WRAPPER_NAME = "eufy-sync-agent"
+LEGACY_LAUNCH_WRAPPER_NAME = "run-sync.sh"
+
+
+def _find_terminal_notifier() -> str | None:
+ """Locate terminal-notifier (optional Homebrew tool). Checked beyond
+ PATH because launchd runs with a minimal PATH that excludes Homebrew."""
+ found = shutil.which("terminal-notifier")
+ if found:
+ return found
+ for candidate in ("/opt/homebrew/bin/terminal-notifier", "/usr/local/bin/terminal-notifier"):
+ if os.path.exists(candidate):
+ return candidate
+ return None
+
+
+def _notify(title: str, message: str, command: str | None = None) -> None:
+ """Send a macOS notification. Fails silently on other platforms.
+
+ When a fix command is given and terminal-notifier is installed,
+ clicking the notification opens a Terminal window running the command,
+ so the user lands in the interactive prompts instead of Script Editor
+ (plain osascript notifications belong to Script Editor and clicking
+ them just launches it).
+ """
+ try:
+ if command:
+ notifier = _find_terminal_notifier()
+ if notifier:
+ do_script = f'tell application "Terminal" to do script "{command}"'
+ activate = 'tell application "Terminal" to activate'
+ subprocess.run(
+ [
+ notifier,
+ "-title", title,
+ "-message", message,
+ "-execute", f"osascript -e {shlex.quote(do_script)} -e {shlex.quote(activate)}",
+ ],
+ capture_output=True,
+ timeout=5,
+ )
+ return
+ safe_title = json.dumps(title)
+ safe_msg = json.dumps(message)
+ subprocess.run(
+ ["osascript", "-e", f'display notification {safe_msg} with title {safe_title}'],
+ capture_output=True,
+ timeout=5,
+ )
+ except Exception:
+ pass
+
+
+def _write_run_script(binary_path: str) -> Path:
+ """Write the stable wrapper script the Launch Agent runs.
+
+ macOS re-announces "can run in the background" whenever a registered
+ background item's executable changes identity, and pipx/uv replace the
+ binary on every update. The agent therefore points at this script, whose
+ bytes never change across updates, so the announcement fires once, not
+ once per release. Skipping the rewrite when content is unchanged is what
+ keeps the file's identity stable. The filename is what macOS shows in that
+ announcement, so it is a recognizable "eufy-sync-agent", not an opaque one.
+ """
+ script_path = shared.DATA_DIR / LAUNCH_WRAPPER_NAME
+ content = f'#!/bin/sh\nexec "{binary_path}" --headless\n'
+ if script_path.exists() and script_path.read_text() == content:
+ return script_path
+ script_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+ script_path.write_text(content)
+ script_path.chmod(0o755)
+ return script_path
+
+
+def _generate_plist(program_path: str) -> str:
+ """Generate a Launch Agent plist that runs the given program every 4 hours."""
+ log_path = str(shared.LOG_FILE)
+ return f"""
+
+
+
+ Label
+ {LAUNCH_AGENT_LABEL}
+
+ ProgramArguments
+
+ {program_path}
+
+
+ StartInterval
+ 14400
+
+ RunAtLoad
+
+
+ StandardOutPath
+ {log_path}
+ StandardErrorPath
+ {log_path}
+
+
+"""
+
+
+def _install_launch_agent() -> None:
+ """Install the macOS Launch Agent for automatic sync."""
+ if platform.system() != "Darwin":
+ print("Auto-sync is only supported on macOS.")
+ return
+
+ binary = shutil.which("eufy-sync")
+ if not binary:
+ print("Warning: could not find eufy-sync on PATH. Skipping auto-sync setup.")
+ return
+
+ already_installed = LAUNCH_AGENT_PATH.exists()
+
+ # Ensure the log directory exists with restricted permissions
+ shared.DATA_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
+
+ wrapper = _write_run_script(binary)
+
+ # Drop the pre-1.7.17 wrapper name so it does not linger as an orphan next
+ # to the new one.
+ legacy_wrapper = shared.DATA_DIR / LEGACY_LAUNCH_WRAPPER_NAME
+ if legacy_wrapper.exists():
+ legacy_wrapper.unlink()
+
+ LAUNCH_AGENT_PATH.parent.mkdir(parents=True, exist_ok=True)
+ LAUNCH_AGENT_PATH.write_text(_generate_plist(str(wrapper)))
+
+ # Unload first in case an old version is loaded
+ subprocess.run(
+ ["launchctl", "unload", str(LAUNCH_AGENT_PATH)],
+ capture_output=True,
+ )
+ subprocess.run(
+ ["launchctl", "load", str(LAUNCH_AGENT_PATH)],
+ capture_output=True,
+ )
+
+ if already_installed:
+ print(f"Launch Agent already installed (reloaded). Logs: {shared.LOG_FILE}")
+ else:
+ print(f"Automatic sync installed. Logs: {shared.LOG_FILE}")
+
+
+def _offer_launch_agent() -> None:
+ """Offer to install a macOS Launch Agent after first-run setup."""
+ if platform.system() != "Darwin":
+ return
+ if not sys.stdin.isatty():
+ return
+
+ print("")
+ answer = input("Set up automatic sync every 4 hours? [y/N] ").strip()
+ if not answer.lower().startswith("y"):
+ return
+
+ _install_launch_agent()
+
+
+def _uninstall_launch_agent() -> None:
+ """Remove the macOS Launch Agent."""
+ if not LAUNCH_AGENT_PATH.exists():
+ print("No Launch Agent installed.")
+ return
+
+ subprocess.run(
+ ["launchctl", "unload", str(LAUNCH_AGENT_PATH)],
+ capture_output=True,
+ )
+ LAUNCH_AGENT_PATH.unlink()
+ print("Launch Agent removed. Auto-sync disabled.")
+
+
+def agent_status() -> dict | None:
+ """Report the Launch Agent's health for --doctor.
+
+ Reshaped from the old doctor._check_launch_agent: it returns the same
+ status/detail/fix strings as a dict instead of calling report().
+ """
+ try:
+ if not LAUNCH_AGENT_PATH.exists():
+ return {"status": "WARN", "label": "launch agent", "detail": "not installed", "fix": "eufy-sync --install-agent"}
+
+ content = LAUNCH_AGENT_PATH.read_text()
+ wrapper_name = LAUNCH_WRAPPER_NAME
+ if wrapper_name not in content:
+ return {
+ "status": "WARN",
+ "label": "launch agent",
+ "detail": "outdated registration (re-announces on every update)",
+ "fix": "eufy-sync --install-agent",
+ }
+
+ result = subprocess.run(
+ ["launchctl", "list"],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
+ if LAUNCH_AGENT_LABEL not in (result.stdout or ""):
+ return {"status": "WARN", "label": "launch agent", "detail": "installed but not loaded", "fix": "eufy-sync --install-agent"}
+
+ return {"status": "PASS", "label": "launch agent", "detail": "loaded, runs every 4h", "fix": None}
+ except Exception as e:
+ return {"status": "WARN", "label": "launch agent", "detail": f"could not check ({e})", "fix": "eufy-sync --install-agent"}
+
+
+def agent_installed() -> bool:
+ """Whether the Launch Agent plist is present on disk."""
+ return LAUNCH_AGENT_PATH.exists()
+
+
+def purge_agent() -> None:
+ """Unload and delete the Launch Agent without printing. Used by the full
+ --uninstall, which reports its own summary."""
+ if LAUNCH_AGENT_PATH.exists():
+ subprocess.run(["launchctl", "unload", str(LAUNCH_AGENT_PATH)], capture_output=True)
+ LAUNCH_AGENT_PATH.unlink()
+
+
+def notify(title: str, message: str, command: str | None = None) -> None:
+ _notify(title, message, command)
+
+
+def install_agent() -> None:
+ _install_launch_agent()
+
+
+def uninstall_agent() -> None:
+ _uninstall_launch_agent()
+
+
+def offer_agent() -> None:
+ _offer_launch_agent()
diff --git a/eufy_sync/platform_support/windows.py b/eufy_sync/platform_support/windows.py
new file mode 100644
index 0000000..3a76370
--- /dev/null
+++ b/eufy_sync/platform_support/windows.py
@@ -0,0 +1,239 @@
+"""Windows implementation: a per-user scheduled task that runs the sync in a
+hidden window, plus the --doctor status check.
+
+Task Scheduler runs the sync through a small VBScript wrapper rather than the
+pipx/uv binary directly. schtasks would flash a console window every four hours;
+launching the sync from VBScript with WScript.Shell.Run and a window style of 0
+keeps it invisible. The wrapper also points the task at a filename whose bytes
+stay stable across updates, so a pipx/uv upgrade that swaps the underlying
+binary does not require re-registering the task.
+
+The public entry points (notify, install_agent, uninstall_agent, offer_agent,
+agent_status, agent_installed, purge_agent) are what the platform_support
+package dispatches to. notify raises a native toast through PowerShell WinRT,
+fire-and-forget.
+"""
+from __future__ import annotations
+
+import base64
+import platform
+import shutil
+import subprocess
+import sys
+
+from eufy_sync.cli import shared
+
+TASK_NAME = "eufy-sync"
+WRAPPER_NAME = "eufy-sync-agent.vbs"
+
+# PowerShell script that raises a native toast through the WinRT notification
+# APIs. The title and message are XML-escaped and interpolated with .format(),
+# so the AppId GUID braces are doubled to survive that pass as single literal
+# braces. The AppId points at the built-in Windows PowerShell shortcut so the
+# toast has a registered source and actually shows (an unregistered AppId is
+# silently dropped).
+_PS_TOAST = """\
+[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType=WindowsRuntime] | Out-Null
+[Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument, ContentType=WindowsRuntime] | Out-Null
+$xml = New-Object Windows.Data.Xml.Dom.XmlDocument
+$xml.LoadXml(@'
+{title}{message}
+'@)
+$appid = '{{1AC14E77-02E7-4E5D-B744-2EB1AE5198B7}}\\WindowsPowerShell\\v1.0\\powershell.exe'
+[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier($appid).Show(
+ [Windows.UI.Notifications.ToastNotification]::new($xml))
+"""
+
+
+def _wrapper_content(binary_path: str) -> str:
+ """Build the VBScript that runs the sync with no visible console.
+
+ WScript.Shell.Run takes one command string and a window style; style 0
+ hides the window. The command is a `cmd /c` line so its stdout and stderr
+ can be appended to the log with `>>`. VBScript has no backslash escaping:
+ a literal double quote inside a string is written as two double quotes.
+ The inner command already contains quotes (around the binary path and the
+ log path), so every one of them is doubled before it is embedded in the
+ Run string literal.
+ """
+ log = str(shared.LOG_FILE)
+ inner = f'cmd /c ""{binary_path}" --headless >> "{log}" 2>&1"'
+ escaped = inner.replace('"', '""')
+ return (
+ 'Set shell = CreateObject("WScript.Shell")\r\n'
+ f'shell.Run "{escaped}", 0, False\r\n'
+ )
+
+
+def _write_wrapper(binary_path: str):
+ """Write the hidden-window wrapper, but only when its content would change.
+
+ The task registration points at this file, so leaving the bytes untouched
+ across updates avoids needless rewrites (the pipx/uv shim path is stable, so
+ the content normally never changes anyway).
+ """
+ wrapper_path = shared.DATA_DIR / WRAPPER_NAME
+ # Write and compare as raw bytes: text mode would translate the CRLF line
+ # endings (dropping the carriage returns on read, and doubling them on write
+ # under Windows), so the round-trip would never match and the file would be
+ # rewritten on every run.
+ content = _wrapper_content(binary_path).encode("utf-8")
+ if wrapper_path.exists() and wrapper_path.read_bytes() == content:
+ return wrapper_path
+ wrapper_path.parent.mkdir(parents=True, exist_ok=True)
+ wrapper_path.write_bytes(content)
+ return wrapper_path
+
+
+def _install_scheduled_task() -> None:
+ """Register the per-user scheduled task for automatic sync."""
+ binary = shutil.which("eufy-sync")
+ if not binary:
+ print("Warning: could not find eufy-sync on PATH. Skipping auto-sync setup.")
+ return
+
+ shared.DATA_DIR.mkdir(parents=True, exist_ok=True)
+ wrapper_path = _write_wrapper(binary)
+
+ result = subprocess.run(
+ ["schtasks", "/Create", "/F", "/TN", TASK_NAME, "/SC", "HOURLY", "/MO", "4",
+ "/TR", f'wscript.exe "{wrapper_path}"'],
+ capture_output=True, text=True, timeout=15,
+ )
+
+ if result.returncode == 0:
+ print(f"Automatic sync installed (every 4 hours). Logs: {shared.LOG_FILE}")
+ else:
+ print(result.stderr.strip() or "Could not register the scheduled task.")
+ # Match the /TR value the code passes programmatically: the wrapper path
+ # is quoted inside the /TR argument, and those inner quotes are escaped
+ # with backslashes so a cmd.exe paste keeps the path (which may contain
+ # spaces) as a single argument.
+ print(
+ "Register it manually with: "
+ f'schtasks /Create /F /TN {TASK_NAME} /SC HOURLY /MO 4 '
+ f'/TR "wscript.exe \\"{wrapper_path}\\""'
+ )
+
+
+def _uninstall_scheduled_task() -> None:
+ """Remove the scheduled task and its wrapper."""
+ result = subprocess.run(
+ ["schtasks", "/Delete", "/F", "/TN", TASK_NAME],
+ capture_output=True, text=True, timeout=15,
+ )
+
+ # schtasks returns nonzero (a "cannot find" error) when nothing is
+ # registered; treat that as "nothing to remove" rather than a failure.
+ if result.returncode != 0:
+ print("No scheduled task installed.")
+ else:
+ print("Scheduled task removed. Auto-sync disabled.")
+
+ wrapper_path = shared.DATA_DIR / WRAPPER_NAME
+ if wrapper_path.exists():
+ wrapper_path.unlink()
+
+
+def _offer_scheduled_task() -> None:
+ """Offer to install the scheduled task after first-run setup."""
+ if platform.system() != "Windows":
+ return
+ if not sys.stdin.isatty():
+ return
+
+ print("")
+ answer = input("Set up automatic sync every 4 hours? [y/N] ").strip()
+ if not answer.lower().startswith("y"):
+ return
+
+ _install_scheduled_task()
+
+
+def agent_status() -> dict | None:
+ """Report the scheduled task's health for --doctor.
+
+ Three states: the wrapper missing or the task query failing means it is not
+ installed; the wrapper present but out of date with the current binary means
+ an outdated registration; both good means it is installed and scheduled.
+ """
+ try:
+ wrapper_path = shared.DATA_DIR / WRAPPER_NAME
+ if not wrapper_path.exists():
+ return {"status": "WARN", "label": "scheduled task", "detail": "not installed", "fix": "eufy-sync --install-agent"}
+
+ result = subprocess.run(
+ ["schtasks", "/Query", "/TN", TASK_NAME],
+ capture_output=True, text=True, timeout=5,
+ )
+ if result.returncode != 0:
+ return {"status": "WARN", "label": "scheduled task", "detail": "not installed", "fix": "eufy-sync --install-agent"}
+
+ binary = shutil.which("eufy-sync")
+ if binary and wrapper_path.read_bytes() != _wrapper_content(binary).encode("utf-8"):
+ return {"status": "WARN", "label": "scheduled task", "detail": "outdated registration", "fix": "eufy-sync --install-agent"}
+
+ return {"status": "PASS", "label": "scheduled task", "detail": "installed, runs every 4h", "fix": None}
+ except Exception as e:
+ return {"status": "WARN", "label": "scheduled task", "detail": f"could not check ({e})", "fix": "eufy-sync --install-agent"}
+
+
+def agent_installed() -> bool:
+ """Whether the scheduled task's wrapper file is present on disk."""
+ return (shared.DATA_DIR / WRAPPER_NAME).exists()
+
+
+def purge_agent() -> None:
+ """Remove the scheduled task and wrapper without printing. Used by the full
+ --uninstall, which reports its own summary."""
+ wrapper_path = shared.DATA_DIR / WRAPPER_NAME
+ if wrapper_path.exists():
+ subprocess.run(["schtasks", "/Delete", "/F", "/TN", TASK_NAME], capture_output=True)
+ wrapper_path.unlink()
+
+
+def notify(title: str, message: str, command: str | None = None) -> None:
+ """Send a native toast through PowerShell WinRT. Fails silently.
+
+ A notification is a courtesy, never the point of the run, so anything that
+ goes wrong here (no powershell, WinRT unavailable, a slow shell) must not
+ turn a completed sync into a crash. Every failure is swallowed, and the
+ call is bounded by a timeout so it cannot hang the process.
+
+ The title and message are XML-escaped and the whole script is handed to
+ powershell as a base64 -EncodedCommand (UTF-16-LE, the encoding PowerShell
+ expects), so no shell quoting layer can misread the text. command is
+ accepted for interface parity but ignored: v1 toasts are not clickable, and
+ call sites already put the fix command in the message text.
+ """
+ try:
+ from xml.sax.saxutils import escape
+ # Flatten all whitespace runs (including CR/LF) to single spaces before
+ # the XML escape. The text is embedded in a PowerShell here-string, which
+ # terminates on any line starting with '@, so a newline followed by '@
+ # in the message (raw exception text reaches here) would close the
+ # here-string early and hand the remainder to PowerShell as code.
+ # Newlines must never reach the embedded XML.
+ title = " ".join(title.split())
+ message = " ".join(message.split())
+ script = _PS_TOAST.format(title=escape(title), message=escape(message))
+ encoded = base64.b64encode(script.encode("utf-16-le")).decode("ascii")
+ subprocess.run(
+ ["powershell", "-NoProfile", "-NonInteractive", "-WindowStyle", "Hidden",
+ "-EncodedCommand", encoded],
+ capture_output=True, timeout=10,
+ )
+ except Exception:
+ pass
+
+
+def install_agent() -> None:
+ _install_scheduled_task()
+
+
+def uninstall_agent() -> None:
+ _uninstall_scheduled_task()
+
+
+def offer_agent() -> None:
+ _offer_scheduled_task()
diff --git a/pyproject.toml b/pyproject.toml
index c137e55..6cc0716 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "eufy-sync"
-version = "1.7.22"
+version = "1.8.0"
description = "Sync Eufy smart scale body composition data to Garmin Connect and Strava"
readme = "README.md"
license = "MIT"
@@ -24,6 +24,7 @@ classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: End Users/Desktop",
"Operating System :: MacOS",
+ "Operating System :: Microsoft :: Windows",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
diff --git a/tests/conftest.py b/tests/conftest.py
index b03fedc..f28ed01 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -25,7 +25,7 @@ def _hermetic_machine(tmp_path, monkeypatch):
monkeypatch.setattr("eufy_sync.cli.shared.DEFAULT_DB", data_dir / "state.db")
monkeypatch.setattr("eufy_sync.cli.shared.LOG_FILE", data_dir / "sync.log")
monkeypatch.setattr(
- "eufy_sync.cli.shared.LAUNCH_AGENT_PATH",
+ "eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH",
tmp_path / "LaunchAgents" / "com.sturimcode.eufy-garmin-sync.plist",
)
# setup.py captures shared.DATA_DIR at import time into this module-level
@@ -57,11 +57,11 @@ def delete_password(service, account):
@pytest.fixture(autouse=True)
def _mute_notifications(monkeypatch):
- """Stub the macOS notifier for every test.
+ """Stub the notifier for every test.
- _notify shells out to osascript, so an unmocked call from any code path
- fires a real notification on the machine running the suite. Tests that
- assert on notifications patch eufy_sync.cli.shared._notify themselves; that
- patch layers over this stub and restores it on exit.
+ notify shells out to osascript on macOS, so an unmocked call from any code
+ path fires a real notification on the machine running the suite. Tests that
+ assert on notifications patch eufy_sync.platform_support.notify themselves;
+ that patch layers over this stub and restores it on exit.
"""
- monkeypatch.setattr("eufy_sync.cli.shared._notify", MagicMock())
+ monkeypatch.setattr("eufy_sync.platform_support.notify", MagicMock())
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 1b6ba1e..f11ef97 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -7,9 +7,9 @@
import pytest
import yaml
-from eufy_sync.cli.shared import _write_config, LAUNCH_AGENT_LABEL, LAUNCH_AGENT_PATH
+from eufy_sync.cli.shared import _write_config
+from eufy_sync.platform_support.macos import LAUNCH_AGENT_LABEL, LAUNCH_AGENT_PATH, _generate_plist
from eufy_sync.cli.maintenance import (
- _generate_plist,
_install_launch_agent,
_uninstall,
_uninstall_launch_agent,
@@ -17,6 +17,46 @@
)
+@pytest.fixture
+def pin_macos_impl(monkeypatch):
+ """Pin the launch-agent dispatch to the macOS implementation.
+
+ maintenance delegates through platform_support._impl(), which resolves the
+ active module from platform.system(). On a Windows CI runner these
+ macOS-specific tests would otherwise drive the real Windows schtasks path;
+ pinning _active (the way test_doctor.py does) keeps them on the macOS module
+ regardless of host OS, so the eufy_sync.platform_support.macos.* patches
+ below actually take effect."""
+ from eufy_sync import platform_support
+ from eufy_sync.platform_support import macos
+ monkeypatch.setattr(platform_support, "_active", macos)
+
+
+def test_main_ctrl_c_exits_cleanly(capsys):
+ from eufy_sync.cli import app
+ with patch.object(app, "_main", side_effect=KeyboardInterrupt):
+ with pytest.raises(SystemExit) as exc_info:
+ app.main()
+ assert exc_info.value.code == 130
+ assert "Cancelled" in capsys.readouterr().out
+
+
+def test_update_password_path_configures_logging(monkeypatch, tmp_path: Path):
+ # --update-password reaches a Garmin login, so main() must configure
+ # logging first or garminconnect's per-strategy 429 warnings leak through
+ # logging's last-resort handler.
+ import sys as _sys
+ from eufy_sync.cli import app, shared
+ calls = []
+ monkeypatch.setattr(shared, "_configure_logging", lambda v: calls.append(v))
+ missing_config = tmp_path / "missing.yaml"
+ monkeypatch.setattr(_sys, "argv",
+ ["eufy-sync", "--update-password", "--config", str(missing_config)])
+ with pytest.raises(SystemExit):
+ app.main()
+ assert calls == [False]
+
+
def test_write_config_creates_file_with_restricted_permissions(tmp_path: Path):
config_path = tmp_path / "subdir" / "config.yaml"
config = {"users": [{"name": "test"}]}
@@ -24,9 +64,11 @@ def test_write_config_creates_file_with_restricted_permissions(tmp_path: Path):
_write_config(config_path, config)
assert config_path.exists()
- # File should be 600 (owner read/write only)
- mode = oct(config_path.stat().st_mode)[-3:]
- assert mode == "600"
+ # File should be 600 (owner read/write only). Windows reports 666/777
+ # regardless of the mode passed, so skip the POSIX-mode check there.
+ if os.name != "nt":
+ mode = oct(config_path.stat().st_mode)[-3:]
+ assert mode == "600"
# Content should be valid YAML
with open(config_path) as f:
@@ -38,8 +80,10 @@ def test_write_config_parent_directory_is_restricted(tmp_path: Path):
config_path = tmp_path / "secure_dir" / "config.yaml"
_write_config(config_path, {"test": True})
- parent_mode = oct(config_path.parent.stat().st_mode)[-3:]
- assert parent_mode == "700"
+ # POSIX modes only; Windows does not honor them.
+ if os.name != "nt":
+ parent_mode = oct(config_path.parent.stat().st_mode)[-3:]
+ assert parent_mode == "700"
def test_write_config_overwrites_existing(tmp_path: Path):
@@ -68,21 +112,23 @@ def test_generate_plist_points_at_given_program():
def test_write_run_script_creates_executable_wrapper(tmp_path):
- from eufy_sync.cli.maintenance import _write_run_script
+ from eufy_sync.platform_support.macos import _write_run_script
with patch("eufy_sync.cli.shared.DATA_DIR", tmp_path):
script = _write_run_script("/home/user/.local/bin/eufy-sync")
assert script == tmp_path / "eufy-sync-agent"
content = script.read_text()
assert content.startswith("#!/bin/sh")
assert 'exec "/home/user/.local/bin/eufy-sync" --headless' in content
- assert oct(script.stat().st_mode)[-3:] == "755"
+ # POSIX modes only; Windows does not honor them.
+ if os.name != "nt":
+ assert oct(script.stat().st_mode)[-3:] == "755"
def test_write_run_script_is_byte_stable_across_installs(tmp_path):
# macOS re-announces background items when the registered file changes;
# a second install with the same binary must not rewrite the script.
import os
- from eufy_sync.cli.maintenance import _write_run_script
+ from eufy_sync.platform_support.macos import _write_run_script
with patch("eufy_sync.cli.shared.DATA_DIR", tmp_path):
first = _write_run_script("/home/user/.local/bin/eufy-sync")
mtime_before = os.stat(first).st_mtime_ns
@@ -93,15 +139,19 @@ def test_write_run_script_is_byte_stable_across_installs(tmp_path):
def test_generate_plist_contains_log_path():
+ # The plist embeds str(shared.LOG_FILE), which conftest points at a tmp
+ # path; assert against that actual path rather than a hardcoded
+ # forward-slash literal, which would not match Windows separators.
+ from eufy_sync.cli import shared
plist = _generate_plist("/any/path")
- assert ".garmin-sync/sync.log" in plist
+ assert str(shared.LOG_FILE) in plist
-@patch("eufy_sync.cli.maintenance.subprocess.run")
-@patch("eufy_sync.cli.maintenance.shutil.which", return_value="/home/user/.local/bin/eufy-sync")
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-def test_install_launch_agent_writes_plist_and_loads(mock_path, mock_system, mock_which, mock_run, tmp_path):
+@patch("eufy_sync.platform_support.macos.subprocess.run")
+@patch("eufy_sync.platform_support.macos.shutil.which", return_value="/home/user/.local/bin/eufy-sync")
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+def test_install_launch_agent_writes_plist_and_loads(mock_path, mock_system, mock_which, mock_run, tmp_path, pin_macos_impl):
mock_path.parent.mkdir = MagicMock()
mock_path.write_text = MagicMock()
@@ -124,11 +174,11 @@ def test_install_launch_agent_writes_plist_and_loads(mock_path, mock_system, moc
assert "load" in mock_run.call_args_list[1][0][0]
-@patch("eufy_sync.cli.maintenance.subprocess.run")
-@patch("eufy_sync.cli.maintenance.shutil.which", return_value="/home/user/.local/bin/eufy-sync")
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-def test_install_launch_agent_removes_legacy_wrapper(mock_path, mock_system, mock_which, mock_run, tmp_path):
+@patch("eufy_sync.platform_support.macos.subprocess.run")
+@patch("eufy_sync.platform_support.macos.shutil.which", return_value="/home/user/.local/bin/eufy-sync")
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+def test_install_launch_agent_removes_legacy_wrapper(mock_path, mock_system, mock_which, mock_run, tmp_path, pin_macos_impl):
"""Re-installing must delete the pre-1.7.17 run-sync.sh wrapper so it does
not linger as an orphan next to the new eufy-sync-agent script."""
mock_path.parent.mkdir = MagicMock()
@@ -145,22 +195,22 @@ def test_install_launch_agent_removes_legacy_wrapper(mock_path, mock_system, moc
assert (tmp_path / "eufy-sync-agent").exists()
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Linux")
-def test_install_launch_agent_skips_on_linux(mock_system, capsys):
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Linux")
+def test_install_launch_agent_skips_on_linux(mock_system, capsys, pin_macos_impl):
_install_launch_agent()
assert "only supported on macOS" in capsys.readouterr().out
-@patch("eufy_sync.cli.maintenance.shutil.which", return_value=None)
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
-def test_install_launch_agent_warns_if_binary_not_found(mock_system, mock_which, capsys):
+@patch("eufy_sync.platform_support.macos.shutil.which", return_value=None)
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
+def test_install_launch_agent_warns_if_binary_not_found(mock_system, mock_which, capsys, pin_macos_impl):
_install_launch_agent()
assert "could not find eufy-sync" in capsys.readouterr().out
-@patch("eufy_sync.cli.maintenance.subprocess.run")
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-def test_uninstall_launch_agent_removes_plist(mock_path, mock_run):
+@patch("eufy_sync.platform_support.macos.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+def test_uninstall_launch_agent_removes_plist(mock_path, mock_run, pin_macos_impl):
mock_path.exists.return_value = True
mock_path.unlink = MagicMock()
@@ -171,8 +221,8 @@ def test_uninstall_launch_agent_removes_plist(mock_path, mock_run):
mock_path.unlink.assert_called_once()
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-def test_uninstall_launch_agent_noop_if_not_installed(mock_path, capsys):
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+def test_uninstall_launch_agent_noop_if_not_installed(mock_path, capsys, pin_macos_impl):
mock_path.exists.return_value = False
_uninstall_launch_agent()
@@ -180,11 +230,11 @@ def test_uninstall_launch_agent_noop_if_not_installed(mock_path, capsys):
assert "No Launch Agent installed" in capsys.readouterr().out
-@patch("eufy_sync.cli.maintenance._install_launch_agent")
+@patch("eufy_sync.platform_support.macos._install_launch_agent")
@patch("builtins.input", return_value="y")
-@patch("eufy_sync.cli.maintenance.sys.stdin")
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
-def test_offer_launch_agent_installs_on_yes(mock_system, mock_stdin, mock_input, mock_install):
+@patch("eufy_sync.platform_support.macos.sys.stdin")
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
+def test_offer_launch_agent_installs_on_yes(mock_system, mock_stdin, mock_input, mock_install, pin_macos_impl):
mock_stdin.isatty.return_value = True
_offer_launch_agent()
@@ -192,10 +242,10 @@ def test_offer_launch_agent_installs_on_yes(mock_system, mock_stdin, mock_input,
mock_install.assert_called_once()
-@patch("eufy_sync.cli.maintenance._install_launch_agent")
+@patch("eufy_sync.platform_support.macos._install_launch_agent")
@patch("builtins.input", return_value="n")
-@patch("eufy_sync.cli.maintenance.sys.stdin")
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
+@patch("eufy_sync.platform_support.macos.sys.stdin")
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
def test_offer_launch_agent_skips_on_no(mock_system, mock_stdin, mock_input, mock_install):
mock_stdin.isatty.return_value = True
@@ -204,9 +254,9 @@ def test_offer_launch_agent_skips_on_no(mock_system, mock_stdin, mock_input, moc
mock_install.assert_not_called()
-@patch("eufy_sync.cli.maintenance._install_launch_agent")
-@patch("eufy_sync.cli.maintenance.sys.stdin")
-@patch("eufy_sync.cli.maintenance.platform.system", return_value="Darwin")
+@patch("eufy_sync.platform_support.macos._install_launch_agent")
+@patch("eufy_sync.platform_support.macos.sys.stdin")
+@patch("eufy_sync.platform_support.macos.platform.system", return_value="Darwin")
def test_offer_launch_agent_skips_non_interactive(mock_system, mock_stdin, mock_install):
mock_stdin.isatty.return_value = False
@@ -218,8 +268,8 @@ def test_offer_launch_agent_skips_non_interactive(mock_system, mock_stdin, mock_
# --- _uninstall keychain cleanup ---
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-@patch("eufy_sync.cli.maintenance.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+@patch("eufy_sync.platform_support.macos.subprocess.run")
@patch("eufy_sync.credentials._keyring_available", return_value=True)
@patch("eufy_sync.credentials.delete_token")
@patch("eufy_sync.credentials.delete_password")
@@ -253,8 +303,8 @@ def test_uninstall_clears_keychain_for_configured_user_name(
assert "elias:garmin" in deleted_accounts
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-@patch("eufy_sync.cli.maintenance.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+@patch("eufy_sync.platform_support.macos.subprocess.run")
@patch("eufy_sync.credentials._keyring_available", return_value=True)
@patch("eufy_sync.credentials.delete_password", side_effect=RuntimeError("keychain locked"))
@patch("eufy_sync.cli.maintenance.sys.stdin")
@@ -283,8 +333,8 @@ def test_uninstall_survives_locked_keychain(
assert "keychain" in out.lower()
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-@patch("eufy_sync.cli.maintenance.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+@patch("eufy_sync.platform_support.macos.subprocess.run")
@patch("eufy_sync.credentials._keyring_available", return_value=True)
@patch("eufy_sync.credentials.delete_token")
@patch("eufy_sync.credentials.delete_password")
@@ -313,8 +363,8 @@ def test_uninstall_names_uv_for_uv_installs(
assert "pipx uninstall" not in out
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-@patch("eufy_sync.cli.maintenance.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+@patch("eufy_sync.platform_support.macos.subprocess.run")
@patch("eufy_sync.credentials._keyring_available", return_value=True)
@patch("eufy_sync.credentials.delete_token")
@patch("eufy_sync.credentials.delete_password")
@@ -341,8 +391,8 @@ def test_uninstall_names_pipx_when_not_uv(
assert "pipx uninstall eufy-sync" in out
-@patch("eufy_sync.cli.shared.LAUNCH_AGENT_PATH")
-@patch("eufy_sync.cli.maintenance.subprocess.run")
+@patch("eufy_sync.platform_support.macos.LAUNCH_AGENT_PATH")
+@patch("eufy_sync.platform_support.macos.subprocess.run")
@patch("eufy_sync.credentials._keyring_available", return_value=True)
@patch("eufy_sync.credentials.delete_token")
@patch("eufy_sync.credentials.delete_password")
@@ -706,7 +756,7 @@ def test_upgrade_notice_file_is_hermetic(tmp_path):
assert str(setup.UPGRADE_NOTICE_FILE).startswith(str(tmp_path))
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
def test_headless_first_run_refuses_wizard(mock_notify, tmp_path):
"""A headless run with no config must never call input() - it should
print guidance, notify, and exit 1 instead of hanging in the wizard."""
@@ -729,7 +779,7 @@ def boom_input(*a, **k):
mock_notify.assert_called()
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
def test_startup_failure_before_harness_notifies_and_exits(mock_notify, tmp_path):
"""A load_config failure (e.g. missing keychain entry -> ValueError) that
happens before the sync try/except harness must still notify and exit 1,
@@ -752,7 +802,7 @@ def test_startup_failure_before_harness_notifies_and_exits(mock_notify, tmp_path
mock_notify.assert_called()
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
@@ -788,7 +838,7 @@ def fake_sync_user(user, state, **kwargs):
@patch("eufy_sync.cli.status._print_summary")
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
@@ -827,7 +877,7 @@ def run_once():
@patch("eufy_sync.cli.status._print_summary")
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
@@ -861,7 +911,7 @@ def run_with(side):
@patch("eufy_sync.cli.status._print_summary")
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
@@ -892,7 +942,7 @@ def fake_sync_user(user, state, **kwargs):
@patch("eufy_sync.cli.status._print_summary")
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
@@ -933,7 +983,7 @@ def fake_sync_user(user, state, **kwargs):
@patch("eufy_sync.cli.status._print_summary")
-@patch("eufy_sync.cli.shared._notify")
+@patch("eufy_sync.platform_support.notify")
@patch("eufy_sync.cli.updater._check_for_updates")
@patch("eufy_sync.cli.setup._show_upgrade_notice")
@patch("eufy_sync.cli.setup._migrate_config_passwords")
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 011d95e..e09e895 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
+import os
import stat
from pathlib import Path
from unittest.mock import patch
@@ -161,8 +162,10 @@ def test_file_backend_creates_0o600_file_with_json(no_keyring, cred_file):
store_password("default:eufy", "hunter2")
assert cred_file.exists()
- mode = stat.S_IMODE(cred_file.stat().st_mode)
- assert mode == 0o600
+ # POSIX modes only; Windows reports 666/777 regardless of the mode passed.
+ if os.name != "nt":
+ mode = stat.S_IMODE(cred_file.stat().st_mode)
+ assert mode == 0o600
on_disk = json.loads(cred_file.read_text())
assert on_disk["passwords"]["default:eufy"] == "hunter2"
@@ -233,8 +236,10 @@ def test_use_file_store_moves_vault_from_keychain_to_file(fake_keyring, cred_fil
use_file_store()
assert cred_file.exists()
- mode = stat.S_IMODE(cred_file.stat().st_mode)
- assert mode == 0o600
+ # POSIX modes only; Windows reports 666/777 regardless of the mode passed.
+ if os.name != "nt":
+ mode = stat.S_IMODE(cred_file.stat().st_mode)
+ assert mode == 0o600
on_disk = json.loads(cred_file.read_text())
assert on_disk["passwords"]["default:eufy"] == "pw1"
assert on_disk["tokens"]["garmin"] == {"di_token": "abc"}
@@ -295,8 +300,10 @@ def test_auto_fallback_creates_file_and_persists_token_with_no_keychain(no_keyri
store_token("eufy", {"access_token": "headless-tok"})
assert cred_file.exists()
- mode = stat.S_IMODE(cred_file.stat().st_mode)
- assert mode == 0o600
+ # POSIX modes only; Windows reports 666/777 regardless of the mode passed.
+ if os.name != "nt":
+ mode = stat.S_IMODE(cred_file.stat().st_mode)
+ assert mode == 0o600
assert get_token("eufy") == {"access_token": "headless-tok"}
@@ -716,3 +723,107 @@ def report(status, label, detail, fix=None):
doctor._check_keychain(report)
assert lines[0][0] == "PASS"
assert "file" in lines[0][2]
+
+
+# --- Vault chunking ----------------------------------------------------------
+#
+# Windows Credential Manager caps one entry at ~2,560 bytes (UTF-16). A vault
+# holding Garmin's two OAuth tokens plus Strava's can exceed that, so the
+# keychain backend splits an oversized vault across numbered entries. A vault
+# that fits keeps today's single-entry shape, so existing installs never see
+# a migration.
+
+from eufy_sync.credentials import (
+ CHUNK_LIMIT,
+ SERVICE_NAME,
+ VAULT_ACCOUNT,
+ store_token,
+ get_token,
+)
+
+
+def _big_token(size: int) -> dict:
+ return {"access_token": "x" * size}
+
+
+def test_small_vault_keeps_single_entry_shape(fake_keyring):
+ store_token("garmin", {"a": 1})
+ raw = fake_keyring.get_password(SERVICE_NAME, VAULT_ACCOUNT)
+ data = json.loads(raw)
+ assert "__chunks__" not in data
+ assert data["tokens"]["garmin"] == {"a": 1}
+ assert fake_keyring.get_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:1") is None
+
+
+def test_oversized_vault_chunks_and_round_trips(fake_keyring):
+ store_token("garmin", _big_token(3 * CHUNK_LIMIT))
+ header = json.loads(fake_keyring.get_password(SERVICE_NAME, VAULT_ACCOUNT))
+ n = header["__chunks__"]
+ assert n >= 3
+ for i in range(1, n + 1):
+ chunk = fake_keyring.get_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:{i}")
+ assert chunk is not None
+ assert len(chunk) <= CHUNK_LIMIT
+ assert get_token("garmin") == _big_token(3 * CHUNK_LIMIT)
+
+
+def test_shrinking_vault_deletes_stale_chunks(fake_keyring):
+ store_token("garmin", _big_token(3 * CHUNK_LIMIT))
+ store_token("garmin", {"a": 1}) # replaces the big token; vault fits again
+ assert get_token("garmin") == {"a": 1}
+ raw = fake_keyring.get_password(SERVICE_NAME, VAULT_ACCOUNT)
+ assert "__chunks__" not in json.loads(raw)
+ assert fake_keyring.get_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:1") is None
+
+
+def test_missing_chunk_reads_as_empty_vault(fake_keyring, caplog):
+ store_token("garmin", _big_token(3 * CHUNK_LIMIT))
+ fake_keyring.delete_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:2")
+ assert get_token("garmin") is None # malformed vault treated as empty
+
+
+def test_chunk_read_failure_raises_friendly_error(fake_keyring, cred_file, monkeypatch):
+ """A keyring exception while reassembling chunks (locked or access denied
+ partway through) is the same unreadable-keychain condition as a failed
+ initial read - it must surface the actionable RuntimeError, not a raw
+ backend exception, so a partial read can never be saved back over the real
+ vault. A genuinely missing chunk keeps its malformed-vault handling
+ (covered above)."""
+ store_token("garmin", _big_token(3 * CHUNK_LIMIT))
+
+ real_get = fake_keyring.get_password
+
+ def flaky(service, account):
+ # The header read (account "vault") succeeds; the first chunk read fails.
+ if account.startswith(f"{VAULT_ACCOUNT}:"):
+ raise OSError("keychain locked")
+ return real_get(service, account)
+
+ monkeypatch.setattr("keyring.get_password", flaky)
+
+ with pytest.raises(RuntimeError, match="could not be read"):
+ get_token("garmin")
+
+
+def test_vault_exactly_at_chunk_limit_stays_single_entry(fake_keyring):
+ """The chunking boundary is inclusive: a serialized vault whose length is
+ exactly CHUNK_LIMIT still fits in one entry (the split only triggers above
+ the limit). Compute the token size that lands the full payload exactly on
+ CHUNK_LIMIT and confirm no chunk entries are written."""
+ def payload_len(n: int) -> int:
+ vault = {"passwords": {}, "tokens": {"garmin": {"access_token": "x" * n}}}
+ return len(json.dumps(vault))
+
+ # Each extra character in the token string adds exactly one byte to the
+ # serialized JSON, so solve for the size directly.
+ n = 1 + (CHUNK_LIMIT - payload_len(1))
+ assert payload_len(n) == CHUNK_LIMIT
+
+ store_token("garmin", {"access_token": "x" * n})
+
+ raw = fake_keyring.get_password(SERVICE_NAME, VAULT_ACCOUNT)
+ assert len(raw) == CHUNK_LIMIT
+ data = json.loads(raw)
+ assert "__chunks__" not in data
+ assert fake_keyring.get_password(SERVICE_NAME, f"{VAULT_ACCOUNT}:1") is None
+ assert get_token("garmin") == {"access_token": "x" * n}
diff --git a/tests/test_doctor.py b/tests/test_doctor.py
index 3fe22e7..644f0d8 100644
--- a/tests/test_doctor.py
+++ b/tests/test_doctor.py
@@ -14,8 +14,10 @@
import pytest
import yaml
+from eufy_sync import platform_support
from eufy_sync.cli import doctor
from eufy_sync.config import AppConfig, EufyConfig, GarminConfig, StravaConfig, UserConfig
+from eufy_sync.platform_support import generic, macos
def _write_config(path: Path, *, garmin: bool = True, strava: bool = True, customer_id: str | None = "abc1234567890867f") -> None:
@@ -73,19 +75,19 @@ def _patch_all_pass(tmp_path, monkeypatch):
strava_client.token_status.return_value = {"state": "valid", "days_remaining": None, "hours_remaining": 5}
monkeypatch.setattr(doctor, "StravaClient", MagicMock(return_value=strava_client))
- monkeypatch.setattr(doctor.platform, "system", MagicMock(return_value="Darwin"))
+ monkeypatch.setattr(platform_support, "_active", macos)
wrapper = tmp_path / "eufy-sync-agent"
wrapper.write_text("#!/bin/sh\n")
- monkeypatch.setattr(doctor.shared, "LAUNCH_AGENT_PATH", tmp_path / "agent.plist")
+ monkeypatch.setattr(macos, "LAUNCH_AGENT_PATH", tmp_path / "agent.plist")
plist = f"""
ProgramArguments
{wrapper}
"""
- doctor.shared.LAUNCH_AGENT_PATH.write_text(plist)
+ macos.LAUNCH_AGENT_PATH.write_text(plist)
monkeypatch.setattr(
- doctor.subprocess, "run",
- MagicMock(return_value=MagicMock(stdout=doctor.shared.LAUNCH_AGENT_LABEL, returncode=0)),
+ macos.subprocess, "run",
+ MagicMock(return_value=MagicMock(stdout=macos.LAUNCH_AGENT_LABEL, returncode=0)),
)
state = MagicMock()
@@ -196,7 +198,7 @@ def test_launch_agent_pointing_at_raw_binary_warns_install_agent(tmp_path, monke
ProgramArguments
/Users/x/.local/bin/eufy-sync
"""
- doctor.shared.LAUNCH_AGENT_PATH.write_text(plist)
+ macos.LAUNCH_AGENT_PATH.write_text(plist)
code = doctor._run_doctor(config_path, db_path)
@@ -221,7 +223,7 @@ def test_launch_agent_pointing_at_legacy_wrapper_warns_install_agent(tmp_path, m
ProgramArguments
{legacy}
"""
- doctor.shared.LAUNCH_AGENT_PATH.write_text(plist)
+ macos.LAUNCH_AGENT_PATH.write_text(plist)
code = doctor._run_doctor(config_path, db_path)
@@ -395,7 +397,7 @@ def test_keychain_fallback_never_fails(tmp_path, monkeypatch, capsys):
def test_launch_agent_not_installed_warns(tmp_path, monkeypatch, capsys):
config_path, db_path = _patch_all_pass(tmp_path, monkeypatch)
- doctor.shared.LAUNCH_AGENT_PATH.unlink()
+ macos.LAUNCH_AGENT_PATH.unlink()
code = doctor._run_doctor(config_path, db_path)
@@ -409,7 +411,7 @@ def test_launch_agent_not_installed_warns(tmp_path, monkeypatch, capsys):
def test_launch_agent_not_loaded_warns(tmp_path, monkeypatch, capsys):
config_path, db_path = _patch_all_pass(tmp_path, monkeypatch)
- doctor.subprocess.run = MagicMock(return_value=MagicMock(stdout="", returncode=0))
+ macos.subprocess.run = MagicMock(return_value=MagicMock(stdout="", returncode=0))
code = doctor._run_doctor(config_path, db_path)
@@ -423,7 +425,7 @@ def test_launch_agent_not_loaded_warns(tmp_path, monkeypatch, capsys):
def test_launch_agent_skipped_on_non_macos(tmp_path, monkeypatch, capsys):
config_path, db_path = _patch_all_pass(tmp_path, monkeypatch)
- doctor.platform.system = MagicMock(return_value="Linux")
+ monkeypatch.setattr(platform_support, "_active", generic)
code = doctor._run_doctor(config_path, db_path)
diff --git a/tests/test_garmin_auth.py b/tests/test_garmin_auth.py
index e997e3c..d57ea3f 100644
--- a/tests/test_garmin_auth.py
+++ b/tests/test_garmin_auth.py
@@ -136,21 +136,61 @@ def test_login_falls_back_to_browser_when_curl_cffi_fails(monkeypatch):
assert result is fake
-def test_login_bad_credentials_fails_both_tiers(monkeypatch):
+def test_login_bad_credentials_skips_browser_and_names_fix(monkeypatch):
+ # Definitively rejected credentials must not open the browser fallback:
+ # it would autofill the same bad password and fail minutes later.
from garminconnect import GarminConnectAuthenticationError
auth = _auth()
monkeypatch.setattr(auth, "_load_token", lambda: None)
fake = MagicMock()
fake.login.side_effect = GarminConnectAuthenticationError("bad")
- monkeypatch.setattr("eufy_sync.garmin_auth._ensure_chromium", lambda: None)
+ browser_calls = []
+ monkeypatch.setattr("eufy_sync.garmin_auth.browser_login",
+ lambda e, p: browser_calls.append(1) or "ticket")
+ with patch("eufy_sync.garmin_auth.Garmin", return_value=fake):
+ with pytest.raises(PermanentSyncError) as exc_info:
+ auth.login(interactive=True)
+ assert "update-password" in str(exc_info.value)
+ assert browser_calls == []
- def _no_ticket(email, password):
- raise RuntimeError("no service ticket captured")
- monkeypatch.setattr("eufy_sync.garmin_auth.browser_login", _no_ticket)
+
+def test_login_mfa_cancel_skips_browser_and_names_fix(monkeypatch):
+ from eufy_sync.garmin_auth import GarminLoginCancelled
+ auth = _auth()
+ monkeypatch.setattr(auth, "_load_token", lambda: None)
+ fake = MagicMock()
+ fake.login.side_effect = GarminLoginCancelled("no MFA code entered")
+ browser_calls = []
+ monkeypatch.setattr("eufy_sync.garmin_auth.browser_login",
+ lambda e, p: browser_calls.append(1) or "ticket")
with patch("eufy_sync.garmin_auth.Garmin", return_value=fake):
with pytest.raises(PermanentSyncError) as exc_info:
auth.login(interactive=True)
assert "update-password" in str(exc_info.value)
+ assert browser_calls == []
+
+
+def test_mfa_prompt_empty_input_cancels(monkeypatch):
+ from eufy_sync.garmin_auth import GarminLoginCancelled, _mfa_prompt
+ monkeypatch.setattr("builtins.input", lambda prompt: "")
+ with pytest.raises(GarminLoginCancelled):
+ _mfa_prompt()
+
+
+def test_mfa_prompt_ctrl_c_cancels(monkeypatch):
+ from eufy_sync.garmin_auth import GarminLoginCancelled, _mfa_prompt
+
+ def _interrupt(prompt):
+ raise KeyboardInterrupt
+ monkeypatch.setattr("builtins.input", _interrupt)
+ with pytest.raises(GarminLoginCancelled):
+ _mfa_prompt()
+
+
+def test_mfa_prompt_returns_stripped_code(monkeypatch):
+ from eufy_sync.garmin_auth import _mfa_prompt
+ monkeypatch.setattr("builtins.input", lambda prompt: " 123456 ")
+ assert _mfa_prompt() == "123456"
def test_login_headless_never_opens_browser(monkeypatch):
diff --git a/tests/test_notify.py b/tests/test_notify.py
index 54b26b5..e6fe89c 100644
--- a/tests/test_notify.py
+++ b/tests/test_notify.py
@@ -5,19 +5,20 @@
from unittest.mock import MagicMock, patch
-from eufy_sync.cli import shared
+from eufy_sync.platform_support import macos
-# conftest's autouse _mute_notifications fixture replaces shared._notify with
-# a MagicMock so tests never fire real notifications. These tests exercise the
-# real function, captured here at import time, with subprocess.run mocked out.
-_real_notify = shared._notify
+# conftest's autouse _mute_notifications fixture replaces platform_support.notify
+# with a MagicMock so tests never fire real notifications. These tests exercise
+# the real macOS function, captured here at import time, with subprocess.run
+# mocked out.
+_real_notify = macos._notify
def _run_notify(which_return, command=None):
run = MagicMock()
- with patch("eufy_sync.cli.shared.shutil.which", return_value=which_return), \
- patch("eufy_sync.cli.shared.os.path.exists", return_value=False), \
- patch("eufy_sync.cli.shared.subprocess.run", run):
+ with patch("eufy_sync.platform_support.macos.shutil.which", return_value=which_return), \
+ patch("eufy_sync.platform_support.macos.os.path.exists", return_value=False), \
+ patch("eufy_sync.platform_support.macos.subprocess.run", run):
_real_notify("eufy-sync: re-login needed", "Run: eufy-sync --reauth garmin", command=command)
return run
@@ -52,10 +53,10 @@ def test_homebrew_path_is_checked_when_which_misses():
"""launchd runs with a minimal PATH that excludes Homebrew, so the
lookup must also try the standard install locations directly."""
run = MagicMock()
- with patch("eufy_sync.cli.shared.shutil.which", return_value=None), \
- patch("eufy_sync.cli.shared.os.path.exists",
+ with patch("eufy_sync.platform_support.macos.shutil.which", return_value=None), \
+ patch("eufy_sync.platform_support.macos.os.path.exists",
side_effect=lambda p: p == "/opt/homebrew/bin/terminal-notifier"), \
- patch("eufy_sync.cli.shared.subprocess.run", run):
+ patch("eufy_sync.platform_support.macos.subprocess.run", run):
_real_notify("t", "m", command="eufy-sync --update")
argv = run.call_args.args[0]
@@ -63,5 +64,5 @@ def test_homebrew_path_is_checked_when_which_misses():
def test_notify_still_fails_silently():
- with patch("eufy_sync.cli.shared.shutil.which", side_effect=RuntimeError("boom")):
+ with patch("eufy_sync.platform_support.macos.shutil.which", side_effect=RuntimeError("boom")):
_real_notify("t", "m", command="eufy-sync --update") # must not raise
diff --git a/tests/test_platform_support.py b/tests/test_platform_support.py
new file mode 100644
index 0000000..d39f260
--- /dev/null
+++ b/tests/test_platform_support.py
@@ -0,0 +1,43 @@
+"""The platform_support package: implementation selection and the generic
+no-op layer, plus a spot-check that the macOS agent status is a faithful move
+of the old doctor launch-agent check.
+
+The full macOS behavior (notifications, plist bytes, launchctl invocations) is
+covered by test_notify.py and test_cli.py against the macos module directly.
+"""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from eufy_sync import platform_support
+from eufy_sync.platform_support import generic, macos
+
+
+def test_select_picks_macos_on_darwin():
+ with patch("eufy_sync.platform_support.platform.system", return_value="Darwin"):
+ assert platform_support._select() is macos
+
+
+def test_select_picks_generic_on_linux():
+ with patch("eufy_sync.platform_support.platform.system", return_value="Linux"):
+ assert platform_support._select() is generic
+
+
+def test_generic_notify_swallows_everything_and_runs_no_subprocess():
+ with patch("subprocess.run") as run:
+ assert generic.notify("t", "m", command="eufy-sync --update") is None
+ run.assert_not_called()
+
+
+def test_generic_agent_status_is_none():
+ assert generic.agent_status() is None
+
+
+def test_macos_agent_status_warns_when_not_installed(tmp_path):
+ with patch.object(macos, "LAUNCH_AGENT_PATH", tmp_path / "missing.plist"):
+ status = macos.agent_status()
+
+ assert status["status"] == "WARN"
+ assert status["label"] == "launch agent"
+ assert status["detail"] == "not installed"
+ assert status["fix"] == "eufy-sync --install-agent"
diff --git a/tests/test_platform_windows.py b/tests/test_platform_windows.py
new file mode 100644
index 0000000..2a197f9
--- /dev/null
+++ b/tests/test_platform_windows.py
@@ -0,0 +1,317 @@
+"""The Windows implementation: the hidden-window VBScript wrapper, the
+schtasks lifecycle, and the --doctor status check.
+
+Every subprocess and shutil.which call is patched, and shared.DATA_DIR /
+LOG_FILE are pointed at a tmp_path, so the suite runs on macOS without
+touching a real Task Scheduler.
+"""
+from __future__ import annotations
+
+import base64
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from eufy_sync import platform_support
+from eufy_sync.cli import shared
+from eufy_sync.platform_support import windows
+
+
+@pytest.fixture
+def data_dir(tmp_path, monkeypatch):
+ """Point the shared paths at a throwaway directory for the test."""
+ monkeypatch.setattr(shared, "DATA_DIR", tmp_path)
+ monkeypatch.setattr(shared, "LOG_FILE", tmp_path / "sync.log")
+ return tmp_path
+
+
+def _ok():
+ return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+
+def _write_current_wrapper(data_dir, binary):
+ """Write the wrapper exactly as the module would, preserving CRLF bytes."""
+ (data_dir / windows.WRAPPER_NAME).write_bytes(windows._wrapper_content(binary).encode("utf-8"))
+
+
+def test_select_picks_windows_on_windows():
+ with patch("eufy_sync.platform_support.platform.system", return_value="Windows"):
+ assert platform_support._select() is windows
+
+
+def test_install_invokes_schtasks_with_expected_arguments(data_dir):
+ binary = "C:\\Tools\\eufy-sync.exe"
+ with patch.object(windows.shutil, "which", return_value=binary), \
+ patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ windows.install_agent()
+
+ wrapper_path = data_dir / windows.WRAPPER_NAME
+ run.assert_called_once_with(
+ ["schtasks", "/Create", "/F", "/TN", "eufy-sync", "/SC", "HOURLY", "/MO", "4",
+ "/TR", f'wscript.exe "{wrapper_path}"'],
+ capture_output=True, text=True, timeout=15,
+ )
+
+
+def test_install_writes_wrapper_with_doubled_quote_escaping(data_dir):
+ binary = "C:\\Tools\\eufy-sync.exe"
+ with patch.object(windows.shutil, "which", return_value=binary), \
+ patch.object(windows.subprocess, "run", return_value=_ok()):
+ windows.install_agent()
+
+ # Read as raw bytes so the CRLF line endings are not translated away.
+ content = (data_dir / windows.WRAPPER_NAME).read_bytes().decode("utf-8")
+
+ # Full fidelity against the module's own generator.
+ assert content == windows._wrapper_content(binary)
+
+ # VBScript escapes a literal quote by doubling it. The original inner command
+ # opens with `cmd /c ""` and wraps the binary path in a quote, so after
+ # doubling every quote the leading marker becomes four quotes and the path is
+ # bracketed by four quotes on the left and two on the right. A wrong quote
+ # level here silently breaks the hidden-window run, so pin the exact bytes.
+ assert content.startswith('Set shell = CreateObject("WScript.Shell")\r\n')
+ assert 'shell.Run "cmd /c """"' in content
+ assert f'""""{binary}""' in content
+ assert content.endswith(', 0, False\r\n')
+
+
+def test_unchanged_wrapper_is_not_rewritten(data_dir):
+ binary = "C:\\Tools\\eufy-sync.exe"
+ _write_current_wrapper(data_dir, binary)
+
+ with patch.object(windows.shutil, "which", return_value=binary), \
+ patch.object(windows.subprocess, "run", return_value=_ok()), \
+ patch.object(Path, "write_bytes", autospec=True) as write_bytes:
+ windows.install_agent()
+
+ # The wrapper already holds the current content, so its bytes stay put.
+ write_bytes.assert_not_called()
+
+
+def test_install_without_binary_warns_and_skips(data_dir, capsys):
+ with patch.object(windows.shutil, "which", return_value=None), \
+ patch.object(windows.subprocess, "run") as run:
+ windows.install_agent()
+
+ run.assert_not_called()
+ assert "could not find eufy-sync on PATH" in capsys.readouterr().out
+
+
+def test_install_failure_prints_stderr(data_dir, capsys):
+ with patch.object(windows.shutil, "which", return_value="C:\\Tools\\eufy-sync.exe"), \
+ patch.object(windows.subprocess, "run",
+ return_value=SimpleNamespace(returncode=1, stdout="", stderr="Access is denied.")):
+ windows.install_agent()
+
+ out = capsys.readouterr().out
+ assert "Access is denied." in out
+ # The printed fallback command must quote the wrapper path inside the /TR
+ # value with escaped inner quotes, matching what the code passes
+ # programmatically, so a cmd.exe paste keeps a spaced path intact.
+ wrapper_path = data_dir / windows.WRAPPER_NAME
+ assert f'/TR "wscript.exe \\"{wrapper_path}\\""' in out
+
+
+def test_uninstall_deletes_task_and_removes_wrapper(data_dir):
+ wrapper_path = data_dir / windows.WRAPPER_NAME
+ wrapper_path.write_text("stale")
+
+ with patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ windows.uninstall_agent()
+
+ run.assert_called_once_with(
+ ["schtasks", "/Delete", "/F", "/TN", "eufy-sync"],
+ capture_output=True, text=True, timeout=15,
+ )
+ assert not wrapper_path.exists()
+
+
+def test_uninstall_without_task_reports_none_installed(data_dir, capsys):
+ with patch.object(windows.subprocess, "run",
+ return_value=SimpleNamespace(returncode=1, stdout="", stderr="ERROR: cannot find")):
+ windows.uninstall_agent()
+
+ assert "No scheduled task installed." in capsys.readouterr().out
+
+
+def test_agent_status_warns_when_wrapper_missing(data_dir):
+ status = windows.agent_status()
+
+ assert status["status"] == "WARN"
+ assert status["label"] == "scheduled task"
+ assert status["detail"] == "not installed"
+ assert status["fix"] == "eufy-sync --install-agent"
+
+
+def test_agent_status_warns_when_task_query_fails(data_dir):
+ binary = "C:\\Tools\\eufy-sync.exe"
+ _write_current_wrapper(data_dir, binary)
+
+ with patch.object(windows.shutil, "which", return_value=binary), \
+ patch.object(windows.subprocess, "run",
+ return_value=SimpleNamespace(returncode=1, stdout="", stderr="")):
+ status = windows.agent_status()
+
+ assert status["status"] == "WARN"
+ assert status["detail"] == "not installed"
+
+
+def test_agent_status_warns_when_wrapper_outdated(data_dir):
+ (data_dir / windows.WRAPPER_NAME).write_text("stale wrapper contents")
+
+ with patch.object(windows.shutil, "which", return_value="C:\\Tools\\eufy-sync.exe"), \
+ patch.object(windows.subprocess, "run", return_value=_ok()):
+ status = windows.agent_status()
+
+ assert status["status"] == "WARN"
+ assert status["detail"] == "outdated registration"
+ assert status["fix"] == "eufy-sync --install-agent"
+
+
+def test_agent_status_passes_when_installed_and_current(data_dir):
+ binary = "C:\\Tools\\eufy-sync.exe"
+ _write_current_wrapper(data_dir, binary)
+
+ with patch.object(windows.shutil, "which", return_value=binary), \
+ patch.object(windows.subprocess, "run", return_value=_ok()):
+ status = windows.agent_status()
+
+ assert status["status"] == "PASS"
+ assert status["detail"] == "installed, runs every 4h"
+ assert status["fix"] is None
+
+
+def test_agent_installed_reflects_wrapper_presence(data_dir):
+ assert windows.agent_installed() is False
+ (data_dir / windows.WRAPPER_NAME).write_text("x")
+ assert windows.agent_installed() is True
+
+
+def test_purge_agent_removes_task_and_wrapper_silently(data_dir, capsys):
+ wrapper_path = data_dir / windows.WRAPPER_NAME
+ wrapper_path.write_text("x")
+
+ with patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ windows.purge_agent()
+
+ run.assert_called_once_with(
+ ["schtasks", "/Delete", "/F", "/TN", "eufy-sync"],
+ capture_output=True,
+ )
+ assert not wrapper_path.exists()
+ assert capsys.readouterr().out == ""
+
+
+def test_purge_agent_noop_when_not_installed(data_dir):
+ with patch.object(windows.subprocess, "run") as run:
+ windows.purge_agent()
+
+ run.assert_not_called()
+
+
+def _decoded_script(run):
+ """Pull the -EncodedCommand payload off a captured subprocess.run call and
+ decode it back into the PowerShell script text."""
+ argv = run.call_args.args[0]
+ encoded = argv[argv.index("-EncodedCommand") + 1]
+ return base64.b64decode(encoded).decode("utf-16-le")
+
+
+def test_notify_invokes_powershell_with_encoded_command(data_dir):
+ with patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ assert windows.notify("Sync failed", "Reauth needed") is None
+
+ run.assert_called_once()
+ argv = run.call_args.args[0]
+ assert argv[:2] == ["powershell", "-NoProfile"]
+ assert "-EncodedCommand" in argv
+ assert run.call_args.kwargs["capture_output"] is True
+ assert run.call_args.kwargs["timeout"] == 10
+
+
+def test_notify_encodes_xml_escaped_message(data_dir):
+ with patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ windows.notify("Title", "Tom & Jerry ")
+
+ decoded = _decoded_script(run)
+ # The ampersand and angle brackets are neutralized as XML entities, so no
+ # raw special survives to break the toast XML.
+ assert "&" in decoded
+ assert "<fixed>" in decoded
+ assert "Tom & Jerry" not in decoded
+
+
+def test_notify_message_with_specials_and_quotes_cannot_break_out(data_dir):
+ # A payload packed with the characters that would end the here-string, close
+ # a quote, or open an XML tag if any of them leaked through unescaped.
+ nasty = "'@ \"; & done"
+ with patch.object(windows.subprocess, "run", return_value=_ok()) as run:
+ windows.notify("t", nasty)
+
+ decoded = _decoded_script(run)
+ # The XML specials are entity-encoded, so no live tag or bare ampersand lands
+ # in the text node.
+ assert "