6de7562cd8
fix(garmin): 同步代码调用的 garminconnect API 全是错的 装上库核对签名后发现（garminconnect 0.2.8），原代码从未被真正 执行过，三处调用都不成立： 1. get_activities(date) —— 该方法签名是 (start, limit)，收的是 分页下标和条数，不是日期。传日期等于在问"第 2026-08-23 条运动"， 而且少传一个参数必然 TypeError。 改为 get_activities_by_date(start, end)，整个窗口一次取回， 顺带把原来"每天一次调用"减为一次。 2. 睡眠数据不在 get_user_summary 里，是独立的 get_sleep_data(cdate)； 原代码从 summary 里读 sleep 字段，结果会把每一晚都记成无睡眠数据。 HRV 同理，走 get_hrv_data(cdate)。 3. 字段名不符：实际是 totalSteps / totalKilocalories / averageStressLevel，原代码用的是 steps / calories.total / stress.average。 其他改进: - 运动记录改用 Garmin 自己的 activityId 作主键，重复同步同一窗口 不再产生重复行（原来每次同步都会把同一条运动再插一遍） - 某天无数据时返回全 None，不再写入空行让读端点再过滤掉 - 原来 bare except 吞掉每天的异常，全部失败也报 success； 现在全窗口失败会如实返回 error 并记录原因 - 同步天数、是否走 Garmin 中国区改为环境变量可配 tests/test_garmin_sync.py (23 通过): - 用 StubClient 模拟真实 0.2.8 的接口形状，无需库、凭证或网络 - 回归用例覆盖上述三处误用：活动必须按日期区间一次取回、 睡眠与 HRV 必须走各自端点 - 重复同步不产生重复的天和重复的运动记录 - 单天失败跳过、全窗口失败报错、登录失败如实上报 部署: - NAS 安装 garminconnect（pydantic-core 有 cp38 x86_64 轮子， 无需 gcc） - 新增 deploy/S99garmin.sh 开机自启脚本，与 NAS 上既有的 S99frpc.sh 同一套惯例；以 root 启动但降权到 ericwyuan 运行， 因为服务不需要特权而 .env 里有数据库和 API 凭证 - PID 文件放应用目录而非 /var/run（非特权用户写不了） NAS 真机全量测试: 240 passed Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""
|
|
Unit tests for the Garmin sync service.
|
|
|
|
A stub client stands in for `garminconnect`, so the suite runs without the
|
|
library, without credentials and without touching Garmin.
|
|
|
|
The stub mirrors the real 0.2.8 API shapes on purpose — the original code
|
|
called `get_activities(date)` when that method actually takes `(start, limit)`
|
|
pagination arguments, a mistake that only surfaces once something exercises it.
|
|
"""
|
|
import datetime
|
|
|
|
import pytest
|
|
|
|
from services import garmin as garmin_svc
|
|
from services import health as health_svc
|
|
|
|
|
|
def day(offset=0):
|
|
return (datetime.date.today() - datetime.timedelta(days=offset)).isoformat()
|
|
|
|
|
|
def summary(steps=8000, rhr=60, stress=40, kcal=2200):
|
|
return {
|
|
"totalSteps": steps,
|
|
"restingHeartRate": rhr,
|
|
"averageStressLevel": stress,
|
|
"totalKilocalories": kcal,
|
|
}
|
|
|
|
|
|
def sleep(hours=7.5, score=82):
|
|
return {
|
|
"dailySleepDTO": {
|
|
"sleepTimeSeconds": int(hours * 3600),
|
|
"sleepScores": {"overall": {"value": score}},
|
|
}
|
|
}
|
|
|
|
|
|
def hrv(value=48):
|
|
return {"hrvSummary": {"lastNightAvg": value}}
|
|
|
|
|
|
def activity(activity_id=1001, type_key="running", duration=1800):
|
|
return {
|
|
"activityId": activity_id,
|
|
"activityType": {"typeKey": type_key},
|
|
"startTimeLocal": f"{day()}T07:00:00",
|
|
"duration": duration,
|
|
"distance": 5000.0,
|
|
"calories": 320.0,
|
|
"averageHR": 145,
|
|
"maxHR": 168,
|
|
}
|
|
|
|
|
|
class StubClient:
|
|
"""Stands in for garminconnect.Garmin, recording how it was called."""
|
|
|
|
def __init__(self, summaries=None, sleeps=None, hrvs=None, activities=None,
|
|
fail_days=(), fail_activities=False):
|
|
self._summaries = summaries if summaries is not None else {}
|
|
self._sleeps = sleeps if sleeps is not None else {}
|
|
self._hrvs = hrvs if hrvs is not None else {}
|
|
self._activities = activities if activities is not None else []
|
|
self._fail_days = set(fail_days)
|
|
self._fail_activities = fail_activities
|
|
self.calls = []
|
|
|
|
def get_user_summary(self, cdate):
|
|
self.calls.append(("summary", cdate))
|
|
if cdate in self._fail_days:
|
|
raise RuntimeError(f"upstream error for {cdate}")
|
|
return self._summaries.get(cdate, summary())
|
|
|
|
def get_sleep_data(self, cdate):
|
|
self.calls.append(("sleep", cdate))
|
|
return self._sleeps.get(cdate, sleep())
|
|
|
|
def get_hrv_data(self, cdate):
|
|
self.calls.append(("hrv", cdate))
|
|
return self._hrvs.get(cdate, hrv())
|
|
|
|
def get_activities_by_date(self, startdate, enddate, activitytype=None):
|
|
self.calls.append(("activities", startdate, enddate))
|
|
if self._fail_activities:
|
|
raise RuntimeError("activities endpoint down")
|
|
return self._activities
|
|
|
|
|
|
CREDS = {"garminEmail": "g@example.com", "garminPassword": "pw"}
|
|
|
|
|
|
class TestHappyPath:
|
|
def test_reports_success(self, db, user):
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=StubClient())
|
|
assert out["status"] == "success"
|
|
assert out["recordsSynced"] == 3
|
|
|
|
def test_stores_the_days(self, db, user):
|
|
garmin_svc.sync_data(user["id"], CREDS, days=3, client=StubClient())
|
|
rows = health_svc.get_summary(user["id"])
|
|
assert len(rows) == 3
|
|
|
|
def test_maps_every_metric(self, db, user):
|
|
client = StubClient(
|
|
summaries={day(): summary(steps=9500, rhr=57, stress=33, kcal=2450)},
|
|
sleeps={day(): sleep(hours=8.0, score=91)},
|
|
hrvs={day(): hrv(52)},
|
|
)
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
row = health_svc.get_summary(user["id"])[0]
|
|
|
|
assert row["steps"] == 9500
|
|
assert row["heartRate"] == 57
|
|
assert row["stress"] == 33
|
|
assert row["caloriesBurned"] == 2450
|
|
assert row["heartRateVariability"] == 52
|
|
assert row["sleep"] == {"duration": 8.0, "quality": 91}
|
|
|
|
def test_sleep_and_hrv_come_from_their_own_endpoints(self, db, user):
|
|
"""Regression: both live outside get_user_summary. Reading only the
|
|
summary recorded every night as having no sleep data."""
|
|
client = StubClient()
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
|
|
kinds = {c[0] for c in client.calls}
|
|
assert "sleep" in kinds
|
|
assert "hrv" in kinds
|
|
|
|
def test_seconds_are_converted_to_hours(self, db, user):
|
|
client = StubClient(sleeps={day(): sleep(hours=6.5)})
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
assert health_svc.get_summary(user["id"])[0]["sleep"]["duration"] == 6.5
|
|
|
|
|
|
class TestActivities:
|
|
def test_fetched_by_date_range_in_one_call(self, db, user):
|
|
"""Regression: the old code called get_activities(date), but that
|
|
method takes (start, limit) pagination arguments, not a date."""
|
|
client = StubClient(activities=[activity()])
|
|
garmin_svc.sync_data(user["id"], CREDS, days=7, client=client)
|
|
|
|
activity_calls = [c for c in client.calls if c[0] == "activities"]
|
|
assert len(activity_calls) == 1, "one range call, not one call per day"
|
|
_, start, end = activity_calls[0]
|
|
assert start == day(6) and end == day(0)
|
|
|
|
def test_stored_with_fields_mapped(self, db, user):
|
|
garmin_svc.sync_data(
|
|
user["id"], CREDS, days=1, client=StubClient(activities=[activity()])
|
|
)
|
|
rows = health_svc.get_activities(user["id"])
|
|
assert len(rows) == 1
|
|
assert rows[0]["activity_type"] == "running"
|
|
assert rows[0]["heart_rate_average"] == 145
|
|
|
|
def test_count_is_reported(self, db, user):
|
|
client = StubClient(activities=[activity(1), activity(2)])
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
assert out["activitiesSynced"] == 2
|
|
|
|
def test_resync_does_not_duplicate(self, db, user):
|
|
"""Garmin activity ids are stable, so a re-synced window must skip
|
|
what is already stored."""
|
|
client = StubClient(activities=[activity(1001)])
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
|
|
assert len(health_svc.get_activities(user["id"])) == 1
|
|
assert out["activitiesSynced"] == 0
|
|
|
|
def test_end_time_derived_from_duration(self, db, user):
|
|
client = StubClient(activities=[activity(duration=1800)])
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
row = health_svc.get_activities(user["id"])[0]
|
|
assert row["start_time"] != row["end_time"]
|
|
|
|
def test_failure_does_not_lose_the_daily_data(self, db, user):
|
|
out = garmin_svc.sync_data(
|
|
user["id"], CREDS, days=2, client=StubClient(fail_activities=True)
|
|
)
|
|
assert out["status"] == "success"
|
|
assert len(health_svc.get_summary(user["id"])) == 2
|
|
|
|
|
|
class TestResync:
|
|
def test_same_day_is_updated_not_duplicated(self, db, user):
|
|
garmin_svc.sync_data(
|
|
user["id"], CREDS, days=1,
|
|
client=StubClient(summaries={day(): summary(steps=5000)}),
|
|
)
|
|
garmin_svc.sync_data(
|
|
user["id"], CREDS, days=1,
|
|
client=StubClient(summaries={day(): summary(steps=9000)}),
|
|
)
|
|
rows = health_svc.get_summary(user["id"])
|
|
assert len(rows) == 1
|
|
assert rows[0]["steps"] == 9000
|
|
|
|
|
|
class TestPartialAndTotalFailure:
|
|
def test_one_bad_day_is_skipped_not_fatal(self, db, user):
|
|
client = StubClient(fail_days=[day(1)])
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=client)
|
|
|
|
assert out["status"] == "success"
|
|
assert out["recordsSynced"] == 2
|
|
assert "跳过" in out["message"]
|
|
|
|
def test_every_day_failing_is_reported_as_an_error(self, db, user):
|
|
"""A systemic failure reported as a clean success would hide it."""
|
|
client = StubClient(fail_days=[day(0), day(1), day(2)])
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=3, client=client)
|
|
|
|
assert out["status"] == "error"
|
|
assert out["recordsSynced"] == 0
|
|
|
|
def test_login_failure_is_reported(self, db, user, monkeypatch):
|
|
def boom(_creds):
|
|
raise RuntimeError("401 Unauthorized")
|
|
|
|
monkeypatch.setattr(garmin_svc, "_connect", boom)
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=1)
|
|
|
|
assert out["status"] == "error"
|
|
assert "401" in out["message"]
|
|
|
|
def test_days_without_data_are_not_stored(self, db, user):
|
|
"""Garmin returns all-None for a day it has nothing for; an empty row
|
|
would just have to be filtered back out by every read endpoint."""
|
|
client = StubClient(
|
|
summaries={day(): {}}, sleeps={day(): {}}, hrvs={day(): {}}
|
|
)
|
|
out = garmin_svc.sync_data(user["id"], CREDS, days=1, client=client)
|
|
assert out["recordsSynced"] == 0
|
|
assert health_svc.get_summary(user["id"]) == []
|
|
|
|
|
|
class TestSyncStatus:
|
|
def test_idle_before_any_sync(self, db, user):
|
|
assert garmin_svc.get_sync_status(user["id"])["status"] == "idle"
|
|
|
|
def test_success_leaves_status_idle(self, db, user):
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=StubClient())
|
|
status = garmin_svc.get_sync_status(user["id"])
|
|
assert status["status"] == "idle"
|
|
assert status["recordsSynced"] == 1
|
|
assert status["lastSyncTime"]
|
|
|
|
def test_total_failure_leaves_status_error(self, db, user):
|
|
garmin_svc.sync_data(
|
|
user["id"], CREDS, days=2, client=StubClient(fail_days=[day(0), day(1)])
|
|
)
|
|
status = garmin_svc.get_sync_status(user["id"])
|
|
assert status["status"] == "error"
|
|
assert status["lastError"]
|
|
|
|
def test_a_later_success_clears_the_error(self, db, user):
|
|
garmin_svc.sync_data(
|
|
user["id"], CREDS, days=1, client=StubClient(fail_days=[day(0)])
|
|
)
|
|
garmin_svc.sync_data(user["id"], CREDS, days=1, client=StubClient())
|
|
status = garmin_svc.get_sync_status(user["id"])
|
|
assert status["status"] == "idle"
|
|
assert not status["lastError"]
|
|
|
|
|
|
class TestEndpoint:
|
|
def test_requires_auth(self, client):
|
|
assert client.post("/api/garmin/sync", json={}).status_code == 401
|
|
|
|
def test_missing_password_is_refused_with_a_reason(self, client, auth):
|
|
r = client.post("/api/garmin/sync", headers=auth, json={})
|
|
assert r.status_code == 400
|
|
assert "garminPassword" in r.get_json()["message"]
|
|
|
|
def test_status_endpoint(self, client, auth):
|
|
r = client.get("/api/garmin/status", headers=auth)
|
|
assert r.status_code == 200
|
|
assert r.get_json()["status"] == "idle"
|