4aaf490bc5
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>
89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
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 = 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
|