From 990a55af140c4a6a27012d2233db21f5f9300c83 Mon Sep 17 00:00:00 2001 From: Bastian Wagner Date: Sat, 15 Aug 2026 17:09:55 +0200 Subject: [PATCH] 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 --- Dockerfile | 4 + app/sync/manager.py | 481 ++++++++++++++++++++++--------------- app/sync/scheduler.py | 17 +- tests/sync/test_manager.py | 132 +++++++++- tests/test_acceptance.py | 18 ++ 5 files changed, 457 insertions(+), 195 deletions(-) diff --git a/Dockerfile b/Dockerfile index 39458b4..eeb5597 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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"] diff --git a/app/sync/manager.py b/app/sync/manager.py index 1551bb7..1c3c4cc 100644 --- a/app/sync/manager.py +++ b/app/sync/manager.py @@ -8,7 +8,12 @@ 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, GarminTransientError, GarminUploadBlocked +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 @@ -21,6 +26,18 @@ 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. @@ -81,207 +98,285 @@ class SyncManager: sync_run_repo = SyncRunRepository(session) run = sync_run_repo.start(user_id) - 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) + # 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 - stop_user_run = False - summary_error: str | None = None + mywhoosh = None 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 - - activity_dir = self.settings.activities_dir / str(user.id) / activity.mywhoosh_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 + 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) - 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 + 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") - 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" + 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 - # 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 + 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() - 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 FitFormatError as exc: - activity_repo.mark_failed(activity.id, str(exc), retryable=False) - failed_count += 1 + 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: - user.health_state = HealthState.DEGRADED - activity_repo.mark_failed( - activity.id, - f"{type(exc).__name__}: {str(exc)[:200]}", - retryable=True, + # 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]}", ) - 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: - 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, - ) + session.commit() + raise + finally: + if mywhoosh is not None: + aclose = getattr(mywhoosh, "aclose", None) + if aclose is not None: + await aclose() diff --git a/app/sync/scheduler.py b/app/sync/scheduler.py index 3194a9b..b87882e 100644 --- a/app/sync/scheduler.py +++ b/app/sync/scheduler.py @@ -2,6 +2,8 @@ import asyncio import logging from datetime import datetime, timedelta, timezone +from app.sync.manager import SyncAlreadyRunning + logger = logging.getLogger(__name__) @@ -17,7 +19,20 @@ class SyncScheduler: async def run_once(self) -> None: self.last_tick = datetime.now(timezone.utc) try: - await self.manager.sync_all_enabled() + 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: diff --git a/tests/sync/test_manager.py b/tests/sync/test_manager.py index c5d713d..da97af0 100644 --- a/tests/sync/test_manager.py +++ b/tests/sync/test_manager.py @@ -3,7 +3,13 @@ from pathlib import Path import pytest from sqlalchemy import select -from app.db.models import ActivityStatus, SyncRun, SyncRunStatus, SyncUser +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 @@ -86,3 +92,127 @@ async def test_sync_run_repository_wiring_records_run(manager, seeded_user: Sync 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"] diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py index 2db5b65..72d8ba4 100644 --- a/tests/test_acceptance.py +++ b/tests/test_acceptance.py @@ -200,6 +200,24 @@ async def test_two_user_happy_path(session_factory, cipher, settings, two_users) 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):