bastian/mywhoosh2garmin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

plans + specs

Browse Source
This commit is contained in:
Bastian Wagner
2026-08-15 09:00:41 +02:00
commit f6da346e18
5 changed files with 3804 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

716
docs/superpowers/plans/2026-08-15-service-clients.md Normal file
View File

@@ -0,0 +1,716 @@
# 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
```text
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**
```python
# 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**
```python
# 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**
```python
# 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**
```bash
git add app/mywhoosh/models.py app/mywhoosh/tokenstore.py tests/mywhoosh/test_tokenstore.py
git commit -m "feat: add MyWhoosh token persistence"
```
## Task 2: Implement MyWhoosh login and cached-session recovery
**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:
```toml
"httpx>=0.27,<1",
```
and test dependencies:
```toml
"pytest-asyncio>=0.24,<1",
```
- [ ] **Step 2: Write authentication tests using `httpx.MockTransport`**
```python
# 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**
```python
# 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:
```python
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**
```bash
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**
```python
@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**
```python
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.
```python
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**
```bash
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:
```toml
"garminconnect>=0.2,<1",
```
- [ ] **Step 2: Write fake-client tests based on the previously working uploader pattern**
```python
# 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:
```python
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**
```python
# 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:
```python
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:
```python
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**
```bash
git add pyproject.toml app/garmin/uploader.py tests/garmin/test_uploader.py
git commit -m "feat: import FIT activities into Garmin"
```