Files

Bastian Wagner f6da346e18 plans + specs

2026-08-15 09:00:41 +02:00

24 KiB

Raw Blame History

MyWhoosh and Garmin Service Clients Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Implement isolated per-user MyWhoosh and Garmin clients with persistent tokenstores, mockable network boundaries, MyWhoosh direct API login/activity download, Garmin import_activity(), duplicate handling, and MFA signaling.

Architecture: Keep external integrations behind narrow protocols so the sync engine never depends directly on httpx or garminconnect. MyWhoosh uses an injected httpx.AsyncClient and a per-user JSON tokenstore. Garmin uses an injected client factory and per-user python-garminconnect token directory; blocking Garmin calls are later run through asyncio.to_thread by the sync layer.

Tech Stack: Python 3.12, httpx, python-garminconnect, pytest, pytest-asyncio.

Global Constraints

Do not automate or defeat MyWhoosh CAPTCHA/reCAPTCHA.
Follow the direct Android-style API login flow used by the reference jdelrue/mywhoosh2garmin project.
Cache MyWhoosh tokens per user under /data/tokens/<user-id>/mywhoosh.json.
Cache Garmin tokens per user under /data/tokens/<user-id>/garmin/ using the library's tokenstore behavior.
Authentication/API changes must become explicit integration/authentication errors, not uncontrolled retries.
Garmin activity transfer must use import_activity(), not upload_activity().
Known duplicate Garmin responses are successful terminal outcomes.
Garmin MFA must raise a dedicated exception so the UI can collect a one-time code.
Never log passwords, bearer tokens, Garmin tokens, or MFA codes.

File Structure

app/mywhoosh/
  __init__.py
  models.py
  tokenstore.py
  client.py
app/garmin/
  __init__.py
  uploader.py
tests/mywhoosh/
  test_tokenstore.py
  test_client_auth.py
  test_client_activities.py
tests/garmin/
  test_uploader.py

Task 1: Implement MyWhoosh models and tokenstore

Files:

Create: app/mywhoosh/models.py
Create: app/mywhoosh/tokenstore.py
Create: tests/mywhoosh/test_tokenstore.py

Interfaces:

Produces MyWhooshToken, MyWhooshActivity, MyWhooshTokenStore.load(), save(), clear().
Step 1: Write tokenstore tests

# tests/mywhoosh/test_tokenstore.py
from pathlib import Path

from app.mywhoosh.models import MyWhooshToken
from app.mywhoosh.tokenstore import MyWhooshTokenStore


def test_tokenstore_round_trip_and_permissions(tmp_path: Path) -> None:
    store = MyWhooshTokenStore(tmp_path / "tokens" / "7" / "mywhoosh.json")
    token = MyWhooshToken(access_token="access", refresh_token="refresh", whoosh_id="whoosh-7")
    store.save(token)

    assert store.load() == token
    assert oct(store.path.stat().st_mode & 0o777) == "0o600"


def test_missing_token_returns_none(tmp_path: Path) -> None:
    store = MyWhooshTokenStore(tmp_path / "missing.json")
    assert store.load() is None

Step 2: Run and verify failure

Run: pytest tests/mywhoosh/test_tokenstore.py -v

Expected: import failure.

Step 3: Implement models

# app/mywhoosh/models.py
from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class MyWhooshToken:
    access_token: str
    refresh_token: str | None
    whoosh_id: str | None


@dataclass(frozen=True)
class MyWhooshActivity:
    id: str
    title: str
    activity_file_id: str
    started_at: datetime | None

Step 4: Implement atomic JSON token persistence

# app/mywhoosh/tokenstore.py
import json
import os
from pathlib import Path

from app.mywhoosh.models import MyWhooshToken


class MyWhooshTokenStore:
    def __init__(self, path: Path) -> None:
        self.path = path

    def load(self) -> MyWhooshToken | None:
        try:
            raw = json.loads(self.path.read_text("utf-8"))
        except FileNotFoundError:
            return None
        return MyWhooshToken(
            access_token=raw["access_token"],
            refresh_token=raw.get("refresh_token"),
            whoosh_id=raw.get("whoosh_id"),
        )

    def save(self, token: MyWhooshToken) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(
            json.dumps(
                {
                    "access_token": token.access_token,
                    "refresh_token": token.refresh_token,
                    "whoosh_id": token.whoosh_id,
                },
                indent=2,
            ),
            "utf-8",
        )
        os.chmod(tmp, 0o600)
        tmp.replace(self.path)
        os.chmod(self.path, 0o600)

    def clear(self) -> None:
        self.path.unlink(missing_ok=True)

Step 5: Run tests

Run: pytest tests/mywhoosh/test_tokenstore.py -v

Expected: PASS.

Step 6: Commit

git add app/mywhoosh/models.py app/mywhoosh/tokenstore.py tests/mywhoosh/test_tokenstore.py
git commit -m "feat: add MyWhoosh token persistence"

Files:

Create: app/mywhoosh/client.py
Create: tests/mywhoosh/test_client_auth.py
Modify: pyproject.toml

Interfaces:

Produces exceptions MyWhooshAuthError, MyWhooshTransientError, MyWhooshIntegrationError.
Produces MyWhooshClient.ensure_authenticated(email: str, password: str) -> None.
Login endpoint: https://services.mywhoosh.com/http-service/api/login.
Login payload fields: Username, Password, Platform="Android", Action=1001, random CorrelationId, random DeviceId, Authorization="".
Step 1: Add test dependencies

Add to pyproject.toml runtime dependencies:

"httpx>=0.27,<1",

and test dependencies:

"pytest-asyncio>=0.24,<1",

Step 2: Write authentication tests using httpx.MockTransport

# tests/mywhoosh/test_client_auth.py
import httpx
import pytest

from app.mywhoosh.client import MyWhooshClient, MyWhooshAuthError
from app.mywhoosh.models import MyWhooshToken
from app.mywhoosh.tokenstore import MyWhooshTokenStore


@pytest.mark.asyncio
async def test_login_saves_access_refresh_and_whoosh_id(tmp_path) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/http-service/api/login"
        return httpx.Response(
            200,
            json={
                "Success": True,
                "AccessToken": "new-access",
                "RefreshToken": "new-refresh",
                "WhooshId": "w-1",
            },
        )

    store = MyWhooshTokenStore(tmp_path / "mywhoosh.json")
    client = MyWhooshClient(store, http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
    await client.login("rider@example.com", "secret")
    assert store.load() == MyWhooshToken("new-access", "new-refresh", "w-1")


@pytest.mark.asyncio
async def test_invalid_credentials_raise_auth_error(tmp_path) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"Success": False, "Message": "Invalid credentials"})

    client = MyWhooshClient(
        MyWhooshTokenStore(tmp_path / "mywhoosh.json"),
        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
    )
    with pytest.raises(MyWhooshAuthError):
        await client.login("rider@example.com", "bad")

Step 3: Run and verify failure

Run: pytest tests/mywhoosh/test_client_auth.py -v

Expected: missing client implementation.

Step 4: Implement exception taxonomy and login

# app/mywhoosh/client.py
from __future__ import annotations

import uuid

import httpx

from app.mywhoosh.models import MyWhooshToken
from app.mywhoosh.tokenstore import MyWhooshTokenStore

LOGIN_URL = "https://services.mywhoosh.com/http-service/api/login"
ACTIVITIES_BASE = "https://service14.mywhoosh.com/v2/"


class MyWhooshError(RuntimeError):
    pass


class MyWhooshAuthError(MyWhooshError):
    pass


class MyWhooshTransientError(MyWhooshError):
    pass


class MyWhooshIntegrationError(MyWhooshError):
    pass


class MyWhooshClient:
    def __init__(self, token_store: MyWhooshTokenStore, http_client: httpx.AsyncClient | None = None) -> None:
        self.token_store = token_store
        self.http = http_client or httpx.AsyncClient(timeout=30.0)
        self.token = token_store.load()

    async def login(self, email: str, password: str) -> None:
        payload = {
            "Username": email,
            "Password": password,
            "Platform": "Android",
            "Action": 1001,
            "CorrelationId": str(uuid.uuid4()),
            "DeviceId": str(uuid.uuid4()),
            "Authorization": "",
        }
        try:
            response = await self.http.post(LOGIN_URL, json=payload)
        except httpx.TransportError as exc:
            raise MyWhooshTransientError("MyWhoosh login request failed") from exc
        if response.status_code >= 500:
            raise MyWhooshTransientError(f"MyWhoosh login returned HTTP {response.status_code}")
        if response.status_code >= 400:
            raise MyWhooshAuthError(f"MyWhoosh login returned HTTP {response.status_code}")
        try:
            body = response.json()
        except ValueError as exc:
            raise MyWhooshIntegrationError("MyWhoosh login returned invalid JSON") from exc
        if body.get("Success") is not True or not body.get("AccessToken"):
            raise MyWhooshAuthError(str(body.get("Message") or "MyWhoosh login failed"))
        self.token = MyWhooshToken(
            access_token=str(body["AccessToken"]),
            refresh_token=str(body["RefreshToken"]) if body.get("RefreshToken") else None,
            whoosh_id=str(body["WhooshId"]) if body.get("WhooshId") else None,
        )
        self.token_store.save(self.token)

Step 5: Implement ensure_authenticated as cache-first validation

Do not invent a refresh endpoint. Validate cached tokens using the normal activities request; on 401/403, clear the cache, login once, and continue. The later list_activities() task provides the request method. Expose the intended behavior now:

async def ensure_authenticated(self, email: str, password: str) -> None:
    if self.token is None:
        await self.login(email, password)

Task 3 extends this with one retry after an unauthorized activities response.

Step 6: Run authentication tests

Run: pytest tests/mywhoosh/test_client_auth.py -v

Expected: PASS.

Step 7: Commit

git add pyproject.toml app/mywhoosh/client.py tests/mywhoosh/test_client_auth.py
git commit -m "feat: add MyWhoosh API login"

Task 3: Implement MyWhoosh activity listing and FIT download

Files:

Modify: app/mywhoosh/client.py
Create: tests/mywhoosh/test_client_activities.py

Interfaces:

Produces list_activities(email: str, password: str) -> list[MyWhooshActivity].
Produces download_fit(activity_file_id: str, email: str, password: str) -> bytes.
Activities endpoint: POST https://service14.mywhoosh.com/v2/rider/profile/activities with {"sortDate":"DESC","page":N}.
Download endpoint: POST https://service14.mywhoosh.com/v2/rider/profile/download-activity-file with {"fileId": activity_file_id}; response data is a pre-signed URL which is fetched with GET.
Step 1: Write paginated activity-list test

@pytest.mark.asyncio
async def test_list_activities_paginates_and_normalizes(tmp_path) -> None:
    calls = []

    async def handler(request: httpx.Request) -> httpx.Response:
        calls.append(str(request.url))
        if request.url.path.endswith("/activities"):
            payload = json.loads(request.content)
            page = payload["page"]
            result = {
                "data": {
                    "totalPages": 2,
                    "results": [{
                        "id": f"a-{page}",
                        "title": f"Ride {page}",
                        "activityFileId": f"f-{page}",
                        "startDatetime": "2026-08-15T06:00:00.000Z",
                    }],
                }
            }
            return httpx.Response(200, json=result)
        raise AssertionError(request.url)

Preload the tokenstore with access_token="cached"; assert two normalized MyWhooshActivity values are returned.

Step 2: Write expired-token reauthentication test

The mock transport sequence must return 401 for the first activities request, a successful login response, then 200 for the retried activities request. Assert login is attempted exactly once and the tokenstore contains the new access token.

Step 3: Write FIT download test

Mock the download-activity-file endpoint to return {"data":"https://signed.example/activity.fit"}, then mock that URL to return bytes beginning with a valid FIT header. Assert download_fit() returns those exact bytes.

Step 4: Implement one authenticated-request retry helper

async def _authenticated_post(self, url: str, payload: dict, email: str, password: str) -> httpx.Response:
    await self.ensure_authenticated(email, password)
    for attempt in range(2):
        assert self.token is not None
        try:
            response = await self.http.post(
                url,
                json=payload,
                headers={"Authorization": f"Bearer {self.token.access_token}"},
            )
        except httpx.TransportError as exc:
            raise MyWhooshTransientError("MyWhoosh request failed") from exc
        if response.status_code not in {401, 403}:
            if response.status_code >= 500:
                raise MyWhooshTransientError(f"MyWhoosh returned HTTP {response.status_code}")
            return response
        if attempt == 0:
            self.token_store.clear()
            self.token = None
            await self.login(email, password)
            continue
        raise MyWhooshAuthError("MyWhoosh session rejected after reauthentication")
    raise AssertionError("unreachable")

Step 5: Implement list normalization and download

Parse startDatetime as UTC when present. Skip malformed activity rows only if they lack no stable id or activityFileId; otherwise surface JSON/schema failures as MyWhooshIntegrationError so API changes are visible.

async def download_fit(self, activity_file_id: str, email: str, password: str) -> bytes:
    response = await self._authenticated_post(
        ACTIVITIES_BASE + "rider/profile/download-activity-file",
        {"fileId": activity_file_id},
        email,
        password,
    )
    if response.status_code >= 400:
        raise MyWhooshIntegrationError(f"download metadata returned HTTP {response.status_code}")
    url = response.json().get("data")
    if not isinstance(url, str) or not url:
        raise MyWhooshIntegrationError("MyWhoosh download response has no URL")
    try:
        fit_response = await self.http.get(url)
    except httpx.TransportError as exc:
        raise MyWhooshTransientError("FIT download failed") from exc
    if fit_response.status_code >= 500:
        raise MyWhooshTransientError(f"FIT host returned HTTP {fit_response.status_code}")
    if fit_response.status_code >= 400:
        raise MyWhooshIntegrationError(f"FIT host returned HTTP {fit_response.status_code}")
    return fit_response.content

Step 6: Run MyWhoosh tests

Run: pytest tests/mywhoosh -v

Expected: PASS.

Step 7: Commit

git add app/mywhoosh/client.py tests/mywhoosh/test_client_activities.py
git commit -m "feat: fetch MyWhoosh activities and FIT files"

Task 4: Implement Garmin import adapter with duplicate and MFA handling

Files:

Create: app/garmin/uploader.py
Create: tests/garmin/test_uploader.py
Modify: pyproject.toml

Interfaces:

Produces GarminUploader.import_fit(fit_path: Path, mfa_code: str | None = None) -> UploadResult.
Produces exceptions GarminUploadBlocked, GarminAuthError, GarminTransientError.
Uses client.login(tokenstore_path) and client.import_activity(activity_path).
Step 1: Add Garmin dependency

Add to runtime dependencies:

"garminconnect>=0.2,<1",

Step 2: Write fake-client tests based on the previously working uploader pattern

# tests/garmin/test_uploader.py
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

Add these concrete tests below the fake client:

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"]

Step 3: Run and verify failure

Run: pytest tests/garmin/test_uploader.py -v

Expected: import failure.

Step 4: Implement the adapter

# app/garmin/uploader.py
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

Implement the uploader fully:

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)

Step 5: Keep duplicate and response-ID extraction deterministic

Use these helpers:

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)

Step 6: Run Garmin tests

Run: pytest tests/garmin/test_uploader.py -v

Expected: PASS.

Step 7: Commit

git add pyproject.toml app/garmin/uploader.py tests/garmin/test_uploader.py
git commit -m "feat: import FIT activities into Garmin"

24 KiB Raw Blame History

MyWhoosh and Garmin Service Clients Implementation Plan

Global Constraints

File Structure

Task 1: Implement MyWhoosh models and tokenstore

Task 2: Implement MyWhoosh login and cached-session recovery

Task 3: Implement MyWhoosh activity listing and FIT download

Task 4: Implement Garmin import adapter with duplicate and MFA handling

24 KiB

Raw Blame History