feat(sync): 运动详情改为同步入库,详情页只读本地

按需回源是错的:点一次运动要等七个 Garmin 接口,网络好的时候慢,
网络差的时候直接超时(实测公网下 Network Error)。

- sync_data 顺带补齐缺详情的运动
- POST /api/garmin/sync-details 后台补齐存量,GET 查进度
- 详情页只读本地库;没有就提示去同步,不再回源
- 同步页新增「补齐运动详情」按钮,带进度

身体年龄:加入公开的阻尼系数
- 34 岁 VO₂max 46 原本算出 21 岁。不是算错,是方法本身会饱和:
  人与人之间的 VO₂max 标准差约 7,而年龄每年只带来约 0.35 的衰减,
  于是稍微能练的人都会撞到参考表最年轻一档。
- 按 50% 向实际年龄收拢,收敛范围 ±20 → ±12 岁,同一算例现在给 27 岁。
- 去掉「高于最年轻一档按 20 岁计」的硬地板,那是一道正好落在用户身上的悬崖。
- 界面同时显示未收拢的原始值,阻尼系数写进评分依据。

路由:为每个路径补无斜杠别名
- F7 写地址栏时去掉尾斜杠,于是 /daily/ 在地址栏是 /daily,
  而那个地址匹配不到任何路由,刷新或分享就落到「找不到页面」。

布局:让页面结构上无法被撑宽
- 网格改用 minmax(min(210px,100%),1fr):裸的 minmax(210px,1fr) 允许
  两列加起来超过窄屏宽度,第二张卡就被切掉在屏幕外。
- .ring-row 用 minmax(0,1fr),1fr 会以 min-content 兜底,一句长说明就能
  把整行顶宽。
- .page-inner 加 overflow-x: clip。
- html/body 用 100dvh:手机浏览器把自己的地址栏盖在布局视口上,
  100% 高的应用会把底部 Tab 栏顶到它们下面——对用户来说就是没有 Tab 栏。

测试:新增 122 项(设置 44、身体年龄 44、运动详情 42),全量 446 项通过。

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-24 04:03:49 +08:00
parent fa865ca8a6
commit 34940cc387
16 changed files with 1218 additions and 69 deletions

View File

@@ -153,20 +153,38 @@ def auto_sync_status():
@bp.route("/activities/<activity_id>/detail", methods=["GET"])
@require_auth
def activity_detail(activity_id):
"""Everything Garmin holds for one activity: stats, laps, zones, series.
"""Everything stored for one activity: stats, laps, zones, series.
The first open goes out to Garmin and takes a few seconds; after that it
is served from the stored copy. `?refresh=1` forces a refetch.
A local read. Detail is fetched during the sync rather than when the user
taps an activity — seven Garmin calls on the critical path of a tap was
slow on a good connection and a timeout on a bad one.
"""
detail = garmin_svc.read_activity_detail(g.user_id, activity_id)
if detail is None:
return jsonify({
"error": "这条运动的详细数据还没同步到本机,去「同步」页拉一次即可。",
"needsSync": True,
}), 404
return jsonify(detail)
@bp.route("/sync-details", methods=["POST"])
@require_auth
def sync_details():
"""Backfill details for activities already stored without one."""
if not garmin_svc.has_token(g.user_id):
return jsonify({"error": "尚未绑定 Garmin 账号"}), 400
refresh = request.args.get("refresh") in ("1", "true", "yes")
body = request.get_json(silent=True) or {}
try:
return jsonify(
garmin_svc.get_activity_detail(g.user_id, activity_id, refresh=refresh)
)
except garmin_svc.MFARequired as e:
return jsonify({"error": str(e)}), 401
except Exception as e: # noqa: BLE001 - surfaced to the user verbatim
return jsonify({"error": garmin_svc.describe(e)}), 502
limit = int(body["limit"]) if body.get("limit") else None
except (TypeError, ValueError):
limit = None
return jsonify(garmin_svc.start_detail_sync(g.user_id, limit)), 202
@bp.route("/sync-details", methods=["GET"])
@require_auth
def sync_details_status():
return jsonify(garmin_svc.detail_sync_status(g.user_id))

View File

@@ -10,12 +10,11 @@ step it took is returned alongside the number for the UI to display.
Method
------
1. Base age from VO2max: the age at which the user's VO2max equals the median
for their sex, interpolated over the reference table below. VO2max is the
single strongest fitness predictor and is what Garmin's own model leans on.
for their sex, interpolated over the reference table below, then damped
towards their real age — see VO2_DAMPING for why that damping has to exist.
2. Resting-heart-rate adjustment, relative to a 60 bpm reference.
3. BMI adjustment, relative to the healthy 18.524.9 band.
4. Clamped to within 20 years of chronological age — beyond that the
extrapolation says more about the table's edges than about the person.
4. Clamped to within MAX_DEVIATION years of chronological age.
Reference values are 50th-percentile VO2max (ml/kg/min) by age and sex, from
the widely published ACSM / Cooper Institute cardiorespiratory fitness norms.
@@ -37,7 +36,15 @@ BMI_LOW, BMI_HIGH = 18.5, 24.9
BMI_YEARS_PER_UNIT = 0.5
BMI_CAP = 5.0
MAX_DEVIATION = 20.0 # years either side of chronological age
# Individual VO2max varies far more between people (SD ~7 ml/kg/min) than it
# declines with age (~0.35 ml/kg/min per year), so a raw "what age is this
# VO2max the median for" answer swings enormously: a VO2max of 46 at 34 reads
# as 20, because it genuinely is the median for a 20-year-old. Published
# fitness-age calculators all compress that swing; this one does too, by an
# explicit factor that is shown to the user rather than buried.
VO2_DAMPING = 0.5
MAX_DEVIATION = 12.0 # years either side of chronological age
AGE_FLOOR, AGE_CEILING = 20.0, 85.0
# Rendered verbatim in 设置 → 评分依据. Kept here, next to the constants it
@@ -52,8 +59,12 @@ BASIS = {
{
"name": "基准VO₂max 对应年龄",
"detail": "找出你的 VO₂max 相当于同性别人群哪个年龄的中位水平,"
"在参考表上线性插值;高于最年轻一档时按 20 岁计,"
"参考表再往上说明不了更多",
f"在参考表上线性插值,再按 {VO2_DAMPING:.0%} 的阻尼"
"向实际年龄收拢"
"不收拢的话,个体差异会淹没年龄效应——人与人之间的 "
"VO₂max 标准差约 7 ml/kg/min而年龄每年只带来约 0.35 "
"的衰减,于是稍微能练的人都会算出 20 岁。"
"界面上同时显示未收拢的原始值。",
"source": "ACSM / Cooper Institute 心肺适能人群常模50 百分位)",
},
{
@@ -84,16 +95,17 @@ BASIS = {
def _interpolate_age(vo2, table):
"""The age whose median VO2max equals `vo2`.
Above the youngest reference row the answer is simply "fitter than the
median 25-year-old", and the table cannot say more: extrapolating its slope
there gives absurdities (VO2max 48 reads as an eleven-year-old), so the
result floors instead.
Both ends continue along the nearest segment's slope. A hard floor here
would put everyone above the youngest row at exactly the same age, which
is a cliff precisely where the app's users sit; the damping applied to the
result afterwards is what keeps the extrapolation from running away.
"""
first_age, first_vo2 = table[0]
last_age, last_vo2 = table[-1]
if vo2 >= first_vo2:
return AGE_FLOOR
slope = (table[1][0] - first_age) / (table[1][1] - first_vo2)
return first_age + (vo2 - first_vo2) * slope
if vo2 <= last_vo2:
slope = (last_age - table[-2][0]) / (last_vo2 - table[-2][1])
return last_age + (vo2 - last_vo2) * slope
@@ -122,11 +134,19 @@ def estimate(*, age, sex, vo2max, resting_hr=None, bmi=None):
return {"value": None, "missing": missing, "basis": BASIS}
table = VO2_MEDIAN[sex]
base = _interpolate_age(float(vo2max), table)
chronological = float(age)
raw = _interpolate_age(float(vo2max), table)
# Pull the raw figure back towards the user's real age by the damping
# factor. Reported alongside the raw value so the compression is visible.
base = chronological + (raw - chronological) * VO2_DAMPING
steps = [{
"label": "VO₂max 基准",
"input": f"{float(vo2max):.0f} ml/kg/min",
"years": round(base, 1),
"raw": round(raw, 1),
"damping": VO2_DAMPING,
"kind": "base",
}]
@@ -160,7 +180,6 @@ def estimate(*, age, sex, vo2max, resting_hr=None, bmi=None):
"kind": "adjust",
})
chronological = float(age)
clamped = max(chronological - MAX_DEVIATION,
min(chronological + MAX_DEVIATION, total))
clamped = max(AGE_FLOOR, min(AGE_CEILING, clamped))

View File

@@ -24,7 +24,7 @@ import json
import os
import threading
from db import execute, query_one
from db import execute, query_one, query_all
from config import DB_TYPE
from services import health
@@ -586,28 +586,7 @@ def _build_detail(client, activity_id):
}
def get_activity_detail(user_id, activity_id, creds=None, refresh=False):
"""Cached detail for one activity, fetched from Garmin on first open."""
activity_id = str(activity_id)
if not refresh:
row = query_one(
"SELECT payload FROM activity_details "
"WHERE user_id = ? AND activity_id = ?",
[user_id, activity_id],
)
if row and row.get("payload"):
try:
cached = json.loads(row["payload"])
cached["cached"] = True
return cached
except ValueError:
# A truncated row is worth refetching, not worth crashing on.
pass
client = _connect(creds or {}, user_id=user_id)
detail = _build_detail(client, activity_id)
def _store_detail(user_id, activity_id, detail):
cols = ["activity_id", "user_id", "payload", "fetched_at"]
values = [activity_id, user_id, json.dumps(detail, default=str),
datetime.datetime.utcnow().isoformat(timespec="seconds")]
@@ -622,8 +601,93 @@ def get_activity_detail(user_id, activity_id, creds=None, refresh=False):
f"({placeholders}) ON CONFLICT(activity_id) DO UPDATE SET {updates}")
execute(sql, values)
detail["cached"] = False
return detail
def read_activity_detail(user_id, activity_id):
"""The stored detail for one activity, or None if it was never synced."""
row = query_one(
"SELECT payload FROM activity_details WHERE user_id = ? AND activity_id = ?",
[user_id, str(activity_id)],
)
if not row or not row.get("payload"):
return None
try:
return json.loads(row["payload"])
except ValueError:
# A truncated row is worth re-syncing, not worth crashing on.
return None
def sync_activity_details(client, user_id, limit=None, on_progress=None):
"""Fetch and store the full detail for activities that lack one.
Detail used to be fetched when the user opened an activity, which meant
seven Garmin calls on the critical path of a tap: slow at best, and a
timeout whenever the link was poor. It belongs in the sync, so the screen
only ever reads the local database.
"""
rows = query_all(
"SELECT a.id FROM activities a "
"LEFT JOIN activity_details d ON d.activity_id = a.id "
"WHERE a.user_id = ? AND d.activity_id IS NULL "
"ORDER BY a.start_time DESC",
[user_id],
)
if limit:
rows = rows[:limit]
stored = 0
for i, row in enumerate(rows):
try:
_store_detail(user_id, row["id"], _build_detail(client, row["id"]))
stored += 1
except Exception: # noqa: BLE001 - one bad activity must not stop the rest
continue
if on_progress:
on_progress(i + 1, len(rows))
return stored
_detail_progress = {}
def detail_sync_status(user_id):
"""Progress of the detail backfill for this account."""
return _detail_progress.get(user_id) or {"running": False, "done": 0, "total": 0}
def start_detail_sync(user_id, limit=None):
"""Backfill activity details in the background.
Each activity costs several Garmin calls, so 170 of them run for minutes —
far too long to hold a request open. The UI polls instead.
"""
state = _detail_progress.get(user_id)
if state and state.get("running"):
return state
_detail_progress[user_id] = {"running": True, "done": 0, "total": 0, "error": None}
def run():
try:
client = _connect({}, user_id=user_id)
def progress(done, total):
_detail_progress[user_id] = {
"running": True, "done": done, "total": total, "error": None,
}
stored = sync_activity_details(client, user_id, limit, on_progress=progress)
_detail_progress[user_id] = {
"running": False, "done": stored,
"total": _detail_progress[user_id].get("total", stored), "error": None,
}
except Exception as e: # noqa: BLE001 - reported through the status endpoint
_detail_progress[user_id] = {
"running": False, "done": 0, "total": 0, "error": describe(e),
}
threading.Thread(target=run, daemon=True, name=f"detail-sync-{user_id}").start()
return _detail_progress[user_id]
# Above this many days a sync is long enough that the caller must not block
@@ -711,6 +775,14 @@ def sync_data(user_id, creds, days=None, client=None):
except Exception as e:
day_errors.append(f"activities: {describe(e)}")
# Full detail for any activity that does not have it stored yet, so that
# opening one later is a local read.
details_synced = 0
try:
details_synced = sync_activity_details(client, user_id)
except Exception as e:
day_errors.append(f"activity_details: {describe(e)}")
# Badges and personal records are account-wide rather than per-day, so
# they are fetched once per sync rather than inside the day loop.
badges_synced = 0
@@ -738,7 +810,8 @@ def sync_data(user_id, creds, days=None, client=None):
last_error="; ".join(day_errors[:3]) if day_errors else None,
)
message = (
f"同步完成,更新 {days_synced} 天数据、{activities_synced} 条运动记录"
f"同步完成,更新 {days_synced} 天数据、{activities_synced} 条运动记录"
f"(含 {details_synced} 条详情)、"
f"{badges_synced} 个奖励、{records_synced_pr} 项个人纪录"
)
if day_errors:

View File

@@ -54,6 +54,21 @@ def _isolate_ai_env(monkeypatch):
monkeypatch.setenv("ALLOW_REGISTRATION", "true")
@pytest.fixture(autouse=True)
def _clear_garmin_client_cache():
"""Drop cached Garmin sessions between tests.
`_connect` keeps an authenticated client per user for 15 minutes, so a
stub installed by one test would otherwise be handed to the next one —
and a test that expects `_connect` to be called would see it skipped.
"""
from services import garmin as garmin_svc
garmin_svc._clients.clear()
yield
garmin_svc._clients.clear()
@pytest.fixture
def db(tmp_path, monkeypatch):
"""A freshly initialized, isolated SQLite database for one test."""

View File

@@ -0,0 +1,398 @@
"""
One activity, in full.
Garmin returns the sampled series as a column store — a descriptor list plus
rows of parallel values — so the fragile part is reading each metric out by the
index its descriptor names. Getting that wrong silently plots the wrong sensor.
"""
import json
import pytest
from services import garmin as svc
def descriptors(*keys):
return [{"key": k, "metricsIndex": i} for i, k in enumerate(keys)]
def rows(*tuples):
return [{"metrics": list(t)} for t in tuples]
class FakeClient:
"""Stands in for garminconnect.Garmin with recorded responses."""
def __init__(self, **responses):
self.responses = responses
self.calls = []
def _get(self, name, default):
self.calls.append(name)
value = self.responses.get(name, default)
if isinstance(value, Exception):
raise value
return value
def get_activity_evaluation(self, _id):
return self._get("evaluation", {})
def get_activity_details(self, _id, maxchart=None, maxpoly=None):
self.maxchart = maxchart
self.maxpoly = maxpoly
return self._get("details", {})
def get_activity_splits(self, _id):
return self._get("splits", {})
def get_activity_hr_in_timezones(self, _id):
return self._get("zones", [])
def get_activity_weather(self, _id):
return self._get("weather", {})
def get_activity_gear(self, _id):
return self._get("gear", [])
def get_activity_exercise_sets(self, _id):
return self._get("sets", {})
class TestSeriesExtraction:
def test_reads_each_metric_by_its_declared_index(self):
details = {
"metricDescriptors": descriptors(
"directTimestamp", "directHeartRate", "directSpeed"
),
"activityDetailMetrics": rows((1000, 120, 2.5), (2000, 130, 3.0)),
}
s = svc._series_from_details(details)
assert s["heartRate"] == [120, 130]
assert s["speed"] == [2.5, 3.0]
def test_index_is_taken_from_the_descriptor_not_its_position(self):
"""The descriptor list order and the metrics order are not the same
thing; assuming they are plots heart rate as elevation."""
details = {
"metricDescriptors": [
{"key": "directHeartRate", "metricsIndex": 2},
{"key": "directElevation", "metricsIndex": 0},
],
"activityDetailMetrics": rows((500, 999, 140)),
}
s = svc._series_from_details(details)
assert s["heartRate"] == [140]
assert s["elevation"] == [500]
def test_unknown_metrics_are_dropped(self):
details = {
"metricDescriptors": descriptors("directHeartRate", "sumMysteryField"),
"activityDetailMetrics": rows((120, 7)),
}
assert set(svc._series_from_details(details)) == {"heartRate"}
def test_all_null_column_is_dropped(self):
"""A column of nothing but nulls is a sensor the watch does not have,
and an empty chart is worse than no chart."""
details = {
"metricDescriptors": descriptors("directHeartRate", "directPower"),
"activityDetailMetrics": rows((120, None), (130, None)),
}
s = svc._series_from_details(details)
assert "heartRate" in s
assert "power" not in s
def test_partially_null_column_is_kept(self):
details = {
"metricDescriptors": descriptors("directHeartRate"),
"activityDetailMetrics": rows((120,), (None,), (130,)),
}
assert svc._series_from_details(details)["heartRate"] == [120, None, 130]
def test_short_row_does_not_raise(self):
"""A truncated row must not take the whole request down."""
details = {
"metricDescriptors": descriptors("directHeartRate", "directSpeed"),
"activityDetailMetrics": [{"metrics": [120, 2.5]}, {"metrics": [130]}],
}
s = svc._series_from_details(details)
assert s["speed"] == [2.5, None]
def test_empty_payload(self):
assert svc._series_from_details({}) == {}
assert svc._series_from_details({"metricDescriptors": []}) == {}
def test_cadence_variants_share_one_key(self):
"""Running, cycling and double cadence are the same chart to a reader."""
for key in ("directRunCadence", "directBikeCadence", "directDoubleCadence"):
details = {
"metricDescriptors": descriptors(key),
"activityDetailMetrics": rows((80,)),
}
assert svc._series_from_details(details)["cadence"] == [80]
class TestThinning:
def test_short_series_is_untouched(self):
assert svc._thin([1, 2, 3], limit=10) == [1, 2, 3]
def test_long_series_is_reduced_to_the_limit(self):
assert len(svc._thin(list(range(5000)), limit=300)) == 300
def test_keeps_the_first_and_last_sample(self):
thinned = svc._thin(list(range(1000)), limit=50)
assert thinned[0] == 0
assert thinned[-1] == 999
def test_stays_in_order(self):
thinned = svc._thin(list(range(1000)), limit=50)
assert thinned == sorted(thinned)
def test_series_are_thinned_together(self):
"""Thinning columns independently would pair a heart rate with another
moment's speed."""
details = {
"metricDescriptors": descriptors("directHeartRate", "directSpeed"),
"activityDetailMetrics": rows(*[(i, i * 2) for i in range(1000)]),
}
s = svc._series_from_details(details)
assert len(s["heartRate"]) == len(s["speed"])
assert all(sp == hr * 2 for hr, sp in zip(s["heartRate"], s["speed"]))
class TestLapsAndZones:
def test_laps_are_numbered_even_without_an_index(self):
laps = svc._lap_rows({"lapDTOs": [{"duration": 60}, {"duration": 70}]})
assert [l["index"] for l in laps] == [1, 2]
def test_lap_fields_are_carried_through(self):
laps = svc._lap_rows({"lapDTOs": [{
"lapIndex": 1, "duration": 2646.0, "distance": 1130.0,
"averageSpeed": 0.42, "averageHR": 105, "elevationGain": 42,
}]})
assert laps[0]["distance"] == 1130.0
assert laps[0]["averageHR"] == 105
assert laps[0]["elevationGain"] == 42
def test_no_laps(self):
assert svc._lap_rows({}) == []
assert svc._lap_rows(None) == []
def test_zones_are_sorted_by_number(self):
zones = svc._hr_zones([
{"zoneNumber": 3, "secsInZone": 10},
{"zoneNumber": 1, "secsInZone": 30},
{"zoneNumber": 2, "secsInZone": 20},
])
assert [z["zone"] for z in zones] == [1, 2, 3]
def test_zone_with_no_time_is_kept(self):
"""The empty zones are part of the picture: they show the session never
got hard, which is exactly what the reader is looking at."""
zones = svc._hr_zones([{"zoneNumber": 5, "secsInZone": 0}])
assert zones[0]["seconds"] == 0
def test_missing_seconds_becomes_zero_not_none(self):
assert svc._hr_zones([{"zoneNumber": 1}])[0]["seconds"] == 0
class TestBuildDetail:
def test_assembles_every_section(self, monkeypatch):
client = FakeClient(
evaluation={"activityName": "Shiyan 登山",
"activityTypeDTO": {"typeKey": "mountaineering"},
"summaryDTO": {"distance": 1130.0, "duration": 2646.0}},
splits={"lapDTOs": [{"lapIndex": 1, "duration": 2646.0}]},
zones=[{"zoneNumber": 1, "secsInZone": 1383}],
details={
"metricDescriptors": descriptors("directHeartRate"),
"activityDetailMetrics": rows((105,)),
},
)
d = svc._build_detail(client, "123")
assert d["activityName"] == "Shiyan 登山"
assert d["activityType"] == "mountaineering"
assert d["summary"]["distance"] == 1130.0
assert d["laps"][0]["duration"] == 2646.0
assert d["hrZones"][0]["seconds"] == 1383
assert d["series"]["heartRate"] == [105]
def test_one_failing_endpoint_does_not_lose_the_rest(self):
"""A treadmill run has no weather and no gear; those 404s must not take
the page down."""
client = FakeClient(
evaluation={"summaryDTO": {"duration": 600}},
weather=RuntimeError("404"),
gear=RuntimeError("404"),
sets=RuntimeError("404"),
)
d = svc._build_detail(client, "123")
assert d["summary"]["duration"] == 600
assert d["weather"] == {}
assert d["gear"] == []
def test_does_not_request_more_samples_than_it_keeps(self):
client = FakeClient()
svc._build_detail(client, "123")
assert client.maxchart <= 500
def test_does_not_fetch_the_map_polyline(self):
"""There is no map in the UI, and the polyline is the largest single
part of the payload."""
client = FakeClient()
svc._build_detail(client, "123")
assert client.maxpoly == 0
class TestStoreAndRead:
def test_read_returns_none_when_never_synced(self, db, user):
assert svc.read_activity_detail(user["id"], "abc") is None
def test_round_trip(self, db, user):
svc._store_detail(user["id"], "abc", {"summary": {"duration": 600}})
assert svc.read_activity_detail(user["id"], "abc")["summary"]["duration"] == 600
def test_corrupt_row_reads_as_missing_rather_than_raising(self, db, user):
db.execute(
"INSERT INTO activity_details (activity_id, user_id, payload) "
"VALUES (?,?,?)", ["abc", user["id"], "{not json"]
)
assert svc.read_activity_detail(user["id"], "abc") is None
def test_details_are_per_account(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
svc._store_detail(user["id"], "abc", {"summary": {"duration": 600}})
assert svc.read_activity_detail(other["id"], "abc") is None
def test_storing_again_replaces(self, db, user):
svc._store_detail(user["id"], "abc", {"summary": {"duration": 1}})
svc._store_detail(user["id"], "abc", {"summary": {"duration": 2}})
assert svc.read_activity_detail(user["id"], "abc")["summary"]["duration"] == 2
class TestSyncActivityDetails:
def seed(self, db, user, ids):
for i, aid in enumerate(ids):
stamp = f"2026-08-{10 + i:02d}T08:00:00"
db.execute(
"INSERT INTO activities (id, user_id, activity_type, start_time, "
"end_time) VALUES (?,?,?,?,?)",
[aid, user["id"], "running", stamp, stamp],
)
def test_fetches_every_activity_without_a_detail(self, db, user):
self.seed(db, user, ["a1", "a2", "a3"])
client = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
assert svc.sync_activity_details(client, user["id"]) == 3
assert svc.read_activity_detail(user["id"], "a2") is not None
def test_skips_activities_that_already_have_one(self, db, user):
self.seed(db, user, ["a1", "a2"])
svc._store_detail(user["id"], "a1", {"summary": {}})
client = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
assert svc.sync_activity_details(client, user["id"]) == 1
def test_is_a_no_op_once_everything_is_stored(self, db, user):
self.seed(db, user, ["a1"])
client = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
svc.sync_activity_details(client, user["id"])
assert svc.sync_activity_details(client, user["id"]) == 0
def test_limit_caps_the_batch(self, db, user):
self.seed(db, user, ["a1", "a2", "a3", "a4"])
client = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
assert svc.sync_activity_details(client, user["id"], limit=2) == 2
def test_newest_activities_are_fetched_first(self, db, user):
"""A backfill that runs for minutes should make the activities the user
is most likely to open available first."""
self.seed(db, user, ["old", "new"]) # start_time ascends with index
client = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
svc.sync_activity_details(client, user["id"], limit=1)
assert svc.read_activity_detail(user["id"], "new") is not None
assert svc.read_activity_detail(user["id"], "old") is None
def test_one_failing_activity_does_not_stop_the_rest(self, db, user):
self.seed(db, user, ["a1", "a2", "a3"])
calls = {"n": 0}
def flaky(_client, activity_id):
calls["n"] += 1
if calls["n"] == 1:
raise RuntimeError("Garmin 挂了")
return {"activityId": activity_id, "summary": {}}
import services.garmin as g
original = g._build_detail
g._build_detail = flaky
try:
assert svc.sync_activity_details(FakeClient(), user["id"]) == 2
finally:
g._build_detail = original
def test_only_this_accounts_activities(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
self.seed(db, user, ["mine"])
db.execute(
"INSERT INTO activities (id, user_id, activity_type, start_time, end_time) "
"VALUES (?,?,?,?,?)",
["theirs", other["id"], "running", "2026-08-11T08:00:00",
"2026-08-11T09:00:00"],
)
fake = FakeClient(evaluation={"summaryDTO": {"duration": 600}})
assert svc.sync_activity_details(fake, user["id"]) == 1
assert svc.read_activity_detail(other["id"], "theirs") is None
def test_progress_is_reported(self, db, user):
self.seed(db, user, ["a1", "a2"])
seen = []
svc.sync_activity_details(
FakeClient(), user["id"], on_progress=lambda d, t: seen.append((d, t))
)
assert seen == [(1, 2), (2, 2)]
class TestEndpoint:
def test_requires_auth(self, client):
assert client.get("/api/garmin/activities/1/detail").status_code == 401
def test_unsynced_activity_says_so(self, client, auth):
r = client.get("/api/garmin/activities/nope/detail", headers=auth)
assert r.status_code == 404
body = r.get_json()
assert body["needsSync"] is True
assert "同步" in body["error"]
def test_returns_the_stored_detail(self, client, auth, user, db):
svc._store_detail(user["id"], "abc", {"summary": {"duration": 600}})
body = client.get("/api/garmin/activities/abc/detail", headers=auth).get_json()
assert body["summary"]["duration"] == 600
def test_reading_never_touches_garmin(self, client, auth, user, db, monkeypatch):
"""The whole point of storing details during the sync: a tap must not
depend on Garmin being reachable."""
svc._store_detail(user["id"], "abc", {"summary": {"duration": 600}})
def explode(*a, **k):
raise AssertionError("opening an activity must be a local read")
monkeypatch.setattr(svc, "_connect", explode)
assert client.get(
"/api/garmin/activities/abc/detail", headers=auth
).status_code == 200
def test_backfill_needs_a_bound_account(self, client, auth):
r = client.post("/api/garmin/sync-details", headers=auth, json={})
assert r.status_code == 400
def test_backfill_status_is_readable(self, client, auth):
body = client.get("/api/garmin/sync-details", headers=auth).get_json()
assert body["running"] is False

View File

@@ -0,0 +1,253 @@
"""
身体年龄.
This is the one calculation in the app that produces a health verdict out of
thin air, so the properties that matter are: it never invents a number from
missing inputs, it moves in the right direction, and it says what it did.
"""
import pytest
from services import fitness_age as fa
def est(**kwargs):
base = {"age": 36, "sex": "male", "vo2max": 40,
"resting_hr": 60, "bmi": 22.0}
base.update(kwargs)
return fa.estimate(**base)
class TestMissingInputs:
def test_no_age(self):
r = est(age=None)
assert r["value"] is None
assert "出生日期" in r["missing"]
def test_no_sex(self):
r = est(sex=None)
assert r["value"] is None
assert "性别" in r["missing"]
def test_unknown_sex_is_treated_as_missing(self):
"""'other' has no reference table; guessing one would be worse than
saying the estimate cannot be made."""
r = est(sex="other")
assert r["value"] is None
assert "性别" in r["missing"]
def test_no_vo2max(self):
r = est(vo2max=None)
assert r["value"] is None
assert any("VO₂max" in m for m in r["missing"])
def test_missing_message_explains_how_to_get_vo2max(self):
r = est(vo2max=None)
assert "跑步" in " ".join(r["missing"])
def test_all_missing_are_listed_at_once(self):
r = est(age=None, sex=None, vo2max=None)
assert len(r["missing"]) == 3, "the user should fix everything in one go"
def test_basis_is_returned_even_when_the_estimate_cannot_be_made(self):
assert est(age=None)["basis"]["steps"]
def test_optional_inputs_are_genuinely_optional(self):
r = est(resting_hr=None, bmi=None)
assert r["value"] is not None
class TestDirection:
def test_higher_vo2max_gives_a_younger_body_age(self):
assert est(vo2max=45)["value"] < est(vo2max=33)["value"]
def test_lower_resting_hr_gives_a_younger_body_age(self):
assert est(resting_hr=48)["value"] < est(resting_hr=78)["value"]
def test_bmi_outside_the_healthy_band_never_helps(self):
healthy = est(bmi=22)["value"]
assert est(bmi=31)["value"] >= healthy
assert est(bmi=16)["value"] >= healthy
def test_bmi_inside_the_band_does_not_move_it(self):
assert est(bmi=18.5)["value"] == est(bmi=24.9)["value"] == est(bmi=22)["value"]
def test_women_need_less_vo2max_for_the_same_body_age(self):
"""The reference distributions differ by sex; using one table for both
would systematically flatter men and penalise women."""
assert est(sex="female", vo2max=37)["value"] < est(sex="male", vo2max=37)["value"]
def test_monotonic_across_the_whole_range(self):
values = [est(vo2max=v)["value"] for v in range(25, 55)]
assert values == sorted(values, reverse=True), values
class TestBounds:
def test_never_exceeds_the_deviation_cap_downwards(self):
r = est(age=60, vo2max=60, resting_hr=40, bmi=22)
assert r["value"] >= 60 - fa.MAX_DEVIATION
def test_never_exceeds_the_deviation_cap_upwards(self):
r = est(age=30, vo2max=15, resting_hr=95, bmi=40)
assert r["value"] <= 30 + fa.MAX_DEVIATION
def test_floor(self):
assert est(age=25, vo2max=70)["value"] >= fa.AGE_FLOOR
def test_ceiling(self):
assert est(age=84, vo2max=12, resting_hr=100, bmi=45)["value"] <= fa.AGE_CEILING
def test_clamping_is_disclosed(self):
"""A clamped number is not the raw result, and the UI says so — that
only works if the flag is set."""
assert est(age=60, vo2max=75, resting_hr=40)["clamped"] is True
def test_an_ordinary_result_is_not_flagged_as_clamped(self):
assert est(age=40, vo2max=37, resting_hr=60, bmi=22)["clamped"] is False
def test_rhr_adjustment_is_capped(self):
"""Without the cap a very high resting heart rate alone would drive the
whole estimate."""
moderate = est(resting_hr=90)["value"]
extreme = est(resting_hr=200)["value"]
assert extreme - moderate <= 2
def test_bmi_adjustment_is_capped(self):
assert est(bmi=60)["value"] - est(bmi=32)["value"] <= 3
class TestExtrapolation:
def test_above_the_youngest_reference_row_keeps_extrapolating(self):
"""A hard floor here put everyone fitter than the median 25-year-old at
exactly the same body age — a cliff right where this app's users sit."""
a = fa._interpolate_age(46, fa.VO2_MEDIAN["male"])
b = fa._interpolate_age(50, fa.VO2_MEDIAN["male"])
assert a > b, "a fitter reading must still move the number"
def test_below_the_oldest_row_keeps_extrapolating(self):
old = fa._interpolate_age(20, fa.VO2_MEDIAN["male"])
assert old > fa.VO2_MEDIAN["male"][-1][0]
def test_interpolates_between_rows(self):
table = fa.VO2_MEDIAN["male"]
# 42.5 sits midway between the 25-year (44) and 35-year (41) rows.
age = fa._interpolate_age(42.5, table)
assert 29 < age < 31
def test_reference_tables_decrease_with_age(self):
for sex, table in fa.VO2_MEDIAN.items():
ages = [a for a, _ in table]
vo2s = [v for _, v in table]
assert ages == sorted(ages), sex
assert vo2s == sorted(vo2s, reverse=True), sex
class TestDamping:
"""Individual VO2max spread (SD ~7) dwarfs the age decline (~0.35/yr), so
an undamped estimate reads 20 for anyone reasonably fit — which is what it
did, and why a 34-year-old with VO2max 46 was told he was 21."""
def test_a_fit_thirtysomething_is_not_told_he_is_twenty(self):
r = fa.estimate(age=34, sex="male", vo2max=46, resting_hr=62, bmi=26.0)
assert r["value"] >= 25, r
def test_damping_pulls_towards_the_real_age(self):
r = est(age=34, vo2max=46)
base = r["steps"][0]
assert base["raw"] < base["years"] < 34
def test_the_raw_value_is_still_reported(self):
"""Damping is a choice, so the undamped figure stays visible rather
than being quietly replaced."""
base = est()["steps"][0]
assert "raw" in base and "damping" in base
def test_an_average_person_lands_near_their_real_age(self):
r = fa.estimate(age=35, sex="male", vo2max=41, resting_hr=60, bmi=22)
assert abs(r["delta"]) <= 1, r
def test_damping_does_not_flip_the_direction(self):
assert est(vo2max=48)["value"] < est(vo2max=32)["value"]
def test_basis_discloses_the_damping_factor(self):
step = fa.BASIS["steps"][0]
assert "阻尼" in step["detail"]
class TestTransparency:
def test_every_step_is_reported(self):
r = est()
labels = [s["label"] for s in r["steps"]]
assert "VO₂max 基准" in labels
assert "静息心率" in labels
assert "BMI" in labels
def test_steps_carry_the_input_they_used(self):
r = est(resting_hr=59)
hr = next(s for s in r["steps"] if s["label"] == "静息心率")
assert "59" in hr["input"]
def test_omitted_inputs_produce_no_step(self):
r = est(resting_hr=None, bmi=None)
assert [s["label"] for s in r["steps"]] == ["VO₂max 基准"]
def test_delta_matches_the_reported_ages(self):
r = est(age=36, vo2max=42)
assert r["delta"] == r["value"] - r["chronologicalAge"]
def test_basis_names_a_source_for_every_step(self):
for step in fa.BASIS["steps"]:
assert step["source"], step
def test_basis_states_this_is_not_garmins_number(self):
assert "Garmin" in fa.BASIS["summary"]
def test_basis_carries_a_medical_caveat(self):
assert "" in fa.BASIS["caveat"]
def test_basis_constants_match_the_code(self):
"""The prose describes the arithmetic; if someone tunes a constant the
description must not silently keep the old number."""
rhr = next(s for s in fa.BASIS["steps"] if "心率" in s["name"])
assert f"{fa.RHR_REFERENCE:.0f}" in rhr["detail"]
bmi = next(s for s in fa.BASIS["steps"] if "BMI" in s["name"])
assert str(fa.BMI_LOW) in bmi["detail"] and str(fa.BMI_HIGH) in bmi["detail"]
class TestEndpoint:
def test_requires_auth(self, client):
assert client.get("/api/health/fitness-age").status_code == 401
def test_reports_what_is_missing_for_an_empty_profile(self, client, auth):
body = client.get("/api/health/fitness-age", headers=auth).get_json()
assert body["value"] is None
assert body["missing"]
def test_uses_the_saved_profile_and_the_latest_readings(
self, client, auth, user, db, seed_health
):
client.put("/api/settings", headers=auth,
json={"birthDate": "1990-05-04", "sex": "male",
"heightCm": 178, "weightKg": 72})
seed_health([{"date": "2026-08-20", "heart_rate": 58}])
db.execute(
"UPDATE health_data SET vo2max = ? WHERE user_id = ?", [42, user["id"]]
)
body = client.get("/api/health/fitness-age", headers=auth).get_json()
assert body["value"] is not None, body
assert body["chronologicalAge"] == 36
def test_falls_back_to_an_older_vo2max_reading(
self, client, auth, user, db, seed_health
):
"""VO2max only refreshes after an outdoor run, so the last recorded
value is still the current one."""
client.put("/api/settings", headers=auth,
json={"birthDate": "1990-05-04", "sex": "male"})
seed_health([{"date": "2026-06-01"}, {"date": "2026-08-20"}])
db.execute(
"UPDATE health_data SET vo2max = ? WHERE user_id = ? AND date = ?",
[44, user["id"], "2026-06-01"],
)
body = client.get("/api/health/fitness-age", headers=auth).get_json()
assert body["value"] is not None, "an old VO2max still counts"

View File

@@ -0,0 +1,250 @@
"""
Profile, units and sync preferences.
The interesting behaviour is the validation boundary: a body measurement that
is out of range is a user mistake and must be rejected with a message they can
act on, while an off-list picker value is a stale client and must be snapped
rather than refused.
"""
import datetime
import pytest
from services import settings as svc
class TestDefaults:
def test_unset_account_gets_defaults(self, db, user):
s = svc.get_settings(user["id"])
assert s["units"] == "metric"
assert s["autoSync"] is True
assert s["autoSyncMinutes"] == 60
assert s["historyDays"] == 365
def test_body_fields_start_empty(self, db, user):
s = svc.get_settings(user["id"])
assert s["heightCm"] is None
assert s["weightKg"] is None
assert s["birthDate"] is None
assert s["sex"] is None
def test_derived_fields_are_none_without_inputs(self, db, user):
s = svc.get_settings(user["id"])
assert s["age"] is None
assert s["bmi"] is None
class TestSaving:
def test_round_trip(self, db, user):
svc.save_settings(user["id"], {"heightCm": 178, "weightKg": 72})
s = svc.get_settings(user["id"])
assert s["heightCm"] == 178
assert s["weightKg"] == 72
def test_partial_patch_leaves_the_rest_alone(self, db, user):
svc.save_settings(user["id"], {"heightCm": 178, "sex": "male"})
svc.save_settings(user["id"], {"weightKg": 70})
s = svc.get_settings(user["id"])
assert s["heightCm"] == 178, "an unrelated field must survive a patch"
assert s["sex"] == "male"
assert s["weightKg"] == 70
def test_clearing_a_field(self, db, user):
svc.save_settings(user["id"], {"heightCm": 178})
svc.save_settings(user["id"], {"heightCm": None})
assert svc.get_settings(user["id"])["heightCm"] is None
def test_unknown_keys_are_ignored_not_rejected(self, db, user):
"""An older or newer client posting a field we do not know about must
still save the fields we do."""
svc.save_settings(user["id"], {"heightCm": 170, "favouriteColour": "blue"})
assert svc.get_settings(user["id"])["heightCm"] == 170
def test_empty_patch_is_a_no_op(self, db, user):
svc.save_settings(user["id"], {"heightCm": 180})
svc.save_settings(user["id"], {})
assert svc.get_settings(user["id"])["heightCm"] == 180
def test_settings_are_per_account(self, db, user, client):
other = client.post(
"/api/auth/register",
json={"email": "b@example.com", "garminEmail": "bg@example.com",
"garminPassword": "pw123456"},
).get_json()
svc.save_settings(user["id"], {"heightCm": 178})
svc.save_settings(other["id"], {"heightCm": 160})
assert svc.get_settings(user["id"])["heightCm"] == 178
assert svc.get_settings(other["id"])["heightCm"] == 160
class TestValidation:
@pytest.mark.parametrize("value", [79, 251, -5, 0])
def test_height_out_of_range(self, db, user, value):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"heightCm": value})
@pytest.mark.parametrize("value", [24, 301])
def test_weight_out_of_range(self, db, user, value):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"weightKg": value})
def test_non_numeric_height(self, db, user):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"heightCm": "tall"})
def test_message_names_the_field(self, db, user):
with pytest.raises(svc.InvalidSetting) as e:
svc.save_settings(user["id"], {"heightCm": 500})
assert "身高" in str(e.value)
def test_birth_date_must_be_a_date(self, db, user):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"birthDate": "1990/05/04"})
def test_birth_date_cannot_be_in_the_future(self, db, user):
future = (datetime.date.today() + datetime.timedelta(days=1)).isoformat()
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"birthDate": future})
def test_birth_date_cannot_be_absurdly_old(self, db, user):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"birthDate": "1850-01-01"})
def test_unknown_sex_rejected(self, db, user):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"sex": "yes"})
def test_unknown_units_rejected(self, db, user):
with pytest.raises(svc.InvalidSetting):
svc.save_settings(user["id"], {"units": "furlongs"})
class TestSnapping:
"""Picker values come from a list the backend published, so an off-list
value means a stale client — snap it rather than fail the whole save."""
def test_off_list_interval_snaps_to_nearest(self, db, user):
svc.save_settings(user["id"], {"autoSyncMinutes": 45})
assert svc.get_settings(user["id"])["autoSyncMinutes"] in svc.INTERVALS
def test_snaps_to_the_actual_nearest(self, db, user):
svc.save_settings(user["id"], {"autoSyncMinutes": 200})
assert svc.get_settings(user["id"])["autoSyncMinutes"] == 180
def test_history_zero_means_everything(self, db, user):
svc.save_settings(user["id"], {"historyDays": 0})
assert svc.get_settings(user["id"])["historyDays"] == 0
def test_negative_history_becomes_everything(self, db, user):
svc.save_settings(user["id"], {"historyDays": -30})
assert svc.get_settings(user["id"])["historyDays"] == 0
def test_garbage_interval_falls_back_to_default(self, db, user):
svc.save_settings(user["id"], {"autoSyncMinutes": "soon"})
assert svc.get_settings(user["id"])["autoSyncMinutes"] == 60
class TestDerived:
def test_age_from_birth_date(self, db, user):
born = datetime.date.today().replace(year=datetime.date.today().year - 30)
svc.save_settings(user["id"], {"birthDate": born.isoformat()})
assert svc.get_settings(user["id"])["age"] == 30
def test_age_on_the_day_before_a_birthday(self, monkeypatch):
real = datetime.date
class Frozen(real):
@classmethod
def today(cls):
return real(2026, 12, 24)
monkeypatch.setattr(svc.datetime, "date", Frozen)
assert svc.age_from("1990-12-25") == 35
def test_age_on_the_birthday(self, monkeypatch):
real = datetime.date
class Frozen(real):
@classmethod
def today(cls):
return real(2026, 12, 25)
monkeypatch.setattr(svc.datetime, "date", Frozen)
assert svc.age_from("1990-12-25") == 36
def test_bmi(self):
assert svc.bmi_from(178, 72) == 22.7
def test_bmi_needs_both(self):
assert svc.bmi_from(178, None) is None
assert svc.bmi_from(None, 72) is None
def test_bmi_appears_in_settings(self, db, user):
svc.save_settings(user["id"], {"heightCm": 180, "weightKg": 81})
assert svc.get_settings(user["id"])["bmi"] == 25.0
class TestEndpoints:
def test_get_requires_auth(self, client):
assert client.get("/api/settings").status_code == 401
def test_put_requires_auth(self, client):
assert client.put("/api/settings", json={}).status_code == 401
def test_get_returns_defaults(self, client, auth):
body = client.get("/api/settings", headers=auth).get_json()
assert body["units"] == "metric"
def test_put_saves_and_returns_the_new_state(self, client, auth):
body = client.put(
"/api/settings", headers=auth, json={"heightCm": 175, "sex": "female"}
).get_json()
assert body["heightCm"] == 175
assert body["sex"] == "female"
def test_put_rejects_bad_input_with_400(self, client, auth):
r = client.put("/api/settings", headers=auth, json={"weightKg": 999})
assert r.status_code == 400
assert "体重" in r.get_json()["error"]
def test_options_lists_only_acceptable_values(self, client, auth):
opts = client.get("/api/settings/options", headers=auth).get_json()
assert set(opts["sexes"]) == set(svc.SEXES)
assert set(opts["autoSyncMinutes"]) == set(svc.INTERVALS)
def test_every_offered_option_is_actually_accepted(self, client, auth):
"""The picker must never offer something the validator would reject."""
opts = client.get("/api/settings/options", headers=auth).get_json()
for sex in opts["sexes"]:
assert client.put("/api/settings", headers=auth,
json={"sex": sex}).status_code == 200
for unit in opts["units"]:
assert client.put("/api/settings", headers=auth,
json={"units": unit}).status_code == 200
for minutes in opts["autoSyncMinutes"]:
r = client.put("/api/settings", headers=auth,
json={"autoSyncMinutes": minutes})
assert r.get_json()["autoSyncMinutes"] == minutes
for days in opts["historyDays"]:
r = client.put("/api/settings", headers=auth, json={"historyDays": days})
assert r.get_json()["historyDays"] == days
class TestRatingBasis:
def test_requires_auth(self, client):
assert client.get("/api/settings/rating-basis").status_code == 401
def test_lists_a_source_for_every_band(self, client, auth):
body = client.get("/api/settings/rating-basis", headers=auth).get_json()
assert body["bands"], "the screen would be empty"
for entry in body["bands"]:
assert entry["metric"] and entry["bands"] and entry["source"], entry
def test_includes_the_fitness_age_method(self, client, auth):
body = client.get("/api/settings/rating-basis", headers=auth).get_json()
assert body["fitnessAge"]["steps"]
assert body["fitnessAge"]["caveat"]
def test_states_that_ai_does_not_set_thresholds(self, client, auth):
body = client.get("/api/settings/rating-basis", headers=auth).get_json()
assert "AI" in body["note"]