diff --git a/app/sync/__init__.py b/app/sync/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/sync/manager.py b/app/sync/manager.py new file mode 100644 index 0000000..b4e5e6c --- /dev/null +++ b/app/sync/manager.py @@ -0,0 +1,252 @@ +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, 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 SyncManager: + """Resumable single-user MyWhoosh -> Garmin sync pipeline. + + Locking/scheduling across multiple users is layered on top of `sync_user` + by a later task; this class only implements the state machine for one + user's sync run. + """ + + 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 + + async def sync_user(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) + + 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) + imported_count = 0 + skipped_count = 0 + failed_count = 0 + stop_user_run = False + summary_error: str | None = 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 + + 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" + + # 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 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() + + 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, + ) diff --git a/app/sync/states.py b/app/sync/states.py new file mode 100644 index 0000000..82c5146 --- /dev/null +++ b/app/sync/states.py @@ -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 diff --git a/tests/sync/__init__.py b/tests/sync/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/sync/conftest.py b/tests/sync/conftest.py new file mode 100644 index 0000000..8bb0302 --- /dev/null +++ b/tests/sync/conftest.py @@ -0,0 +1,200 @@ +from pathlib import Path +from typing import Callable + +import pytest +from cryptography.fernet import Fernet +from sqlalchemy import create_engine +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 = ActivityRepository(session).session.query(Activity).filter_by(user_id=user_id).all() + 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 diff --git a/tests/sync/fakes.py b/tests/sync/fakes.py new file mode 100644 index 0000000..234bbe7 --- /dev/null +++ b/tests/sync/fakes.py @@ -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 diff --git a/tests/sync/test_manager.py b/tests/sync/test_manager.py new file mode 100644 index 0000000..332009f --- /dev/null +++ b/tests/sync/test_manager.py @@ -0,0 +1,87 @@ +from pathlib import Path + +import pytest + +from app.db.models import ActivityStatus, SyncRun, SyncRunStatus, SyncUser + + +@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 = session.query(SyncRun).filter_by(user_id=seeded_user.id).all() + 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