diff --git a/app/garmin/__init__.py b/app/garmin/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/garmin/uploader.py b/app/garmin/uploader.py new file mode 100644 index 0000000..69d67bd --- /dev/null +++ b/app/garmin/uploader.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Protocol + + +class GarminClientProtocol(Protocol): + def login(self, tokenstore: str | None = None) -> Any: ... + def import_activity(self, activity_path: str) -> Any: ... + + +@dataclass(frozen=True) +class UploadResult: + status: str + duplicate: bool + garmin_activity_id: str | None + raw_response: Any + + +class GarminUploadBlocked(RuntimeError): + pass + + +class GarminAuthError(RuntimeError): + pass + + +class GarminTransientError(RuntimeError): + pass + + +class GarminUploader: + def __init__( + self, + *, + email: str, + password: str, + tokenstore: Path, + client_factory: Callable[..., GarminClientProtocol] | None = None, + ) -> None: + self.email = email + self.password = password + self.tokenstore = tokenstore + self.client_factory = client_factory + self._client: GarminClientProtocol | None = None + self._mfa_code: str | None = None + + def import_fit(self, fit_path: Path, mfa_code: str | None = None) -> UploadResult: + self._mfa_code = mfa_code + try: + client = self._ensure_client() + try: + response = client.import_activity(str(fit_path)) + except Exception as exc: + if _looks_duplicate_error(exc): + return UploadResult("duplicate", True, None, str(exc)) + text = str(exc).lower() + if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")): + raise GarminTransientError("Garmin import failed transiently") from exc + raise + return UploadResult("imported", False, _extract_activity_id(response), response) + finally: + self._mfa_code = None + + def _ensure_client(self) -> GarminClientProtocol: + if self._client is not None: + return self._client + self.tokenstore.mkdir(parents=True, exist_ok=True) + factory = self.client_factory or _default_garmin_factory + client = factory(self.email, self.password, prompt_mfa=self._prompt_mfa) + try: + client.login(str(self.tokenstore)) + except GarminUploadBlocked: + raise + except Exception as exc: + text = str(exc).lower() + if "mfa" in text: + raise GarminUploadBlocked("Garmin MFA is required") from exc + if any(token in text for token in ("password", "credential", "unauthorized", "401")): + raise GarminAuthError("Garmin authentication failed") from exc + if any(token in text for token in ("timeout", "temporar", "connection", "502", "503", "504")): + raise GarminTransientError("Garmin login failed transiently") from exc + raise GarminAuthError("Garmin login failed") from exc + self._client = client + return client + + def _prompt_mfa(self) -> str: + if self._mfa_code: + return self._mfa_code + raise GarminUploadBlocked("Garmin requested MFA") + + +def _default_garmin_factory(*args: Any, **kwargs: Any) -> GarminClientProtocol: + from garminconnect import Garmin + + return Garmin(*args, **kwargs) + + +def _looks_duplicate_error(exc: Exception) -> bool: + text = str(exc).lower() + return any(token in text for token in ("duplicate", "already exists", "409")) + + +def _extract_activity_id(response: Any) -> str | None: + if not isinstance(response, dict): + return None + candidates = [response.get("activityId"), response.get("activity_id"), response.get("id")] + detailed = response.get("detailedImportResult") + if isinstance(detailed, dict): + candidates.extend([detailed.get("uploadId"), detailed.get("activityId")]) + for key in ("successes", "success", "importedActivities"): + items = response.get(key) + if isinstance(items, list) and items and isinstance(items[0], dict): + candidates.extend([items[0].get("activityId"), items[0].get("id")]) + return next((str(value) for value in candidates if value is not None), None) diff --git a/pyproject.toml b/pyproject.toml index f7197af..b618751 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "python-multipart>=0.0.9,<1", "itsdangerous>=2.1,<3", "httpx>=0.27,<1", + "garminconnect>=0.2,<1", ] [project.optional-dependencies] diff --git a/tests/garmin/__init__.py b/tests/garmin/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/garmin/test_uploader.py b/tests/garmin/test_uploader.py new file mode 100644 index 0000000..3e2a2c9 --- /dev/null +++ b/tests/garmin/test_uploader.py @@ -0,0 +1,80 @@ +from pathlib import Path + +import pytest + +from app.garmin.uploader import GarminUploadBlocked, GarminUploader + + +class FakeGarmin: + def __init__(self, *args, prompt_mfa=None, import_result=None, login_error=None, import_error=None, **kwargs): + self.prompt_mfa = prompt_mfa + self.import_result = import_result or {"activityId": 42} + self.login_error = login_error + self.import_error = import_error + self.login_path = None + + def login(self, tokenstore=None): + self.login_path = tokenstore + if self.login_error: + raise self.login_error + + def import_activity(self, activity_path: str): + if self.import_error: + raise self.import_error + return self.import_result + + +def test_successful_import_returns_activity_id(tmp_path: Path) -> None: + uploader = GarminUploader( + email="g@example.com", + password="pw", + tokenstore=tmp_path / "garmin", + client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_result={"activityId": 42}), + ) + result = uploader.import_fit(tmp_path / "ride.fit") + assert result.status == "imported" + assert result.garmin_activity_id == "42" + + +def test_duplicate_is_terminal_success(tmp_path: Path) -> None: + uploader = GarminUploader( + email="g@example.com", + password="pw", + tokenstore=tmp_path / "garmin", + client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_error=RuntimeError("409 duplicate")), + ) + result = uploader.import_fit(tmp_path / "ride.fit") + assert result.duplicate is True + assert result.status == "duplicate" + + +def test_mfa_without_code_is_blocked(tmp_path: Path) -> None: + class MfaGarmin(FakeGarmin): + def login(self, tokenstore=None): + self.prompt_mfa() + + uploader = GarminUploader( + email="g@example.com", + password="pw", + tokenstore=tmp_path / "garmin", + client_factory=MfaGarmin, + ) + with pytest.raises(GarminUploadBlocked): + uploader.import_fit(tmp_path / "ride.fit") + + +def test_mfa_code_is_returned_only_to_prompt(tmp_path: Path) -> None: + seen = [] + + class MfaGarmin(FakeGarmin): + def login(self, tokenstore=None): + seen.append(self.prompt_mfa()) + + uploader = GarminUploader( + email="g@example.com", + password="pw", + tokenstore=tmp_path / "garmin", + client_factory=MfaGarmin, + ) + uploader.import_fit(tmp_path / "ride.fit", mfa_code="123456") + assert seen == ["123456"]