feat: add MyWhoosh API login

This commit is contained in:
Bastian Wagner
2026-08-15 14:45:19 +02:00
parent a28d06ffdd
commit 01d2266c8e
3 changed files with 110 additions and 1 deletions

69
app/mywhoosh/client.py Normal file
View File

@@ -0,0 +1,69 @@
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)
async def ensure_authenticated(self, email: str, password: str) -> None:
if self.token is None:
await self.login(email, password)

View File

@@ -15,12 +15,13 @@ dependencies = [
"jinja2>=3.1,<4",
"python-multipart>=0.0.9,<1",
"itsdangerous>=2.1,<3",
"httpx>=0.27,<1",
]
[project.optional-dependencies]
test = [
"pytest>=8,<9",
"httpx>=0.27,<1",
"pytest-asyncio>=0.24,<1",
]
[tool.setuptools.packages]

View File

@@ -0,0 +1,39 @@
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")