96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mywhoosh_garmin_sync.models import DownloadedActivity
|
|
from mywhoosh_garmin_sync.service import ProcessingResult, SyncService
|
|
from mywhoosh_garmin_sync.state import ActivityStore
|
|
|
|
|
|
class FakeCrawler:
|
|
def __init__(self, downloads=None, exc: Exception | None = None):
|
|
self.downloads = downloads or []
|
|
self.exc = exc
|
|
|
|
async def download_new_activities(self, should_skip_source):
|
|
if self.exc is not None:
|
|
raise self.exc
|
|
return self.downloads
|
|
|
|
|
|
class FakeUploader:
|
|
pass
|
|
|
|
|
|
def test_run_once_records_ok_check_when_no_downloads(settings):
|
|
store = ActivityStore(settings.db_path)
|
|
service = SyncService(
|
|
settings,
|
|
crawler=FakeCrawler(),
|
|
uploader=FakeUploader(),
|
|
store=store,
|
|
)
|
|
|
|
asyncio.run(service.run_once())
|
|
|
|
latest = store.get_latest_sync_check()
|
|
assert latest is not None
|
|
assert latest.status == "ok"
|
|
assert latest.message == "OK: no new activities found."
|
|
assert latest.downloaded_count == 0
|
|
assert latest.failed_count == 0
|
|
|
|
|
|
def test_run_once_records_mywhoosh_problem_before_downloads(settings):
|
|
store = ActivityStore(settings.db_path)
|
|
service = SyncService(
|
|
settings,
|
|
crawler=FakeCrawler(exc=RuntimeError("MyWhoosh login did not complete")),
|
|
uploader=FakeUploader(),
|
|
store=store,
|
|
)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
asyncio.run(service.run_once())
|
|
|
|
latest = store.get_latest_sync_check()
|
|
assert latest is not None
|
|
assert latest.status == "problem"
|
|
assert latest.error_type == "mywhoosh_login"
|
|
assert latest.message == "MyWhoosh login did not complete"
|
|
|
|
|
|
def test_run_once_records_activity_processing_problem(settings, tmp_path: Path):
|
|
raw_path = tmp_path / "activity.fit"
|
|
raw_path.write_bytes(b"not used")
|
|
activity = DownloadedActivity(
|
|
source_ref="source-1",
|
|
title="Ride",
|
|
url="https://example.test/ride.fit",
|
|
raw_path=raw_path,
|
|
)
|
|
store = ActivityStore(settings.db_path)
|
|
service = SyncService(
|
|
settings,
|
|
crawler=FakeCrawler(downloads=[activity]),
|
|
uploader=FakeUploader(),
|
|
store=store,
|
|
)
|
|
service._process_activity = lambda activity: ProcessingResult(
|
|
"failed",
|
|
"garmin_login",
|
|
"Garmin requested MFA but GARMIN_MFA_CODE is not set.",
|
|
)
|
|
|
|
asyncio.run(service.run_once())
|
|
|
|
latest = store.get_latest_sync_check()
|
|
assert latest is not None
|
|
assert latest.status == "problem"
|
|
assert latest.error_type == "garmin_login"
|
|
assert latest.downloaded_count == 1
|
|
assert latest.processed_count == 1
|
|
assert latest.failed_count == 1
|
|
assert "1 of 1 downloaded activities failed" in latest.message
|