Compare commits

..

10 Commits

Author SHA1 Message Date
Bastian Wagner
990a55af14 fix: address final review findings for sync-scheduler-web plan
Close the leaked httpx.AsyncClient in MyWhoosh sync runs, ensure hard
failures finish sync_runs as FAILED instead of leaving them stuck at
RUNNING, log (non-benign) exceptions surfaced by sync_all_enabled during
scheduled ticks, classify GarminImportRejected as a non-retryable
per-activity failure, fix a bug where a live Garmin-class action_required
state could be silently cleared by a run that did no Garmin work, use the
activity's DB primary key instead of the unsanitized remote id for
filesystem paths, add regression/coverage tests for the health-state fix
and MFA code threading through the real SyncManager, add idempotency
coverage to the two-user acceptance test, and note the Dockerfile's
single-process assumption.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-15 17:09:55 +02:00
Bastian Wagner
aed9d6bb48 test: cover multi-user sync acceptance 2026-08-15 16:43:45 +02:00
Bastian Wagner
6657124983 feat: handle Garmin MFA and activity retries 2026-08-15 16:33:16 +02:00
Bastian Wagner
c2f13611b9 feat: add operational sync controls 2026-08-15 16:24:11 +02:00
Bastian Wagner
fd50bbbab7 feat: schedule periodic user synchronization 2026-08-15 16:14:25 +02:00
Bastian Wagner
1d414d6298 feat: isolate concurrent user syncs
Wrap SyncManager.sync_user in a per-user asyncio.Lock (raising
SyncAlreadyRunning on overlap) and add sync_all_enabled() to fan out
across all enabled users with per-user failure isolation via
asyncio.gather(return_exceptions=True). The prior sync_user body is
renamed to _sync_user_locked with no logic changes.
2026-08-15 16:08:00 +02:00
Bastian Wagner
4aaf490bc5 style: use SQLAlchemy 2.0 select() in sync test helpers
Match the select()/scalars() convention already used throughout
app/db/repositories.py instead of the legacy Query API.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-15 15:54:52 +02:00
Bastian Wagner
c4e986e3f8 feat: add resumable per-user sync pipeline 2026-08-15 15:53:27 +02:00
Bastian Wagner
1d5bbdb2a2 feat: add durable sync state transitions
Add state-transition methods to ActivityRepository for advancing activity
stages (mark_downloaded, mark_converted, mark_imported, mark_duplicate,
mark_failed) with proper retention of last_completed_stage on failure.
Add list_pending_for_user to filter activities for processing. Implement
SyncRunRepository for creating and finalizing sync runs with counts and
summary errors.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-15 15:33:46 +02:00
Bastian Wagner
2f65c0178c fix: address final review findings for service clients plan
Fixes 9 numbered findings + 7 minor fixes from the whole-plan review of
the MyWhoosh/Garmin service clients (Plan 3):

Garmin uploader (app/garmin/uploader.py):
- Detect Garmin-rejected imports (failures without successes) and raise
  new GarminImportRejected instead of reporting them as successful.
- Reclassify 429/rate-limit/500 login failures as transient instead of
  falling through to permanent auth errors; unrecognized login failures
  are now treated as transient (retryable) rather than GarminAuthError.
- Mirror the auth-token check from the login branch into the import
  branch so 401-at-import-time raises GarminAuthError instead of
  propagating raw.
- Add common GarminError base class, hoist transient-token tuple to a
  shared module constant, check response.status_code==409 before the
  duplicate substring fallback, and create the tokenstore dir 0o700.

MyWhoosh client (app/mywhoosh/client.py):
- Add optional max_pages bound to list_activities pagination.
- Add aclose()/__aenter__/__aexit__ so the client's own httpx.AsyncClient
  gets closed, while never closing an injected client.
- Guard the two remaining unguarded JSON-decode paths (login body,
  download-fit metadata) so malformed bodies raise
  MyWhooshIntegrationError instead of raw ValueError/AttributeError.
- Row-level malformation (missing id/activityFileId, unparseable
  startDatetime) is now skipped rather than aborting the whole page;
  envelope-shape failures still raise. id/activityFileId checks use
  explicit None/"" comparisons instead of Python falsiness.
- Replace asserts in _authenticated_post with explicit exceptions;
  restrict the reauth retry to 401 only, treat 403 as immediately
  terminal; naive startDatetime values are now treated as already-UTC
  instead of host-local.

MyWhoosh tokenstore (app/mywhoosh/tokenstore.py):
- load() now treats any corrupt/malformed token file (bad JSON, missing
  keys, OS errors) as "absent" instead of raising, so a bad cache no
  longer permanently wedges a user.

pyproject.toml:
- Tighten garminconnect pin to >=0.3.10,<1 (import_activity requires
  0.3.10+).

Adds/updates tests across tests/mywhoosh/ and tests/garmin/ covering
all of the above, including a fake client that wraps GarminUploadBlocked
in a plain RuntimeError to mirror the real garminconnect library's MFA
error wrapping.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-15 15:23:14 +02:00
34 changed files with 2656 additions and 50 deletions

View File

@@ -7,4 +7,8 @@ COPY app /app/app
RUN mkdir -p /data && chmod 700 /data
ENV DATA_DIR=/data
EXPOSE 8080
# Single-process assumption: per-user sync locking and the in-process
# scheduler both live in this one worker's memory. Do not scale this to
# multiple uvicorn workers or container replicas without adding a
# cross-process lock -- otherwise duplicate imports become possible.
CMD ["uvicorn", "app.main:create_app", "--factory", "--host", "0.0.0.0", "--port", "8080"]

View File

@@ -1,10 +1,23 @@
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy import select
from sqlalchemy import and_, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.db.models import Activity, ActivityStatus, SyncUser
from app.db.models import Activity, ActivityStatus, SyncRun, SyncRunStatus, SyncUser, utcnow
@dataclass(frozen=True)
class UserDashboardRow:
id: int
name: str
enabled: bool
health_state: str
action_reason: str | None
last_sync_at: datetime | None
last_activity_name: str | None
last_activity_status: str | None
class UserRepository:
@@ -32,11 +45,42 @@ class UserRepository:
self.session.commit()
return user
def dashboard_rows(self) -> list[UserDashboardRow]:
users = self.list_all()
rows = []
for user in users:
last_run = self.session.scalar(
select(SyncRun).where(SyncRun.user_id == user.id).order_by(SyncRun.started_at.desc()).limit(1)
)
last_activity = self.session.scalar(
select(Activity).where(Activity.user_id == user.id).order_by(Activity.created_at.desc()).limit(1)
)
rows.append(UserDashboardRow(
id=user.id,
name=user.name,
enabled=user.enabled,
health_state=user.health_state.value,
action_reason=user.action_reason,
last_sync_at=last_run.finished_at if last_run else None,
last_activity_name=last_activity.activity_name if last_activity else None,
last_activity_status=last_activity.status.value if last_activity else None,
))
return rows
class ActivityRepository:
def __init__(self, session: Session) -> None:
self.session = session
def _require(self, activity_id: int) -> Activity:
activity = self.session.get(Activity, activity_id)
if activity is None:
raise ValueError(f"activity {activity_id} not found")
return activity
def get(self, activity_id: int) -> Activity | None:
return self.session.get(Activity, activity_id)
def get_or_create_discovered(
self,
*,
@@ -76,3 +120,110 @@ class ActivityRepository:
raise
return existing, False
return activity, True
def mark_downloaded(self, activity_id: int, path: str) -> Activity:
activity = self._require(activity_id)
activity.source_fit_path = path
activity.status = ActivityStatus.DOWNLOADED
activity.last_completed_stage = ActivityStatus.DOWNLOADED
activity.last_error = None
activity.retryable = True
self.session.commit()
return activity
def mark_converted(self, activity_id: int, path: str) -> Activity:
activity = self._require(activity_id)
activity.converted_fit_path = path
activity.status = ActivityStatus.CONVERTED
activity.last_completed_stage = ActivityStatus.CONVERTED
activity.last_error = None
activity.retryable = True
self.session.commit()
return activity
def mark_imported(self, activity_id: int, garmin_activity_id: str | None) -> Activity:
activity = self._require(activity_id)
activity.status = ActivityStatus.IMPORTED
activity.last_completed_stage = ActivityStatus.IMPORTED
activity.garmin_activity_id = garmin_activity_id
activity.last_error = None
activity.retryable = False
self.session.commit()
return activity
def mark_duplicate(self, activity_id: int) -> Activity:
activity = self._require(activity_id)
activity.status = ActivityStatus.DUPLICATE
activity.last_completed_stage = ActivityStatus.DUPLICATE
activity.last_error = None
activity.retryable = False
self.session.commit()
return activity
def mark_failed(self, activity_id: int, error: str, *, retryable: bool) -> Activity:
activity = self._require(activity_id)
activity.status = ActivityStatus.FAILED
activity.last_error = error[:2000]
activity.retryable = retryable
self.session.commit()
return activity
def reset_retryable_failure(self, activity_id: int) -> Activity:
activity = self._require(activity_id)
if activity.status != ActivityStatus.FAILED or not activity.retryable:
raise ValueError("activity is not retryable")
activity.status = activity.last_completed_stage
activity.last_error = None
self.session.commit()
return activity
def list_pending_for_user(self, user_id: int) -> list[Activity]:
return list(
self.session.scalars(
select(Activity).where(
Activity.user_id == user_id,
or_(
Activity.status.in_([ActivityStatus.DISCOVERED, ActivityStatus.DOWNLOADED, ActivityStatus.CONVERTED]),
and_(Activity.status == ActivityStatus.FAILED, Activity.retryable.is_(True)),
),
).order_by(Activity.id)
)
)
class SyncRunRepository:
def __init__(self, session: Session) -> None:
self.session = session
def start(self, user_id: int) -> SyncRun:
sync_run = SyncRun(user_id=user_id, status=SyncRunStatus.RUNNING)
self.session.add(sync_run)
self.session.commit()
return sync_run
def get(self, sync_run_id: int) -> SyncRun | None:
return self.session.get(SyncRun, sync_run_id)
def finish(
self,
sync_run_id: int,
*,
status: SyncRunStatus,
discovered: int,
imported: int,
skipped: int,
failed: int,
summary_error: str | None = None,
) -> SyncRun:
sync_run = self.session.get(SyncRun, sync_run_id)
if sync_run is None:
raise ValueError(f"sync_run {sync_run_id} not found")
sync_run.finished_at = utcnow()
sync_run.status = status
sync_run.discovered_count = discovered
sync_run.imported_count = imported
sync_run.skipped_count = skipped
sync_run.failed_count = failed
sync_run.summary_error = summary_error[:2000] if summary_error else None
self.session.commit()
return sync_run

View File

@@ -18,18 +18,40 @@ class UploadResult:
raw_response: Any
class GarminUploadBlocked(RuntimeError):
class GarminError(RuntimeError):
pass
class GarminAuthError(RuntimeError):
class GarminUploadBlocked(GarminError):
pass
class GarminTransientError(RuntimeError):
class GarminAuthError(GarminError):
pass
class GarminTransientError(GarminError):
pass
class GarminImportRejected(GarminError):
pass
_TRANSIENT_ERROR_TOKENS = (
"timeout",
"temporar",
"connection",
"429",
"too many",
"rate limit",
"500",
"502",
"503",
"504",
)
class GarminUploader:
def __init__(
self,
@@ -56,9 +78,12 @@ class GarminUploader:
if _looks_duplicate_error(exc):
return UploadResult("duplicate", True, None, str(exc))
text = str(exc).lower()
if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")):
if any(token in text for token in _TRANSIENT_ERROR_TOKENS):
raise GarminTransientError("Garmin import failed transiently") from exc
if any(token in text for token in ("password", "credential", "unauthorized", "401")):
raise GarminAuthError("Garmin import failed: authentication rejected") from exc
raise
_raise_if_import_rejected(response)
return UploadResult("imported", False, _extract_activity_id(response), response)
finally:
self._mfa_code = None
@@ -66,7 +91,7 @@ class GarminUploader:
def _ensure_client(self) -> GarminClientProtocol:
if self._client is not None:
return self._client
self.tokenstore.mkdir(parents=True, exist_ok=True)
self.tokenstore.mkdir(parents=True, exist_ok=True, mode=0o700)
factory = self.client_factory or _default_garmin_factory
client = factory(self.email, self.password, prompt_mfa=self._prompt_mfa)
try:
@@ -79,9 +104,9 @@ class GarminUploader:
raise GarminUploadBlocked("Garmin MFA is required") from exc
if any(token in text for token in ("password", "credential", "unauthorized", "401")):
raise GarminAuthError("Garmin authentication failed") from exc
if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")):
if any(token in text for token in _TRANSIENT_ERROR_TOKENS):
raise GarminTransientError("Garmin login failed transiently") from exc
raise GarminAuthError("Garmin login failed") from exc
raise GarminTransientError("Garmin login failed") from exc
self._client = client
return client
@@ -98,10 +123,25 @@ def _default_garmin_factory(*args: Any, **kwargs: Any) -> GarminClientProtocol:
def _looks_duplicate_error(exc: Exception) -> bool:
status_code = getattr(getattr(exc, "response", None), "status_code", None)
if status_code == 409:
return True
text = str(exc).lower()
return any(token in text for token in ("duplicate", "already exists", "409"))
def _raise_if_import_rejected(response: Any) -> None:
if not isinstance(response, dict):
return
detailed = response.get("detailedImportResult")
if not isinstance(detailed, dict):
return
failures = detailed.get("failures")
successes = detailed.get("successes")
if isinstance(failures, list) and failures and not successes:
raise GarminImportRejected(f"Garmin rejected the import ({len(failures)} failure(s))")
def _extract_activity_id(response: Any) -> str | None:
if not isinstance(response, dict):
return None

View File

@@ -1,8 +1,17 @@
from contextlib import asynccontextmanager
from fastapi import FastAPI
from starlette.middleware.sessions import SessionMiddleware
from app.config import Settings, get_settings
from app.db.session import create_db_engine, create_session_factory, initialize_schema
from app.fit.rewriter import convert_fit_device
from app.garmin.uploader import GarminUploader
from app.mywhoosh.client import MyWhooshClient
from app.security.credentials import CredentialCipher
from app.sync.manager import SyncManager
from app.sync.scheduler import SyncScheduler
from app.web.operations import router as operations_router
from app.web.routes import router as web_router
@@ -12,7 +21,35 @@ def create_app(settings: Settings | None = None) -> FastAPI:
resolved.tokens_dir.mkdir(parents=True, exist_ok=True)
resolved.activities_dir.mkdir(parents=True, exist_ok=True)
app = FastAPI(title="MyWhoosh Garmin Sync")
@asynccontextmanager
async def lifespan(app: FastAPI):
cipher = CredentialCipher(resolved.credential_encryption_key)
def mywhoosh_factory(token_store):
return MyWhooshClient(token_store)
def garmin_factory(email, password, tokenstore):
return GarminUploader(email=email, password=password, tokenstore=tokenstore)
sync_manager = SyncManager(
session_factory=app.state.session_factory,
credential_cipher=cipher,
settings=resolved,
mywhoosh_factory=mywhoosh_factory,
garmin_factory=garmin_factory,
fit_converter=convert_fit_device,
)
app.state.sync_manager = sync_manager
scheduler = SyncScheduler(sync_manager, interval_seconds=resolved.sync_interval_minutes * 60)
app.state.scheduler = scheduler
await scheduler.start()
yield
await scheduler.stop()
app = FastAPI(title="MyWhoosh Garmin Sync", lifespan=lifespan)
app.state.settings = resolved
engine = create_db_engine(resolved.database_url)
@@ -27,6 +64,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
https_only=resolved.session_https_only,
)
app.include_router(web_router)
app.include_router(operations_router)
@app.get("/healthz")
def healthz() -> dict[str, str]:

View File

@@ -31,9 +31,20 @@ class MyWhooshIntegrationError(MyWhooshError):
class MyWhooshClient:
def __init__(self, token_store: MyWhooshTokenStore, http_client: httpx.AsyncClient | None = None) -> None:
self.token_store = token_store
self._owns_http = http_client is None
self.http = http_client or httpx.AsyncClient(timeout=30.0)
self.token = token_store.load()
async def aclose(self) -> None:
if self._owns_http:
await self.http.aclose()
async def __aenter__(self) -> "MyWhooshClient":
return self
async def __aexit__(self, *exc_info: object) -> None:
await self.aclose()
async def login(self, email: str, password: str) -> None:
payload = {
"Username": email,
@@ -56,6 +67,8 @@ class MyWhooshClient:
body = response.json()
except ValueError as exc:
raise MyWhooshIntegrationError("MyWhoosh login returned invalid JSON") from exc
if not isinstance(body, dict):
raise MyWhooshIntegrationError("MyWhoosh login response is not a JSON object")
if body.get("Success") is not True or not body.get("AccessToken"):
raise MyWhooshAuthError(str(body.get("Message") or "MyWhoosh login failed"))
self.token = MyWhooshToken(
@@ -72,7 +85,8 @@ class MyWhooshClient:
async def _authenticated_post(self, url: str, payload: dict, email: str, password: str) -> httpx.Response:
await self.ensure_authenticated(email, password)
for attempt in range(2):
assert self.token is not None
if self.token is None:
raise MyWhooshIntegrationError("no token after ensure_authenticated")
try:
response = await self.http.post(
url,
@@ -81,7 +95,9 @@ class MyWhooshClient:
)
except httpx.TransportError as exc:
raise MyWhooshTransientError("MyWhoosh request failed") from exc
if response.status_code not in {401, 403}:
if response.status_code == 403:
raise MyWhooshAuthError(f"MyWhoosh returned HTTP {response.status_code}")
if response.status_code != 401:
if response.status_code >= 500:
raise MyWhooshTransientError(f"MyWhoosh returned HTTP {response.status_code}")
return response
@@ -91,13 +107,15 @@ class MyWhooshClient:
await self.login(email, password)
continue
raise MyWhooshAuthError("MyWhoosh session rejected after reauthentication")
raise AssertionError("unreachable")
raise MyWhooshIntegrationError("unreachable state in _authenticated_post")
async def list_activities(self, email: str, password: str) -> list[MyWhooshActivity]:
async def list_activities(
self, email: str, password: str, max_pages: int | None = None
) -> list[MyWhooshActivity]:
activities: list[MyWhooshActivity] = []
page = 1
total_pages = 1
while page <= total_pages:
while page <= total_pages and (max_pages is None or page <= max_pages):
response = await self._authenticated_post(
ACTIVITIES_BASE + "rider/profile/activities",
{"sortDate": "DESC", "page": page},
@@ -119,26 +137,30 @@ class MyWhooshClient:
if not isinstance(results, list):
raise MyWhooshIntegrationError("MyWhoosh activities response has unexpected shape")
for row in results:
activities.append(self._normalize_activity(row))
activity = self._normalize_activity(row)
if activity is not None:
activities.append(activity)
page += 1
return activities
def _normalize_activity(self, row: object) -> MyWhooshActivity:
def _normalize_activity(self, row: object) -> MyWhooshActivity | None:
if not isinstance(row, dict):
raise MyWhooshIntegrationError("MyWhoosh activity row is not an object")
activity_id = row.get("id")
activity_file_id = row.get("activityFileId")
if not activity_id or not activity_file_id:
raise MyWhooshIntegrationError("MyWhoosh activity row missing stable id or activityFileId")
if activity_id is None or activity_id == "" or activity_file_id is None or activity_file_id == "":
return None
raw_started = row.get("startDatetime")
started_at: datetime | None = None
if raw_started:
try:
started_at = datetime.fromisoformat(str(raw_started).replace("Z", "+00:00")).astimezone(
timezone.utc
)
except ValueError as exc:
raise MyWhooshIntegrationError("MyWhoosh activity row has invalid startDatetime") from exc
parsed = datetime.fromisoformat(str(raw_started).replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
started_at = parsed.replace(tzinfo=timezone.utc)
else:
started_at = parsed.astimezone(timezone.utc)
return MyWhooshActivity(
id=str(activity_id),
title=str(row.get("title") or ""),
@@ -155,7 +177,13 @@ class MyWhooshClient:
)
if response.status_code >= 400:
raise MyWhooshIntegrationError(f"download metadata returned HTTP {response.status_code}")
url = response.json().get("data")
try:
body = response.json()
except ValueError as exc:
raise MyWhooshIntegrationError("MyWhoosh download response returned invalid JSON") from exc
if not isinstance(body, dict):
raise MyWhooshIntegrationError("MyWhoosh download response is not a JSON object")
url = body.get("data")
if not isinstance(url, str) or not url:
raise MyWhooshIntegrationError("MyWhoosh download response has no URL")
try:

View File

@@ -12,13 +12,13 @@ class MyWhooshTokenStore:
def load(self) -> MyWhooshToken | None:
try:
raw = json.loads(self.path.read_text("utf-8"))
except FileNotFoundError:
return MyWhooshToken(
access_token=raw["access_token"],
refresh_token=raw.get("refresh_token"),
whoosh_id=raw.get("whoosh_id"),
)
except (FileNotFoundError, OSError, ValueError, KeyError, TypeError):
return None
return MyWhooshToken(
access_token=raw["access_token"],
refresh_token=raw.get("refresh_token"),
whoosh_id=raw.get("whoosh_id"),
)
def save(self, token: MyWhooshToken) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)

0
app/sync/__init__.py Normal file
View File

382
app/sync/manager.py Normal file
View File

@@ -0,0 +1,382 @@
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
from typing import Any, Callable
from app.db.models import ActivityStatus, HealthState, SyncRunStatus
from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository
from app.fit.rewriter import FitFormatError
from app.garmin.uploader import (
GarminAuthError,
GarminImportRejected,
GarminTransientError,
GarminUploadBlocked,
)
from app.mywhoosh.client import MyWhooshAuthError, MyWhooshIntegrationError, MyWhooshTransientError
from app.mywhoosh.tokenstore import MyWhooshTokenStore
from app.security.credentials import CredentialCipher
from app.sync.states import SyncOutcome
logger = logging.getLogger(__name__)
class SyncAlreadyRunning(RuntimeError):
pass
# action_reason values that are resolved purely by this run's MyWhoosh listing
# succeeding (reaching the end of the activity loop with stop_user_run still
# False already proves MyWhoosh is working again -- every MyWhoosh-related
# exception branch that could fire also sets stop_user_run=True).
_MYWHOOSH_ACTION_REASONS = frozenset({"mywhoosh_auth_required", "mywhoosh_integration_changed"})
# action_reason values that can only be resolved by actual, this-run evidence
# of a successful Garmin import -- the mere absence of a Garmin exception does
# NOT prove anything, since no Garmin work may have been attempted this run.
_GARMIN_ACTION_REASONS = frozenset({"garmin_mfa_required", "garmin_auth_required"})
class SyncManager:
"""Resumable single-user MyWhoosh -> Garmin sync pipeline.
`_sync_user_locked` implements the state machine for one user's sync run.
`sync_user` wraps it with a per-user `asyncio.Lock` so only one sync can
run for a given user at a time (raising `SyncAlreadyRunning` on overlap),
and `sync_all_enabled` fans out across all enabled users, isolating each
user's failure to its own result rather than cancelling siblings.
"""
def __init__(
self,
*,
session_factory: Callable[[], Any],
credential_cipher: CredentialCipher,
settings: Any,
mywhoosh_factory: Callable[[MyWhooshTokenStore], Any],
garmin_factory: Callable[[str, str, Path], Any],
fit_converter: Callable[[Path, Path], Any],
) -> None:
self.session_factory = session_factory
self.credential_cipher = credential_cipher
self.settings = settings
self.mywhoosh_factory = mywhoosh_factory
self.garmin_factory = garmin_factory
self.fit_converter = fit_converter
self._locks: dict[int, asyncio.Lock] = {}
self._locks_guard = asyncio.Lock()
async def _lock_for(self, user_id: int) -> asyncio.Lock:
async with self._locks_guard:
return self._locks.setdefault(user_id, asyncio.Lock())
async def sync_user(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome:
lock = await self._lock_for(user_id)
if lock.locked():
raise SyncAlreadyRunning(f"sync already running for user {user_id}")
async with lock:
return await self._sync_user_locked(user_id, mfa_code)
def _load_enabled_user_ids(self) -> list[int]:
with self.session_factory() as session:
return [user.id for user in UserRepository(session).list_enabled()]
async def sync_all_enabled(self) -> list[SyncOutcome | Exception]:
user_ids = self._load_enabled_user_ids()
return await asyncio.gather(
*(self.sync_user(user_id) for user_id in user_ids),
return_exceptions=True,
)
async def _sync_user_locked(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome:
with self.session_factory() as session:
user = UserRepository(session).get(user_id)
if user is None:
raise ValueError(f"user {user_id} not found")
sync_run_repo = SyncRunRepository(session)
run = sync_run_repo.start(user_id)
# Best-available counters, tracked outside the inner try so that if
# something raises before/while they'd normally be populated, the
# outer except below can still report whatever we do know.
discovered = 0
imported_count = 0
skipped_count = 0
failed_count = 0
mywhoosh = None
try:
try:
mw_email = self.credential_cipher.decrypt(user.mywhoosh_email_enc)
mw_password = self.credential_cipher.decrypt(user.mywhoosh_password_enc)
garmin_email = self.credential_cipher.decrypt(user.garmin_email_enc)
garmin_password = self.credential_cipher.decrypt(user.garmin_password_enc)
token_dir = self.settings.tokens_dir / str(user.id)
mywhoosh = self.mywhoosh_factory(MyWhooshTokenStore(token_dir / "mywhoosh.json"))
garmin = self.garmin_factory(garmin_email, garmin_password, token_dir / "garmin")
activity_repo = ActivityRepository(session)
stop_user_run = False
summary_error: str | None = None
# True only once an actual, this-run Garmin import/duplicate
# check has succeeded for a real activity -- the sole
# evidence that can justify clearing a Garmin-class
# action_reason (see _GARMIN_ACTION_REASONS below).
garmin_succeeded_this_run = False
try:
remote_activities = await mywhoosh.list_activities(mw_email, mw_password)
except MyWhooshTransientError as exc:
user.health_state = HealthState.DEGRADED
user.mywhoosh_state = "error"
session.commit()
remote_activities = []
stop_user_run = True
summary_error = str(exc)
except MyWhooshAuthError as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.mywhoosh_state = "auth_required"
user.action_reason = "mywhoosh_auth_required"
session.commit()
remote_activities = []
stop_user_run = True
summary_error = str(exc)
except MyWhooshIntegrationError as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.mywhoosh_state = "integration_error"
user.action_reason = "mywhoosh_integration_changed"
session.commit()
remote_activities = []
stop_user_run = True
summary_error = str(exc)
except Exception as exc:
user.health_state = HealthState.DEGRADED
user.mywhoosh_state = "error"
session.commit()
remote_activities = []
stop_user_run = True
summary_error = f"{type(exc).__name__}: {str(exc)[:200]}"
logger.warning(
"sync_user: unexpected error listing activities for user %s: %s",
user.id,
exc.__class__.__name__,
)
else:
user.mywhoosh_state = "connected"
session.commit()
discovered = len(remote_activities)
for remote in remote_activities:
if stop_user_run:
break
activity, _created = activity_repo.get_or_create_discovered(
user_id=user.id,
mywhoosh_activity_id=remote.id,
activity_name=remote.title,
activity_timestamp=remote.started_at,
)
# Non-retryable failures (e.g. corrupt/unsupported FIT
# files) are terminal: never re-attempt them, and don't
# count them in any counter for this run.
if activity.status == ActivityStatus.FAILED and not activity.retryable:
continue
stage = (
activity.last_completed_stage
if activity.status == ActivityStatus.FAILED
else activity.status
)
initial_stage = stage
# Use the DB's own numeric primary key rather than the
# upstream-supplied mywhoosh_activity_id as a directory
# component: activity.id is always a safe integer, so
# this avoids any path-traversal risk from an
# unsanitized remote id (e.g. "../../etc") while
# remaining just as stable across resumed syncs.
activity_dir = self.settings.activities_dir / str(user.id) / str(activity.id)
source_path = activity_dir / "source.fit"
converted_path = activity_dir / "edge-1030-plus.fit"
try:
if stage == ActivityStatus.DISCOVERED:
fit_bytes = await mywhoosh.download_fit(
remote.activity_file_id, mw_email, mw_password
)
activity_dir.mkdir(parents=True, exist_ok=True)
source_path.write_bytes(fit_bytes)
activity = activity_repo.mark_downloaded(activity.id, str(source_path))
stage = activity.status
if stage in {ActivityStatus.DOWNLOADED}:
self.fit_converter(source_path, converted_path)
activity = activity_repo.mark_converted(activity.id, str(converted_path))
stage = activity.status
if stage in {ActivityStatus.CONVERTED}:
upload = await asyncio.to_thread(garmin.import_fit, converted_path, mfa_code)
if upload.duplicate:
activity = activity_repo.mark_duplicate(activity.id)
else:
activity = activity_repo.mark_imported(activity.id, upload.garmin_activity_id)
stage = activity.status
user.garmin_state = "connected"
garmin_succeeded_this_run = True
# Only count this activity's outcome toward this
# run's totals if the state machine actually did
# work this call. An activity that was already
# terminal (IMPORTED/DUPLICATE) before this call is
# resume history, not this run's work.
if initial_stage not in (ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE):
if activity.status == ActivityStatus.IMPORTED:
imported_count += 1
elif activity.status == ActivityStatus.DUPLICATE:
skipped_count += 1
except MyWhooshTransientError as exc:
user.health_state = HealthState.DEGRADED
user.mywhoosh_state = "error"
activity_repo.mark_failed(activity.id, str(exc), retryable=True)
failed_count += 1
except MyWhooshAuthError as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.mywhoosh_state = "auth_required"
user.action_reason = "mywhoosh_auth_required"
stop_user_run = True
summary_error = str(exc)
except MyWhooshIntegrationError as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.mywhoosh_state = "integration_error"
user.action_reason = "mywhoosh_integration_changed"
stop_user_run = True
summary_error = str(exc)
except GarminUploadBlocked as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.garmin_state = "mfa_required"
user.action_reason = "garmin_mfa_required"
stop_user_run = True
summary_error = str(exc)
except GarminAuthError as exc:
user.health_state = HealthState.ACTION_REQUIRED
user.garmin_state = "auth_required"
user.action_reason = "garmin_auth_required"
stop_user_run = True
summary_error = str(exc)
except GarminTransientError as exc:
user.health_state = HealthState.DEGRADED
user.garmin_state = "error"
activity_repo.mark_failed(activity.id, str(exc), retryable=True)
failed_count += 1
except GarminImportRejected as exc:
# Garmin-side permanent content rejection (real
# failures, no successes in detailedImportResult):
# same category as a corrupt/unsupported FIT file --
# a non-retryable per-activity failure, no user
# health/state change.
activity_repo.mark_failed(activity.id, str(exc), retryable=False)
failed_count += 1
except FitFormatError as exc:
activity_repo.mark_failed(activity.id, str(exc), retryable=False)
failed_count += 1
except Exception as exc:
user.health_state = HealthState.DEGRADED
activity_repo.mark_failed(
activity.id,
f"{type(exc).__name__}: {str(exc)[:200]}",
retryable=True,
)
failed_count += 1
logger.warning(
"sync_user: unexpected error for user %s activity %s: %s",
user.id,
activity.id,
exc.__class__.__name__,
)
session.commit()
if not stop_user_run:
# Reaching here with stop_user_run still False proves
# this run's MyWhoosh listing succeeded (every
# MyWhoosh-exception branch above also sets
# stop_user_run=True) -- so a MyWhoosh-class
# action_reason is always safe to clear here. It does
# NOT prove any Garmin problem was fixed: a Garmin-class
# action_reason may only be cleared when this run
# actually succeeded at a real Garmin import
# (garmin_succeeded_this_run). Otherwise the
# action-required condition is still live and
# unverified, so leave both action_reason and
# health_state untouched.
reason = user.action_reason
can_clear = (
reason is None
or reason in _MYWHOOSH_ACTION_REASONS
or (reason in _GARMIN_ACTION_REASONS and garmin_succeeded_this_run)
)
if can_clear:
user.action_reason = None
user.health_state = (
HealthState.DEGRADED if failed_count > 0 else HealthState.HEALTHY
)
session.commit()
status = (
SyncRunStatus.SUCCESS
if failed_count == 0 and not stop_user_run
else SyncRunStatus.PARTIAL
if (imported_count + skipped_count) > 0
else SyncRunStatus.FAILED
)
sync_run_repo.finish(
run.id,
status=status,
discovered=discovered,
imported=imported_count,
skipped=skipped_count,
failed=failed_count,
summary_error=summary_error,
)
session.commit()
return SyncOutcome(
user_id=user.id,
status=status.value,
discovered=discovered,
imported=imported_count,
skipped=skipped_count,
failed=failed_count,
message=summary_error,
)
except Exception as exc:
# Any unhandled exception escaping the block above (e.g. a
# credential-decrypt failure after key rotation, a factory
# constructor raising, an unexpected DB error) would
# otherwise leave this SyncRun stuck at
# status=RUNNING/finished_at=NULL forever. Finish it as
# FAILED with whatever counters we do have, then
# re-propagate so callers (e.g. sync_all_enabled's
# asyncio.gather(return_exceptions=True)) still see it.
session.rollback()
sync_run_repo.finish(
run.id,
status=SyncRunStatus.FAILED,
discovered=discovered,
imported=imported_count,
skipped=skipped_count,
failed=failed_count,
summary_error=f"{type(exc).__name__}: {str(exc)[:200]}",
)
session.commit()
raise
finally:
if mywhoosh is not None:
aclose = getattr(mywhoosh, "aclose", None)
if aclose is not None:
await aclose()

57
app/sync/scheduler.py Normal file
View File

@@ -0,0 +1,57 @@
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from app.sync.manager import SyncAlreadyRunning
logger = logging.getLogger(__name__)
class SyncScheduler:
def __init__(self, manager, *, interval_seconds: float) -> None:
self.manager = manager
self.interval_seconds = interval_seconds
self._task: asyncio.Task | None = None
self._stop = asyncio.Event()
self.last_tick = None
self.next_tick = None
async def run_once(self) -> None:
self.last_tick = datetime.now(timezone.utc)
try:
results = await self.manager.sync_all_enabled()
for result in results:
if not isinstance(result, Exception):
continue
# SyncAlreadyRunning is an expected, benign outcome when a
# manual sync and a scheduler tick overlap for the same user
# -- not an error worth logging.
if isinstance(result, SyncAlreadyRunning):
continue
logger.warning(
"sync_all_enabled: user sync failed during scheduled tick: %s: %s",
type(result).__name__,
str(result)[:200],
)
except Exception:
logger.exception("sync_all_enabled failed during scheduled tick")
finally:
self.next_tick = datetime.now(timezone.utc) + timedelta(seconds=self.interval_seconds)
async def _run(self) -> None:
while not self._stop.is_set():
await self.run_once()
try:
await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds)
except TimeoutError:
pass
async def start(self) -> None:
self._stop.clear()
self._task = asyncio.create_task(self._run())
async def stop(self) -> None:
self._stop.set()
if self._task is not None:
await self._task
self._task = None

12
app/sync/states.py Normal file
View File

@@ -0,0 +1,12 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class SyncOutcome:
user_id: int
status: str
discovered: int
imported: int
skipped: int
failed: int
message: str | None = None

114
app/web/operations.py Normal file
View File

@@ -0,0 +1,114 @@
from fastapi import APIRouter, Form, HTTPException, Request
from fastapi.responses import HTMLResponse
from sqlalchemy import func, select
from app.auth.admin import require_admin
from app.auth.csrf import ensure_csrf_token, validate_csrf
from app.db.models import Activity
from app.db.repositories import ActivityRepository, UserRepository
from app.sync.manager import SyncAlreadyRunning
from app.web.routes import templates
router = APIRouter()
APP_VERSION = "1.0.0"
def _normalize_outcome(item):
if isinstance(item, Exception):
return {
"status": "error",
"user_id": None,
"discovered": 0,
"imported": 0,
"skipped": 0,
"failed": 0,
"message": str(item),
}
return {
"status": item.status,
"user_id": item.user_id,
"discovered": item.discovered,
"imported": item.imported,
"skipped": item.skipped,
"failed": item.failed,
"message": item.message,
}
@router.post("/users/{user_id}/sync", response_class=HTMLResponse)
async def manual_sync(request: Request, user_id: int, csrf_token: str = Form(...)):
require_admin(request)
validate_csrf(request, csrf_token)
try:
outcome = await request.app.state.sync_manager.sync_user(user_id)
except SyncAlreadyRunning:
return HTMLResponse("Sync already running for this user", status_code=409)
return templates.TemplateResponse(
request, "fragments/sync_result.html", {"outcomes": [_normalize_outcome(outcome)]}
)
@router.post("/sync-all", response_class=HTMLResponse)
async def manual_sync_all(request: Request, csrf_token: str = Form(...)):
require_admin(request)
validate_csrf(request, csrf_token)
outcomes = await request.app.state.sync_manager.sync_all_enabled()
return templates.TemplateResponse(
request, "fragments/sync_result.html", {"outcomes": [_normalize_outcome(o) for o in outcomes]}
)
@router.post("/users/{user_id}/garmin-mfa", response_class=HTMLResponse)
async def garmin_mfa(request: Request, user_id: int, csrf_token: str = Form(...), code: str = Form(...)):
require_admin(request)
validate_csrf(request, csrf_token)
stripped = code.strip()
if not stripped or len(stripped) > 20:
raise HTTPException(status_code=400, detail="Invalid MFA code")
try:
outcome = await request.app.state.sync_manager.sync_user(user_id, mfa_code=stripped)
except SyncAlreadyRunning:
return HTMLResponse("Sync already running for this user", status_code=409)
return templates.TemplateResponse(
request, "fragments/sync_result.html", {"outcomes": [_normalize_outcome(outcome)]}
)
@router.post("/activities/{activity_id}/retry", response_class=HTMLResponse)
async def retry_activity(request: Request, activity_id: int, csrf_token: str = Form(...)):
require_admin(request)
validate_csrf(request, csrf_token)
with request.app.state.session_factory() as session:
activity_repo = ActivityRepository(session)
try:
activity = activity_repo.reset_retryable_failure(activity_id)
except ValueError:
return HTMLResponse("Activity is not retryable", status_code=409)
user_id = activity.user_id
try:
outcome = await request.app.state.sync_manager.sync_user(user_id)
except SyncAlreadyRunning:
return HTMLResponse("Sync already running for this user", status_code=409)
return templates.TemplateResponse(
request, "fragments/sync_result.html", {"outcomes": [_normalize_outcome(outcome)]}
)
@router.get("/system", response_class=HTMLResponse)
def system_page(request: Request):
require_admin(request)
settings = request.app.state.settings
scheduler = request.app.state.scheduler
with request.app.state.session_factory() as session:
user_count = len(UserRepository(session).list_all())
activity_count = session.scalar(select(func.count()).select_from(Activity)) or 0
return templates.TemplateResponse(request, "system.html", {
"csrf_token": ensure_csrf_token(request),
"app_version": APP_VERSION,
"sync_interval_minutes": settings.sync_interval_minutes,
"last_tick": scheduler.last_tick,
"next_tick": scheduler.next_tick,
"user_count": user_count,
"activity_count": activity_count,
})

View File

@@ -7,7 +7,7 @@ from fastapi.templating import Jinja2Templates
from app.auth.admin import password_matches, require_admin
from app.auth.csrf import ensure_csrf_token, validate_csrf
from app.db.models import SyncUser
from app.db.repositories import UserRepository
from app.db.repositories import ActivityRepository, UserRepository
from app.security.credentials import CredentialCipher
from app.web.forms import UserFormData
@@ -62,11 +62,11 @@ def login(
def dashboard(request: Request):
require_admin(request)
with request.app.state.session_factory() as session:
users = UserRepository(session).list_all()
rows = UserRepository(session).dashboard_rows()
return templates.TemplateResponse(
request,
"dashboard.html",
{"users": users, "csrf_token": ensure_csrf_token(request)},
{"rows": rows, "csrf_token": ensure_csrf_token(request)},
)
@@ -131,12 +131,14 @@ def user_detail(request: Request, user_id: int):
require_admin(request)
with request.app.state.session_factory() as session:
user = _get_user_or_404(UserRepository(session), user_id)
activities = ActivityRepository(session).list_pending_for_user(user_id)
return templates.TemplateResponse(
request,
"users/detail.html",
{
"csrf_token": ensure_csrf_token(request),
"user": user,
"activities": activities,
},
)

View File

@@ -4,13 +4,28 @@
{% block content %}
<h1>Dashboard</h1>
<p><a href="/users/new">Add user</a></p>
<p><a href="/users/new">Add user</a> | <a href="/system">System</a></p>
<form method="post" action="/sync-all">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit">Sync all now</button>
</form>
<ul>
{% for user in users %}
{% for row in rows %}
<li>
<a href="/users/{{ user.id }}">{{ user.name }}</a>
&mdash; {{ "enabled" if user.enabled else "disabled" }}
&mdash; {{ user.health_state.value }}
<a href="/users/{{ row.id }}">{{ row.name }}</a>
&mdash; {{ "enabled" if row.enabled else "disabled" }}
&mdash; {{ row.health_state }}
&mdash; last sync: {{ row.last_sync_at or "-" }}
&mdash; last activity: {{ row.last_activity_name or "-" }} ({{ row.last_activity_status or "-" }})
{% if row.action_reason == "garmin_mfa_required" %}
<span class="action-required">Garmin MFA required — <a href="/users/{{ row.id }}">resolve</a></span>
{% endif %}
<form method="post" action="/users/{{ row.id }}/sync" style="display:inline">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit">Sync now</button>
</form>
</li>
{% else %}
<li>No users yet.</li>

View File

@@ -0,0 +1,5 @@
<form method="post" action="/users/{{ user.id }}/garmin-mfa">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<label>Garmin MFA code: <input type="text" name="code" maxlength="20" required></label>
<button type="submit">Submit code</button>
</form>

View File

@@ -0,0 +1,17 @@
<ul>
{% for outcome in outcomes %}
<li>
User {{ outcome.user_id if outcome.user_id is not none else "unknown" }}:
status={{ outcome.status }}
discovered={{ outcome.discovered }}
imported={{ outcome.imported }}
skipped={{ outcome.skipped }}
failed={{ outcome.failed }}
{% if outcome.message %}
&mdash; {{ outcome.message }}
{% endif %}
</li>
{% else %}
<li>No outcomes.</li>
{% endfor %}
</ul>

View File

@@ -0,0 +1,33 @@
{% extends "base.html" %}
{% block title %}System - MyWhoosh Garmin Sync{% endblock %}
{% block content %}
<h1>System</h1>
<p><a href="/">Back to dashboard</a></p>
<dl>
<dt>Application version</dt>
<dd>{{ app_version }}</dd>
<dt>Sync interval (minutes)</dt>
<dd>{{ sync_interval_minutes }}</dd>
<dt>Last scheduler tick</dt>
<dd>{{ last_tick or "-" }}</dd>
<dt>Next scheduler tick</dt>
<dd>{{ next_tick or "-" }}</dd>
<dt>User count</dt>
<dd>{{ user_count }}</dd>
<dt>Activity count</dt>
<dd>{{ activity_count }}</dd>
</dl>
<form method="post" action="/sync-all">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit">Sync all now</button>
</form>
{% endblock %}

View File

@@ -28,4 +28,26 @@
<dt>Updated at</dt>
<dd>{{ user.updated_at }}</dd>
</dl>
{% if user.action_reason == "garmin_mfa_required" %}
<h2>Garmin MFA required</h2>
{% include "fragments/mfa_form.html" %}
{% endif %}
<h2>Activities</h2>
<ul>
{% for activity in activities %}
<li>
{{ activity.activity_name }} &mdash; {{ activity.status.value }}
{% if activity.status.value == "failed" and activity.retryable %}
<form method="post" action="/activities/{{ activity.id }}/retry" style="display:inline">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit">Retry</button>
</form>
{% endif %}
</li>
{% else %}
<li>No pending activities.</li>
{% endfor %}
</ul>
{% endblock %}

View File

@@ -16,7 +16,7 @@ dependencies = [
"python-multipart>=0.0.9,<1",
"itsdangerous>=2.1,<3",
"httpx>=0.27,<1",
"garminconnect>=0.2,<1",
"garminconnect>=0.3.10,<1",
]
[project.optional-dependencies]

View File

@@ -8,8 +8,8 @@ from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.config import Settings
from app.db.models import Base
from app.db.repositories import ActivityRepository, UserRepository
from app.db.models import Activity, Base, HealthState
from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository
from app.main import create_app
@@ -40,7 +40,12 @@ def activity_repository(db_session: Session) -> ActivityRepository:
@pytest.fixture
def client(tmp_path: Path) -> TestClient:
def sync_run_repository(db_session: Session) -> SyncRunRepository:
return SyncRunRepository(db_session)
@pytest.fixture
def app(tmp_path: Path):
settings = Settings(
ADMIN_PASSWORD="admin-secret",
SECRET_KEY="0123456789abcdef0123456789abcdef",
@@ -49,8 +54,80 @@ def client(tmp_path: Path) -> TestClient:
DATABASE_URL=f"sqlite:///{tmp_path / 'app.db'}",
SYNC_INTERVAL_MINUTES=5,
)
app = create_app(settings)
application = create_app(settings)
try:
yield TestClient(app)
yield application
finally:
app.state.db_engine.dispose()
application.state.db_engine.dispose()
@pytest.fixture
def client(app) -> TestClient:
return TestClient(app)
def _extract_csrf(html: str) -> str:
marker = 'name="csrf_token" value="'
start = html.index(marker) + len(marker)
end = html.index('"', start)
return html[start:end]
class FakeSyncManager:
def __init__(self) -> None:
self.user_calls: list[int] = []
self.all_calls = 0
self.raise_already_running = False
self.mfa_calls: list[tuple[int, str]] = []
async def sync_user(self, user_id: int, mfa_code: str | None = None):
if self.raise_already_running:
from app.sync.manager import SyncAlreadyRunning
raise SyncAlreadyRunning(f"sync already running for user {user_id}")
self.user_calls.append(user_id)
if mfa_code is not None:
self.mfa_calls.append((user_id, mfa_code))
from app.sync.states import SyncOutcome
return SyncOutcome(user_id=user_id, status="success", discovered=0, imported=0, skipped=0, failed=0)
async def sync_all_enabled(self):
self.all_calls += 1
return []
@pytest.fixture
def fake_sync_manager() -> FakeSyncManager:
return FakeSyncManager()
@pytest.fixture
def authenticated_client(app, client: TestClient, fake_sync_manager: FakeSyncManager) -> TestClient:
page = client.get("/login")
csrf = _extract_csrf(page.text)
response = client.post("/login", data={"password": "admin-secret", "csrf_token": csrf}, follow_redirects=False)
assert response.status_code == 303
app.state.sync_manager = fake_sync_manager
client.csrf_token = csrf
return client
@pytest.fixture
def seeded_activity(db_session: Session, user_repository: UserRepository, activity_repository: ActivityRepository) -> Activity:
user = user_repository.create(
name="Test User",
enabled=True,
health_state=HealthState.HEALTHY,
mywhoosh_email_enc="test@example.com",
mywhoosh_password_enc="password",
garmin_email_enc="test@garmin.com",
garmin_password_enc="garmin_password",
)
activity, _ = activity_repository.get_or_create_discovered(
user_id=user.id,
mywhoosh_activity_id="mw-test-123",
activity_name="Test Activity",
activity_timestamp=None,
)
return activity

132
tests/db/test_sync_state.py Normal file
View File

@@ -0,0 +1,132 @@
from app.db.models import ActivityStatus, SyncRunStatus
def test_failure_retains_last_completed_stage(activity_repository, seeded_activity) -> None:
activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
activity_repository.mark_failed(seeded_activity.id, "Garmin timeout", retryable=True)
activity = activity_repository.get(seeded_activity.id)
assert activity.status == ActivityStatus.FAILED
assert activity.last_completed_stage == ActivityStatus.DOWNLOADED
assert activity.retryable is True
def test_converted_activity_is_pending_until_terminal(activity_repository, seeded_activity) -> None:
activity_repository.mark_converted(seeded_activity.id, "/data/activities/1/a/converted.fit")
ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
assert seeded_activity.id in ids
def test_list_pending_excludes_imported(activity_repository, seeded_activity) -> None:
activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
activity_repository.mark_converted(seeded_activity.id, "/data/activities/1/a/converted.fit")
activity_repository.mark_imported(seeded_activity.id, "garmin-123")
ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
assert seeded_activity.id not in ids
def test_list_pending_excludes_duplicate(activity_repository, seeded_activity) -> None:
activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
activity_repository.mark_duplicate(seeded_activity.id)
ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
assert seeded_activity.id not in ids
def test_list_pending_excludes_non_retryable_failed(activity_repository, seeded_activity) -> None:
activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
activity_repository.mark_failed(seeded_activity.id, "Cannot retry", retryable=False)
ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
assert seeded_activity.id not in ids
def test_list_pending_includes_retryable_failed(activity_repository, seeded_activity) -> None:
activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
activity_repository.mark_failed(seeded_activity.id, "Temporary error", retryable=True)
ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
assert seeded_activity.id in ids
def test_sync_run_start_and_finish(sync_run_repository, user_repository) -> None:
user = user_repository.create(
name="Test User",
enabled=True,
health_state="healthy",
mywhoosh_email_enc="test@example.com",
mywhoosh_password_enc="password",
garmin_email_enc="test@garmin.com",
garmin_password_enc="garmin_password",
)
sync_run = sync_run_repository.start(user.id)
assert sync_run.status == SyncRunStatus.RUNNING
assert sync_run.discovered_count == 0
assert sync_run.imported_count == 0
assert sync_run.skipped_count == 0
assert sync_run.failed_count == 0
finished = sync_run_repository.finish(
sync_run.id,
status=SyncRunStatus.SUCCESS,
discovered=5,
imported=3,
skipped=1,
failed=1,
)
assert finished.status == SyncRunStatus.SUCCESS
assert finished.discovered_count == 5
assert finished.imported_count == 3
assert finished.skipped_count == 1
assert finished.failed_count == 1
assert finished.finished_at is not None
# Reload from DB to verify persisted
reloaded = sync_run_repository.get(sync_run.id)
assert reloaded.status == SyncRunStatus.SUCCESS
assert reloaded.discovered_count == 5
assert reloaded.imported_count == 3
def test_sync_run_finish_with_error(sync_run_repository, user_repository) -> None:
user = user_repository.create(
name="Test User",
enabled=True,
health_state="healthy",
mywhoosh_email_enc="test@example.com",
mywhoosh_password_enc="password",
garmin_email_enc="test@garmin.com",
garmin_password_enc="garmin_password",
)
sync_run = sync_run_repository.start(user.id)
error_msg = "Connection timeout"
finished = sync_run_repository.finish(
sync_run.id,
status=SyncRunStatus.FAILED,
discovered=0,
imported=0,
skipped=0,
failed=0,
summary_error=error_msg,
)
assert finished.summary_error == error_msg
# Verify truncation works
long_error = "x" * 5000
finished_long = sync_run_repository.finish(
sync_run.id,
status=SyncRunStatus.FAILED,
discovered=0,
imported=0,
skipped=0,
failed=0,
summary_error=long_error,
)
assert len(finished_long.summary_error) == 2000
assert finished_long.summary_error == long_error[:2000]

View File

@@ -2,7 +2,13 @@ from pathlib import Path
import pytest
from app.garmin.uploader import GarminUploadBlocked, GarminUploader
from app.garmin.uploader import (
GarminAuthError,
GarminImportRejected,
GarminTransientError,
GarminUploadBlocked,
GarminUploader,
)
class FakeGarmin:
@@ -78,3 +84,93 @@ def test_mfa_code_is_returned_only_to_prompt(tmp_path: Path) -> None:
)
uploader.import_fit(tmp_path / "ride.fit", mfa_code="123456")
assert seen == ["123456"]
def test_import_rejected_by_garmin_raises_garmin_import_rejected(tmp_path: Path) -> None:
rejected_response = {
"detailedImportResult": {
"successes": [],
"failures": [{"internalId": 1, "messages": ["Invalid FIT file"]}],
}
}
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_result=rejected_response),
)
with pytest.raises(GarminImportRejected):
uploader.import_fit(tmp_path / "ride.fit")
def test_login_failure_with_429_is_transient(tmp_path: Path) -> None:
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("429 too many requests")),
)
with pytest.raises(GarminTransientError):
uploader.import_fit(tmp_path / "ride.fit")
def test_login_failure_with_timeout_is_transient(tmp_path: Path) -> None:
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("connection timeout")),
)
with pytest.raises(GarminTransientError):
uploader.import_fit(tmp_path / "ride.fit")
def test_login_failure_with_credential_message_is_auth_error(tmp_path: Path) -> None:
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("invalid credential")),
)
with pytest.raises(GarminAuthError):
uploader.import_fit(tmp_path / "ride.fit")
def test_login_failure_unrecognized_is_transient(tmp_path: Path) -> None:
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("something odd happened")),
)
with pytest.raises(GarminTransientError):
uploader.import_fit(tmp_path / "ride.fit")
def test_import_time_401_raises_auth_error(tmp_path: Path) -> None:
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_error=RuntimeError("401 unauthorized")),
)
with pytest.raises(GarminAuthError):
uploader.import_fit(tmp_path / "ride.fit")
def test_mfa_wrapped_in_generic_exception_still_blocks(tmp_path: Path) -> None:
class WrappedMfaGarmin(FakeGarmin):
def login(self, tokenstore=None):
try:
self.prompt_mfa()
except GarminUploadBlocked as exc:
raise RuntimeError(f"Login failed: Garmin requested MFA ({exc})") from exc
uploader = GarminUploader(
email="g@example.com",
password="pw",
tokenstore=tmp_path / "garmin",
client_factory=WrappedMfaGarmin,
)
with pytest.raises(GarminUploadBlocked):
uploader.import_fit(tmp_path / "ride.fit")

View File

@@ -134,7 +134,7 @@ async def test_download_fit_fetches_signed_url_bytes(tmp_path) -> None:
@pytest.mark.asyncio
async def test_list_activities_raises_integration_error_on_missing_stable_id(tmp_path) -> None:
async def test_list_activities_skips_row_missing_stable_id(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/activities"):
return httpx.Response(
@@ -147,7 +147,13 @@ async def test_list_activities_raises_integration_error_on_missing_stable_id(tmp
"title": "Ride without id",
"activityFileId": "f-1",
"startDatetime": "2026-08-15T06:00:00.000Z",
}
},
{
"id": "a-2",
"title": "Ride with id",
"activityFileId": "f-2",
"startDatetime": "2026-08-15T06:00:00.000Z",
},
],
}
},
@@ -158,5 +164,108 @@ async def test_list_activities_raises_integration_error_on_missing_stable_id(tmp
store.save(MyWhooshToken(access_token="cached", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
activities = await client.list_activities("rider@example.com", "secret")
assert [a.id for a in activities] == ["a-2"]
@pytest.mark.asyncio
async def test_list_activities_skips_row_with_unparseable_start_datetime(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/activities"):
return httpx.Response(
200,
json={
"data": {
"totalPages": 1,
"results": [
{
"id": "a-1",
"title": "Ride with bad date",
"activityFileId": "f-1",
"startDatetime": "not-a-date",
},
{
"id": "a-2",
"title": "Ride with good date",
"activityFileId": "f-2",
"startDatetime": "2026-08-15T06:00:00.000Z",
},
],
}
},
)
raise AssertionError(request.url)
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
store.save(MyWhooshToken(access_token="cached", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
activities = await client.list_activities("rider@example.com", "secret")
assert [a.id for a in activities] == ["a-2"]
@pytest.mark.asyncio
async def test_list_activities_raises_integration_error_on_envelope_shape_failure(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/activities"):
return httpx.Response(200, json={"data": {"totalPages": 1}})
raise AssertionError(request.url)
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
store.save(MyWhooshToken(access_token="cached", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
with pytest.raises(MyWhooshIntegrationError):
await client.list_activities("rider@example.com", "secret")
@pytest.mark.asyncio
async def test_list_activities_respects_max_pages(tmp_path) -> None:
calls = []
async def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/activities"):
payload = json.loads(request.content)
page = payload["page"]
calls.append(page)
result = {
"data": {
"totalPages": 5,
"results": [
{
"id": f"a-{page}",
"title": f"Ride {page}",
"activityFileId": f"f-{page}",
"startDatetime": "2026-08-15T06:00:00.000Z",
}
],
}
}
return httpx.Response(200, json=result)
raise AssertionError(request.url)
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
store.save(MyWhooshToken(access_token="cached", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
activities = await client.list_activities("rider@example.com", "secret", max_pages=1)
assert calls == [1]
assert [a.id for a in activities] == ["a-1"]
@pytest.mark.asyncio
async def test_download_fit_raises_integration_error_on_invalid_json(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/download-activity-file"):
return httpx.Response(200, content=b"not json", headers={"content-type": "application/json"})
raise AssertionError(request.url)
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
store.save(MyWhooshToken(access_token="cached", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
with pytest.raises(MyWhooshIntegrationError):
await client.download_fit("f-1", "rider@example.com", "secret")

View File

@@ -1,7 +1,12 @@
import httpx
import pytest
from app.mywhoosh.client import MyWhooshClient, MyWhooshAuthError
from app.mywhoosh.client import (
MyWhooshAuthError,
MyWhooshClient,
MyWhooshIntegrationError,
MyWhooshTransientError,
)
from app.mywhoosh.models import MyWhooshToken
from app.mywhoosh.tokenstore import MyWhooshTokenStore
@@ -37,3 +42,93 @@ async def test_invalid_credentials_raise_auth_error(tmp_path) -> None:
)
with pytest.raises(MyWhooshAuthError):
await client.login("rider@example.com", "bad")
@pytest.mark.asyncio
async def test_login_returns_json_array_raises_integration_error(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=["not", "an", "object"])
client = MyWhooshClient(
MyWhooshTokenStore(tmp_path / "mywhoosh.json"),
http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
)
with pytest.raises(MyWhooshIntegrationError):
await client.login("rider@example.com", "bad")
@pytest.mark.asyncio
async def test_login_5xx_raises_transient_error(tmp_path) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(503, text="Service Unavailable")
client = MyWhooshClient(
MyWhooshTokenStore(tmp_path / "mywhoosh.json"),
http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
)
with pytest.raises(MyWhooshTransientError):
await client.login("rider@example.com", "secret")
@pytest.mark.asyncio
async def test_second_401_after_reauth_raises_auth_error(tmp_path) -> None:
login_call_count = 0
async def handler(request: httpx.Request) -> httpx.Response:
nonlocal login_call_count
if request.url.path.endswith("/activities"):
return httpx.Response(401, json={"message": "expired"})
if request.url.path.endswith("/login"):
login_call_count += 1
return httpx.Response(
200,
json={
"Success": True,
"AccessToken": "fresh-access",
"RefreshToken": "fresh-refresh",
"WhooshId": "w-1",
},
)
raise AssertionError(request.url)
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
store.save(MyWhooshToken(access_token="stale", refresh_token=None, whoosh_id=None))
client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
with pytest.raises(MyWhooshAuthError):
await client.list_activities("rider@example.com", "secret")
assert login_call_count == 1
@pytest.mark.asyncio
async def test_aclose_closes_self_owned_http_client(tmp_path) -> None:
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
client = MyWhooshClient(store)
await client.aclose()
assert client.http.is_closed is True
@pytest.mark.asyncio
async def test_aclose_does_not_close_injected_http_client(tmp_path) -> None:
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
injected = httpx.AsyncClient()
client = MyWhooshClient(store, http_client=injected)
await client.aclose()
assert injected.is_closed is False
await injected.aclose()
@pytest.mark.asyncio
async def test_client_usable_as_async_context_manager(tmp_path) -> None:
store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
async with MyWhooshClient(store) as client:
http_client = client.http
assert http_client.is_closed is False
assert http_client.is_closed is True

View File

@@ -16,3 +16,28 @@ def test_tokenstore_round_trip_and_permissions(tmp_path: Path) -> None:
def test_missing_token_returns_none(tmp_path: Path) -> None:
store = MyWhooshTokenStore(tmp_path / "missing.json")
assert store.load() is None
def test_corrupt_token_file_returns_none(tmp_path: Path) -> None:
path = tmp_path / "mywhoosh.json"
path.write_bytes(b"not valid json {{{")
store = MyWhooshTokenStore(path)
assert store.load() is None
def test_token_file_missing_access_token_returns_none(tmp_path: Path) -> None:
path = tmp_path / "mywhoosh.json"
path.write_text('{"refresh_token": "r", "whoosh_id": "w"}', encoding="utf-8")
store = MyWhooshTokenStore(path)
assert store.load() is None
def test_clear_removes_token_and_load_returns_none(tmp_path: Path) -> None:
store = MyWhooshTokenStore(tmp_path / "tokens" / "mywhoosh.json")
token = MyWhooshToken(access_token="access", refresh_token="refresh", whoosh_id="whoosh-1")
store.save(token)
store.clear()
assert store.load() is None
assert not store.path.exists()

0
tests/sync/__init__.py Normal file
View File

200
tests/sync/conftest.py Normal file
View File

@@ -0,0 +1,200 @@
from pathlib import Path
from typing import Callable
import pytest
from cryptography.fernet import Fernet
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.db.models import Activity, ActivityStatus, Base, SyncUser
from app.db.repositories import ActivityRepository, UserRepository
from app.mywhoosh.models import MyWhooshActivity
from app.security.credentials import CredentialCipher
from app.sync.manager import SyncManager
from tests.sync.fakes import FakeGarminUploader, FakeMyWhooshClient
class FakeFitConverter:
"""Fit converter stub that mimics convert_fit_device's side effect of
writing bytes to output_path, without doing any real FIT parsing."""
def __init__(self) -> None:
self.calls = 0
def __call__(self, source_path: Path, output_path: Path):
self.calls += 1
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(b"fake-fit-bytes")
return None
class StubSettings:
"""Minimal stand-in for app.config.Settings exposing only the two
properties SyncManager needs; avoids constructing a full Settings with
its several required env-backed fields."""
def __init__(self, tmp_path: Path) -> None:
self.tokens_dir = tmp_path / "tokens"
self.activities_dir = tmp_path / "activities"
@pytest.fixture
def session_factory():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
try:
yield factory
finally:
engine.dispose()
@pytest.fixture
def cipher() -> CredentialCipher:
return CredentialCipher(Fernet.generate_key().decode("ascii"))
@pytest.fixture
def settings(tmp_path: Path) -> StubSettings:
return StubSettings(tmp_path)
@pytest.fixture
def load_only_activity(session_factory) -> Callable[[int], Activity]:
def _load(user_id: int) -> Activity:
with session_factory() as session:
activities = list(session.scalars(select(Activity).where(Activity.user_id == user_id)))
assert len(activities) == 1, f"expected exactly one activity for user {user_id}, found {len(activities)}"
return activities[0]
return _load
def _create_user(session: Session, cipher: CredentialCipher) -> SyncUser:
return UserRepository(session).create(
name="Test User",
enabled=True,
mywhoosh_email_enc=cipher.encrypt("mywhoosh@example.com"),
mywhoosh_password_enc=cipher.encrypt("mywhoosh-pass"),
garmin_email_enc=cipher.encrypt("garmin@example.com"),
garmin_password_enc=cipher.encrypt("garmin-pass"),
)
@pytest.fixture
def seeded_user(session_factory, cipher: CredentialCipher) -> SyncUser:
with session_factory() as session:
return _create_user(session, cipher)
@pytest.fixture
def manager_factory(session_factory, cipher: CredentialCipher, settings: StubSettings):
"""Build a SyncManager plus its injected fakes, wired so the fake
MyWhoosh client's single remote activity matches the given (already
seeded) Activity's mywhoosh_activity_id -- so get_or_create_discovered
resolves to the existing row instead of creating a new one."""
def _factory(activity: Activity):
remote = MyWhooshActivity(
id=activity.mywhoosh_activity_id,
title=activity.activity_name,
activity_file_id=f"file-{activity.mywhoosh_activity_id}",
started_at=activity.activity_timestamp,
)
mywhoosh = FakeMyWhooshClient(activities=[remote], fit_bytes=b"source-bytes")
converter = FakeFitConverter()
garmin = FakeGarminUploader()
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
return manager, mywhoosh, converter, garmin
return _factory
@pytest.fixture
def manager(session_factory, cipher: CredentialCipher, settings: StubSettings, seeded_user: SyncUser):
"""A manager wired for the happy-path new-activity scenario: one remote
MyWhoosh activity that seeded_user has never seen before."""
remote = MyWhooshActivity(
id="mw-1",
title="Morning Ride",
activity_file_id="file-mw-1",
started_at=None,
)
mywhoosh = FakeMyWhooshClient(activities=[remote], fit_bytes=b"source-bytes")
converter = FakeFitConverter()
garmin = FakeGarminUploader()
sync_manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
# Exposed for tests that want to introspect fakes without a
# manager_factory-style scenario.
sync_manager.fake_mywhoosh = mywhoosh
sync_manager.fake_converter = converter
sync_manager.fake_garmin = garmin
return sync_manager
@pytest.fixture
def seeded_activity_factory(session_factory, cipher: CredentialCipher):
def _factory(
*,
status: ActivityStatus,
last_completed_stage: ActivityStatus,
retryable: bool,
mywhoosh_activity_id: str = "mw-1",
) -> Activity:
with session_factory() as session:
user = _create_user(session, cipher)
activity_repo = ActivityRepository(session)
activity, _created = activity_repo.get_or_create_discovered(
user_id=user.id,
mywhoosh_activity_id=mywhoosh_activity_id,
activity_name="Test Activity",
activity_timestamp=None,
)
activity.status = status
activity.last_completed_stage = last_completed_stage
activity.retryable = retryable
if status in (
ActivityStatus.DOWNLOADED,
ActivityStatus.CONVERTED,
ActivityStatus.IMPORTED,
ActivityStatus.DUPLICATE,
) or last_completed_stage in (
ActivityStatus.DOWNLOADED,
ActivityStatus.CONVERTED,
ActivityStatus.IMPORTED,
ActivityStatus.DUPLICATE,
):
activity.source_fit_path = "seed-source.fit"
if status in (
ActivityStatus.CONVERTED,
ActivityStatus.IMPORTED,
ActivityStatus.DUPLICATE,
) or last_completed_stage in (
ActivityStatus.CONVERTED,
ActivityStatus.IMPORTED,
ActivityStatus.DUPLICATE,
):
activity.converted_fit_path = "seed-converted.fit"
session.commit()
return activity
return _factory

30
tests/sync/fakes.py Normal file
View File

@@ -0,0 +1,30 @@
from app.garmin.uploader import UploadResult
class FakeMyWhooshClient:
def __init__(self, activities, fit_bytes: bytes) -> None:
self.activities = activities
self.fit_bytes = fit_bytes
self.list_calls = 0
self.download_calls = 0
async def list_activities(self, email: str, password: str):
self.list_calls += 1
return list(self.activities)
async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes:
self.download_calls += 1
return self.fit_bytes
class FakeGarminUploader:
def __init__(self, result: UploadResult | None = None, error: Exception | None = None) -> None:
self.result = result or UploadResult("imported", False, "g-1", {"activityId": "g-1"})
self.error = error
self.calls = 0
def import_fit(self, fit_path, mfa_code=None):
self.calls += 1
if self.error is not None:
raise self.error
return self.result

View File

@@ -0,0 +1,163 @@
import asyncio
import pytest
from app.db.repositories import UserRepository
from app.mywhoosh.client import MyWhooshAuthError
from app.mywhoosh.models import MyWhooshActivity
from app.sync.manager import SyncAlreadyRunning, SyncManager
from tests.sync.conftest import FakeFitConverter, _create_user
from tests.sync.fakes import FakeGarminUploader
class BlockingMyWhooshClient:
"""Fake MyWhoosh client whose list_activities() blocks on test-controlled
events, so a test can deterministically observe "sync has started but not
finished" without any production-only test hooks."""
def __init__(self, first_started: asyncio.Event, release: asyncio.Event) -> None:
self.first_started = first_started
self.release = release
self.list_calls = 0
async def list_activities(self, email: str, password: str):
self.list_calls += 1
self.first_started.set()
await self.release.wait()
return []
async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes:
raise AssertionError("download_fit should not be reached in this test")
class ConditionalFailureMyWhooshClient:
"""Fake MyWhoosh client that raises MyWhooshAuthError only for a specific
account email, letting one user's sync fail while others succeed."""
def __init__(self, activities, fit_bytes: bytes, failing_email: str) -> None:
self.activities = activities
self.fit_bytes = fit_bytes
self.failing_email = failing_email
self.list_calls = 0
self.download_calls = 0
async def list_activities(self, email: str, password: str):
self.list_calls += 1
if email == self.failing_email:
raise MyWhooshAuthError("simulated auth failure")
return list(self.activities)
async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes:
self.download_calls += 1
return self.fit_bytes
@pytest.fixture
def user_a(session_factory, cipher):
with session_factory() as session:
return _create_user(session, cipher)
@pytest.fixture
def user_b(session_factory, cipher):
with session_factory() as session:
return _create_user(session, cipher)
@pytest.mark.asyncio
async def test_same_user_cannot_run_twice(session_factory, cipher, settings, seeded_user) -> None:
first_started = asyncio.Event()
release_first = asyncio.Event()
mywhoosh = BlockingMyWhooshClient(first_started, release_first)
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: FakeGarminUploader(),
fit_converter=FakeFitConverter(),
)
first = asyncio.create_task(manager.sync_user(seeded_user.id))
await first_started.wait()
with pytest.raises(SyncAlreadyRunning):
await manager.sync_user(seeded_user.id)
release_first.set()
outcome = await first
assert outcome.user_id == seeded_user.id
assert mywhoosh.list_calls == 1
@pytest.mark.asyncio
async def test_different_users_can_run_concurrently(manager, user_a, user_b) -> None:
results = await asyncio.gather(manager.sync_user(user_a.id), manager.sync_user(user_b.id))
assert {result.user_id for result in results} == {user_a.id, user_b.id}
assert all(result.status == "success" for result in results)
@pytest.mark.asyncio
async def test_sync_all_enabled_isolates_failures(session_factory, cipher, settings) -> None:
with session_factory() as session:
failing_user = UserRepository(session).create(
name="Failing User",
enabled=True,
mywhoosh_email_enc=cipher.encrypt("failing-mw@example.com"),
mywhoosh_password_enc=cipher.encrypt("failing-mw-pass"),
garmin_email_enc=cipher.encrypt("failing-garmin@example.com"),
garmin_password_enc=cipher.encrypt("failing-garmin-pass"),
)
healthy_user = UserRepository(session).create(
name="Healthy User",
enabled=True,
mywhoosh_email_enc=cipher.encrypt("healthy-mw@example.com"),
mywhoosh_password_enc=cipher.encrypt("healthy-mw-pass"),
garmin_email_enc=cipher.encrypt("healthy-garmin@example.com"),
garmin_password_enc=cipher.encrypt("healthy-garmin-pass"),
)
remote = MyWhooshActivity(
id="mw-shared",
title="Shared Ride",
activity_file_id="file-mw-shared",
started_at=None,
)
mywhoosh = ConditionalFailureMyWhooshClient(
activities=[remote],
fit_bytes=b"source-bytes",
failing_email="failing-mw@example.com",
)
converter = FakeFitConverter()
garmin = FakeGarminUploader()
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
results = await manager.sync_all_enabled()
assert len(results) == 2
for result in results:
assert not isinstance(result, Exception)
by_user = {result.user_id: result for result in results}
failing_outcome = by_user[failing_user.id]
healthy_outcome = by_user[healthy_user.id]
# The failing user's auth error is classified by _sync_user_locked's own
# exception handling and returned as a non-success SyncOutcome rather than
# raised -- so asyncio.gather never sees an exception for this failure
# mode. It must not affect the healthy user's independent outcome.
assert failing_outcome.status != "success"
assert failing_outcome.message is not None
assert failing_outcome.discovered == 0
assert healthy_outcome.status == "success"
assert healthy_outcome.imported == 1
assert healthy_outcome.failed == 0

218
tests/sync/test_manager.py Normal file
View File

@@ -0,0 +1,218 @@
from pathlib import Path
import pytest
from sqlalchemy import select
from app.db.models import ActivityStatus, HealthState, SyncRun, SyncRunStatus, SyncUser
from app.db.repositories import UserRepository
from app.garmin.uploader import UploadResult
from app.mywhoosh.models import MyWhooshActivity
from app.sync.manager import SyncManager
from tests.sync.conftest import FakeFitConverter, _create_user
from tests.sync.fakes import FakeMyWhooshClient
@pytest.mark.asyncio
async def test_new_activity_downloads_converts_and_imports(manager, seeded_user: SyncUser, load_only_activity) -> None:
outcome = await manager.sync_user(seeded_user.id)
assert outcome.discovered == 1
assert outcome.imported == 1
assert outcome.failed == 0
activity = load_only_activity(seeded_user.id)
assert activity.status == ActivityStatus.IMPORTED
assert Path(activity.source_fit_path).exists()
assert Path(activity.converted_fit_path).exists()
@pytest.mark.asyncio
@pytest.mark.parametrize(
("status", "last_stage", "expected_downloads", "expected_conversions", "expected_imports"),
[
(ActivityStatus.DOWNLOADED, ActivityStatus.DOWNLOADED, 0, 1, 1),
(ActivityStatus.CONVERTED, ActivityStatus.CONVERTED, 0, 0, 1),
(ActivityStatus.IMPORTED, ActivityStatus.IMPORTED, 0, 0, 0),
(ActivityStatus.FAILED, ActivityStatus.CONVERTED, 0, 0, 1),
],
)
async def test_resume_from_durable_stage(
manager_factory,
seeded_activity_factory,
status,
last_stage,
expected_downloads,
expected_conversions,
expected_imports,
) -> None:
activity = seeded_activity_factory(status=status, last_completed_stage=last_stage, retryable=True)
manager, mywhoosh, converter, garmin = manager_factory(activity)
await manager.sync_user(activity.user_id)
assert mywhoosh.download_calls == expected_downloads
assert converter.calls == expected_conversions
assert garmin.calls == expected_imports
@pytest.mark.asyncio
async def test_non_retryable_failed_activity_is_never_retried(
manager_factory,
seeded_activity_factory,
load_only_activity,
) -> None:
activity = seeded_activity_factory(
status=ActivityStatus.FAILED,
last_completed_stage=ActivityStatus.CONVERTED,
retryable=False,
)
manager, mywhoosh, converter, garmin = manager_factory(activity)
outcome = await manager.sync_user(activity.user_id)
assert mywhoosh.download_calls == 0
assert converter.calls == 0
assert garmin.calls == 0
assert outcome.imported == 0
assert outcome.skipped == 0
assert outcome.failed == 0
reloaded = load_only_activity(activity.user_id)
assert reloaded.status == ActivityStatus.FAILED
assert reloaded.retryable is False
@pytest.mark.asyncio
async def test_sync_run_repository_wiring_records_run(manager, seeded_user: SyncUser, session_factory) -> None:
await manager.sync_user(seeded_user.id)
with session_factory() as session:
runs = list(session.scalars(select(SyncRun).where(SyncRun.user_id == seeded_user.id)))
assert len(runs) == 1
run = runs[0]
assert run.status == SyncRunStatus.SUCCESS
assert run.discovered_count == 1
assert run.imported_count == 1
assert run.finished_at is not None
class RecordingGarminUploader:
"""Fake Garmin uploader that records the mfa_code it was actually called
with, so a test can prove the value genuinely threads through
SyncManager._sync_user_locked's asyncio.to_thread(garmin.import_fit, ...)
call rather than just through the web route's own fake manager."""
def __init__(self) -> None:
self.received_mfa_codes: list[str | None] = []
def import_fit(self, fit_path, mfa_code=None):
self.received_mfa_codes.append(mfa_code)
return UploadResult("imported", False, "g-1", {"activityId": "g-1"})
@pytest.mark.asyncio
async def test_garmin_action_required_not_cleared_by_run_with_no_garmin_work(
session_factory, cipher, settings
) -> None:
"""Regression test for the health-state-reset bug: a user stuck at
garmin_auth_required must NOT be silently cleared back to healthy just
because a run's MyWhoosh listing succeeded trivially (zero remote
activities means zero Garmin work was attempted -- no evidence Garmin
was actually fixed)."""
with session_factory() as session:
user = _create_user(session, cipher)
user.health_state = HealthState.ACTION_REQUIRED
user.garmin_state = "auth_required"
user.action_reason = "garmin_auth_required"
session.commit()
user_id = user.id
mywhoosh = FakeMyWhooshClient(activities=[], fit_bytes=b"unused")
converter = FakeFitConverter()
garmin = RecordingGarminUploader()
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
outcome = await manager.sync_user(user_id)
assert outcome.status == "success"
assert outcome.discovered == 0
with session_factory() as session:
reloaded = UserRepository(session).get(user_id)
assert reloaded.action_reason == "garmin_auth_required"
assert reloaded.health_state == HealthState.ACTION_REQUIRED
@pytest.mark.asyncio
async def test_garmin_action_required_cleared_after_successful_import(
session_factory, cipher, settings
) -> None:
"""Companion to the regression test above: the same starting
action_required/garmin_auth_required state IS cleared once this run
actually succeeds at a real Garmin import -- proving the fix doesn't just
always refuse to clear."""
with session_factory() as session:
user = _create_user(session, cipher)
user.health_state = HealthState.ACTION_REQUIRED
user.garmin_state = "auth_required"
user.action_reason = "garmin_auth_required"
session.commit()
user_id = user.id
remote = MyWhooshActivity(
id="mw-recovery-1", title="Recovery Ride", activity_file_id="file-mw-recovery-1", started_at=None
)
mywhoosh = FakeMyWhooshClient(activities=[remote], fit_bytes=b"source-bytes")
converter = FakeFitConverter()
garmin = RecordingGarminUploader()
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
outcome = await manager.sync_user(user_id)
assert outcome.status == "success"
assert outcome.imported == 1
with session_factory() as session:
reloaded = UserRepository(session).get(user_id)
assert reloaded.action_reason is None
assert reloaded.health_state == HealthState.HEALTHY
@pytest.mark.asyncio
async def test_mfa_code_reaches_real_garmin_uploader_via_sync_manager(
session_factory, cipher, settings, seeded_user: SyncUser
) -> None:
"""Proves mfa_code genuinely threads through _sync_user_locked's
asyncio.to_thread(garmin.import_fit, converted_path, mfa_code) call in the
real (non-web-route) code path -- tests/web/test_mfa.py already covers the
web route's own fake manager, but not the real SyncManager/GarminUploader
interface."""
remote = MyWhooshActivity(
id="mw-mfa-1", title="MFA Ride", activity_file_id="file-mw-mfa-1", started_at=None
)
mywhoosh = FakeMyWhooshClient(activities=[remote], fit_bytes=b"source-bytes")
converter = FakeFitConverter()
garmin = RecordingGarminUploader()
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=lambda token_store: mywhoosh,
garmin_factory=lambda email, password, tokenstore: garmin,
fit_converter=converter,
)
outcome = await manager.sync_user(seeded_user.id, mfa_code="123456")
assert outcome.status == "success"
assert garmin.received_mfa_codes == ["123456"]

View File

@@ -0,0 +1,42 @@
import asyncio
import pytest
from app.sync.scheduler import SyncScheduler
class FakeSyncManager:
def __init__(self, results=None) -> None:
self.results = results if results is not None else []
self.calls = 0
async def sync_all_enabled(self):
self.calls += 1
if self.results and isinstance(self.results[0], Exception):
raise self.results[0]
return list(self.results)
@pytest.mark.asyncio
async def test_scheduler_calls_sync_all_and_survives_failure() -> None:
fake = FakeSyncManager(results=[RuntimeError("one user failed")])
scheduler = SyncScheduler(fake, interval_seconds=0.01)
await scheduler.start()
await asyncio.sleep(0.035)
await scheduler.stop()
assert fake.calls >= 2
assert scheduler.last_tick is not None
@pytest.mark.asyncio
async def test_scheduler_stop_cancels_the_loop() -> None:
fake = FakeSyncManager()
scheduler = SyncScheduler(fake, interval_seconds=0.01)
await scheduler.start()
await asyncio.sleep(0.035)
await scheduler.stop()
assert scheduler._task is None
calls_after_stop = fake.calls
await asyncio.sleep(0.05)
assert fake.calls == calls_after_stop

268
tests/test_acceptance.py Normal file
View File

@@ -0,0 +1,268 @@
"""Application-level, multi-user acceptance tests for the full MyWhoosh -> Garmin
sync pipeline (Task 7 of the sync-scheduler-web plan).
These tests build a real `SyncManager` wired to a real SQLite database and a
real `CredentialCipher`, but with fake MyWhoosh/Garmin factories, so they
exercise the whole state machine end-to-end for two independent users without
touching any real external service.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from cryptography.fernet import Fernet
from sqlalchemy import func, select
from sqlalchemy.orm import sessionmaker
from app.db.models import Activity, ActivityStatus, HealthState, SyncUser
from app.db.repositories import UserRepository
from app.db.session import create_db_engine, create_session_factory, initialize_schema
from app.garmin.uploader import GarminUploadBlocked, UploadResult
from app.mywhoosh.models import MyWhooshActivity
from app.security.credentials import CredentialCipher
from app.sync.manager import SyncManager
class StubSettings:
"""Minimal stand-in for app.config.Settings exposing only the two
properties SyncManager needs."""
def __init__(self, tmp_path: Path) -> None:
self.tokens_dir = tmp_path / "tokens"
self.activities_dir = tmp_path / "activities"
class FakeMyWhooshClient:
def __init__(self, activities, fit_bytes: bytes) -> None:
self.activities = activities
self.fit_bytes = fit_bytes
self.list_calls = 0
self.download_calls = 0
async def list_activities(self, email, password):
self.list_calls += 1
return list(self.activities)
async def download_fit(self, activity_file_id, email, password):
self.download_calls += 1
return self.fit_bytes
class FakeGarminUploader:
def __init__(self, result: UploadResult | None = None, error: Exception | None = None) -> None:
self.result = result or UploadResult("imported", False, "g-1", {"activityId": "g-1"})
self.error = error
self.calls = 0
def import_fit(self, fit_path, mfa_code=None):
self.calls += 1
if self.error is not None:
raise self.error
return self.result
def fake_fit_converter(source_path: Path, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(b"fake-converted-fit-bytes")
def count_terminal_activities(session_factory: sessionmaker, user_id: int) -> int:
with session_factory() as session:
return session.scalar(
select(func.count())
.select_from(Activity)
.where(
Activity.user_id == user_id,
Activity.status.in_([ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE]),
)
)
def _create_user(session, cipher: CredentialCipher, *, name: str) -> SyncUser:
return UserRepository(session).create(
name=name,
enabled=True,
mywhoosh_email_enc=cipher.encrypt(f"{name.lower()}-mywhoosh@example.com"),
mywhoosh_password_enc=cipher.encrypt("mw-secret"),
garmin_email_enc=cipher.encrypt(f"{name.lower()}-garmin@example.com"),
garmin_password_enc=cipher.encrypt("garmin-secret"),
)
@pytest.fixture
def session_factory(tmp_path: Path):
db_path = tmp_path / "acceptance.db"
engine = create_db_engine(f"sqlite:///{db_path}")
initialize_schema(engine)
factory = create_session_factory(engine)
try:
yield factory
finally:
engine.dispose()
@pytest.fixture
def cipher() -> CredentialCipher:
return CredentialCipher(Fernet.generate_key().decode("ascii"))
@pytest.fixture
def settings(tmp_path: Path) -> StubSettings:
return StubSettings(tmp_path)
@pytest.fixture
def two_users(session_factory, cipher: CredentialCipher):
with session_factory() as session:
user_a = _create_user(session, cipher, name="Alice")
user_b = _create_user(session, cipher, name="Bob")
return user_a, user_b
def _build_manager(
*,
session_factory,
cipher: CredentialCipher,
settings: StubSettings,
user_a: SyncUser,
user_b: SyncUser,
fake_mw_a: FakeMyWhooshClient,
fake_mw_b: FakeMyWhooshClient,
fake_garmin_a: FakeGarminUploader,
fake_garmin_b: FakeGarminUploader,
):
mywhoosh_fakes = {str(user_a.id): fake_mw_a, str(user_b.id): fake_mw_b}
garmin_fakes = {str(user_a.id): fake_garmin_a, str(user_b.id): fake_garmin_b}
garmin_factory_calls: list[tuple[str, str, Path]] = []
def mywhoosh_factory(token_store):
user_key = token_store.path.parent.name
return mywhoosh_fakes[user_key]
def garmin_factory(email, password, tokenstore):
garmin_factory_calls.append((email, password, tokenstore))
user_key = tokenstore.parent.name
return garmin_fakes[user_key]
manager = SyncManager(
session_factory=session_factory,
credential_cipher=cipher,
settings=settings,
mywhoosh_factory=mywhoosh_factory,
garmin_factory=garmin_factory,
fit_converter=fake_fit_converter,
)
return manager, garmin_factory_calls
@pytest.mark.asyncio
async def test_two_user_happy_path(session_factory, cipher, settings, two_users):
user_a, user_b = two_users
remote_a = MyWhooshActivity(
id="remote-a-1", title="Alice Ride", activity_file_id="file-a-1", started_at=None
)
remote_b = MyWhooshActivity(
id="remote-b-1", title="Bob Ride", activity_file_id="file-b-1", started_at=None
)
fake_mw_a = FakeMyWhooshClient(activities=[remote_a], fit_bytes=b"alice-source-bytes")
fake_mw_b = FakeMyWhooshClient(activities=[remote_b], fit_bytes=b"bob-source-bytes")
fake_garmin_a = FakeGarminUploader()
fake_garmin_b = FakeGarminUploader()
manager, garmin_factory_calls = _build_manager(
session_factory=session_factory,
cipher=cipher,
settings=settings,
user_a=user_a,
user_b=user_b,
fake_mw_a=fake_mw_a,
fake_mw_b=fake_mw_b,
fake_garmin_a=fake_garmin_a,
fake_garmin_b=fake_garmin_b,
)
results = await manager.sync_all_enabled()
assert all(result.status == "success" for result in results)
assert count_terminal_activities(session_factory, user_a.id) == 1
assert count_terminal_activities(session_factory, user_b.id) == 1
with session_factory() as session:
activity_a = session.scalar(select(Activity).where(Activity.user_id == user_a.id))
activity_b = session.scalar(select(Activity).where(Activity.user_id == user_b.id))
assert Path(activity_a.source_fit_path).exists()
assert Path(activity_b.source_fit_path).exists()
assert Path(activity_a.source_fit_path).parent != Path(activity_b.source_fit_path).parent
tokenstores = {str(call[2]) for call in garmin_factory_calls}
assert len(tokenstores) == 2
# Idempotency (spec 26.8): re-running sync_all_enabled with no new remote
# activities must not re-download/re-convert/re-import anything, and must
# not create a second terminal Activity row for the same MyWhoosh activity.
download_calls_a_before = fake_mw_a.download_calls
download_calls_b_before = fake_mw_b.download_calls
garmin_calls_a_before = fake_garmin_a.calls
garmin_calls_b_before = fake_garmin_b.calls
second_results = await manager.sync_all_enabled()
assert all(result.status == "success" for result in second_results)
assert fake_mw_a.download_calls == download_calls_a_before
assert fake_mw_b.download_calls == download_calls_b_before
assert fake_garmin_a.calls == garmin_calls_a_before
assert fake_garmin_b.calls == garmin_calls_b_before
assert count_terminal_activities(session_factory, user_a.id) == 1
assert count_terminal_activities(session_factory, user_b.id) == 1
@pytest.mark.asyncio
async def test_isolation_when_one_user_requires_mfa(session_factory, cipher, settings, two_users):
user_a, user_b = two_users
remote_a = MyWhooshActivity(
id="remote-a-1", title="Alice Ride", activity_file_id="file-a-1", started_at=None
)
remote_b = MyWhooshActivity(
id="remote-b-1", title="Bob Ride", activity_file_id="file-b-1", started_at=None
)
fake_mw_a = FakeMyWhooshClient(activities=[remote_a], fit_bytes=b"alice-source-bytes")
fake_mw_b = FakeMyWhooshClient(activities=[remote_b], fit_bytes=b"bob-source-bytes")
fake_garmin_a = FakeGarminUploader()
fake_garmin_b = FakeGarminUploader(error=GarminUploadBlocked("Garmin requested MFA"))
manager, garmin_factory_calls = _build_manager(
session_factory=session_factory,
cipher=cipher,
settings=settings,
user_a=user_a,
user_b=user_b,
fake_mw_a=fake_mw_a,
fake_mw_b=fake_mw_b,
fake_garmin_a=fake_garmin_a,
fake_garmin_b=fake_garmin_b,
)
results = await manager.sync_all_enabled()
results_by_user = {result.user_id: result for result in results}
assert results_by_user[user_a.id].status == "success"
assert results_by_user[user_b.id].status != "success"
assert results_by_user[user_b.id].status == "failed"
assert count_terminal_activities(session_factory, user_a.id) == 1
with session_factory() as session:
reloaded_b = UserRepository(session).get(user_b.id)
assert reloaded_b.health_state == HealthState.ACTION_REQUIRED
assert reloaded_b.action_reason == "garmin_mfa_required"
reloaded_a = UserRepository(session).get(user_a.id)
assert reloaded_a.health_state == HealthState.HEALTHY
assert reloaded_a.action_reason is None
activity_a = session.scalar(select(Activity).where(Activity.user_id == user_a.id))
assert activity_a.status in (ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE)

View File

@@ -0,0 +1,30 @@
from pathlib import Path
from cryptography.fernet import Fernet
from fastapi.testclient import TestClient
from app.config import Settings
from app.main import create_app
from app.sync.manager import SyncManager
def test_lifespan_wires_sync_manager_and_scheduler(tmp_path: Path) -> None:
settings = Settings(
ADMIN_PASSWORD="admin-secret",
SECRET_KEY="0123456789abcdef0123456789abcdef",
CREDENTIAL_ENCRYPTION_KEY=Fernet.generate_key().decode("ascii"),
DATA_DIR=str(tmp_path),
DATABASE_URL=f"sqlite:///{tmp_path / 'app.db'}",
SYNC_INTERVAL_MINUTES=5,
)
app = create_app(settings)
# No users are seeded in this database, so sync_all_enabled() has nothing
# to iterate over and the real MyWhoosh/Garmin factories are never invoked.
with TestClient(app):
assert app.state.sync_manager is not None
assert isinstance(app.state.sync_manager, SyncManager)
assert app.state.scheduler is not None
assert app.state.scheduler.last_tick is not None
app.state.db_engine.dispose()

139
tests/web/test_mfa.py Normal file
View File

@@ -0,0 +1,139 @@
import logging
from sqlalchemy import text
from app.db.models import Activity, ActivityStatus
def test_mfa_code_is_used_once_and_not_persisted(authenticated_client, fake_sync_manager, app, caplog) -> None:
with caplog.at_level(logging.DEBUG):
response = authenticated_client.post(
"/users/1/garmin-mfa",
data={"csrf_token": authenticated_client.csrf_token, "code": "123456"},
)
assert response.status_code == 200
assert fake_sync_manager.mfa_calls == [(1, "123456")]
# Defense-in-depth: the MFA code must never end up written to the real
# app database (the one authenticated_client's requests actually hit),
# not some unrelated in-memory db.
with app.state.session_factory() as session:
persisted_text = " ".join(str(row) for row in session.execute(text("select * from sync_runs")).all())
assert "123456" not in persisted_text
# The check that actually matters: the code must never be logged,
# regardless of which SyncManager implementation is in play.
assert "123456" not in caplog.text
def test_mfa_code_rejects_empty_code(authenticated_client, fake_sync_manager) -> None:
response = authenticated_client.post(
"/users/1/garmin-mfa",
data={"csrf_token": authenticated_client.csrf_token, "code": " "},
)
assert response.status_code == 400
assert fake_sync_manager.mfa_calls == []
def test_mfa_code_rejects_overlong_code(authenticated_client, fake_sync_manager) -> None:
response = authenticated_client.post(
"/users/1/garmin-mfa",
data={"csrf_token": authenticated_client.csrf_token, "code": "1" * 21},
)
assert response.status_code == 400
assert fake_sync_manager.mfa_calls == []
def _seed_activity(app, *, status: ActivityStatus, last_completed_stage: ActivityStatus, retryable: bool, last_error: str | None = None) -> tuple[int, int]:
"""Seed a real Activity (and its owning user) in the app fixture's actual
database -- the same database authenticated_client's HTTP requests hit --
and return (user_id, activity_id)."""
from app.db.repositories import ActivityRepository, UserRepository
with app.state.session_factory() as session:
user = UserRepository(session).create(
name="MFA Test User",
enabled=True,
mywhoosh_email_enc="mw@example.com",
mywhoosh_password_enc="mw-pass",
garmin_email_enc="garmin@example.com",
garmin_password_enc="garmin-pass",
)
activity_repo = ActivityRepository(session)
activity, _ = activity_repo.get_or_create_discovered(
user_id=user.id,
mywhoosh_activity_id="mw-activity-1",
activity_name="Test Activity",
activity_timestamp=None,
)
activity.status = status
activity.last_completed_stage = last_completed_stage
activity.retryable = retryable
if last_error is not None:
activity.last_error = last_error
session.commit()
return user.id, activity.id
def test_retry_resets_and_calls_sync_for_retryable_failed_activity(authenticated_client, fake_sync_manager, app) -> None:
user_id, activity_id = _seed_activity(
app,
status=ActivityStatus.FAILED,
last_completed_stage=ActivityStatus.CONVERTED,
retryable=True,
last_error="some transient error",
)
response = authenticated_client.post(
f"/activities/{activity_id}/retry",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 200
assert fake_sync_manager.user_calls == [user_id]
with app.state.session_factory() as session:
reloaded = session.get(Activity, activity_id)
assert reloaded.status == ActivityStatus.CONVERTED
assert reloaded.last_error is None
def test_retry_rejects_non_retryable_activity(authenticated_client, fake_sync_manager, app) -> None:
user_id, activity_id = _seed_activity(
app,
status=ActivityStatus.FAILED,
last_completed_stage=ActivityStatus.CONVERTED,
retryable=False,
last_error="permanent failure",
)
response = authenticated_client.post(
f"/activities/{activity_id}/retry",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 409
assert fake_sync_manager.user_calls == []
with app.state.session_factory() as session:
reloaded = session.get(Activity, activity_id)
assert reloaded.status == ActivityStatus.FAILED
assert reloaded.retryable is False
assert reloaded.last_error == "permanent failure"
def test_retry_rejects_non_failed_activity(authenticated_client, fake_sync_manager, app) -> None:
user_id, activity_id = _seed_activity(
app,
status=ActivityStatus.DISCOVERED,
last_completed_stage=ActivityStatus.DISCOVERED,
retryable=True,
)
response = authenticated_client.post(
f"/activities/{activity_id}/retry",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 409
assert fake_sync_manager.user_calls == []

View File

@@ -0,0 +1,62 @@
from fastapi.testclient import TestClient
def test_manual_sync_calls_shared_manager(authenticated_client, fake_sync_manager) -> None:
response = authenticated_client.post(
"/users/1/sync",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 200
assert fake_sync_manager.user_calls == [1]
def test_manual_sync_reports_already_running(authenticated_client, fake_sync_manager) -> None:
fake_sync_manager.raise_already_running = True
response = authenticated_client.post(
"/users/1/sync",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 409
assert "already running" in response.text.lower()
def test_sync_all_calls_shared_manager(authenticated_client, fake_sync_manager) -> None:
response = authenticated_client.post(
"/sync-all",
data={"csrf_token": authenticated_client.csrf_token},
)
assert response.status_code == 200
assert fake_sync_manager.all_calls == 1
def test_manual_sync_requires_admin(client: TestClient) -> None:
response = client.post(
"/users/1/sync",
data={"csrf_token": "whatever"},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/login"
def test_manual_sync_rejects_invalid_csrf(authenticated_client) -> None:
response = authenticated_client.post(
"/users/1/sync",
data={"csrf_token": "invalid-token"},
)
assert response.status_code == 403
def test_system_page_shows_scheduler_state(app, authenticated_client) -> None:
class FakeScheduler:
def __init__(self) -> None:
self.last_tick = None
self.next_tick = None
app.state.scheduler = FakeScheduler()
response = authenticated_client.get("/system")
assert response.status_code == 200
assert "1.0.0" in response.text
assert "5" in response.text # sync_interval_minutes
assert "0" in response.text # user_count / activity_count fresh DB