diff --git a/app/db/repositories.py b/app/db/repositories.py index 7be9d6a..7f08d4d 100644 --- a/app/db/repositories.py +++ b/app/db/repositories.py @@ -1,10 +1,10 @@ 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 class UserRepository: @@ -37,6 +37,15 @@ 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 +85,101 @@ 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 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 diff --git a/tests/conftest.py b/tests/conftest.py index aa8aab8..c7322df 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 @@ -39,6 +39,11 @@ def activity_repository(db_session: Session) -> ActivityRepository: return ActivityRepository(db_session) +@pytest.fixture +def sync_run_repository(db_session: Session) -> SyncRunRepository: + return SyncRunRepository(db_session) + + @pytest.fixture def client(tmp_path: Path) -> TestClient: settings = Settings( @@ -54,3 +59,23 @@ def client(tmp_path: Path) -> TestClient: yield TestClient(app) finally: app.state.db_engine.dispose() + + +@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 diff --git a/tests/db/test_sync_state.py b/tests/db/test_sync_state.py new file mode 100644 index 0000000..f6904b4 --- /dev/null +++ b/tests/db/test_sync_state.py @@ -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]