ericwyuan/GarminHealthLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[阶段3.1] pytest 测试基建 + auth/health/analysis 单元测试 - 102 用例全绿

Browse Source 
测试基建:
- pytest.ini / requirements-dev.txt
- tests/conftest.py: 每个用例独立 SQLite 库，提供
  db / app / client / user / auth / seed_health 六个 fixture

tests/test_auth.py (38 通过, 1 跳过):
- 密码哈希: 加盐唯一性、格式自描述、明文不入库、畸形哈希不抛异常
- 回归用例: 无 scrypt 的解释器上也能哈希（覆盖上一个 commit 的 bug）
- JWT: 过期/换密钥/篡改签名均拒绝
- 登录错误不区分"邮箱不存在"与"密码错误"（防用户枚举）
- require_auth: 缺失/畸形/过期 header 一律 401 而非 500

tests/test_health.py (30 通过):
- 日期范围上下界均为闭区间
- 数据按 user_id 隔离，查不到他人数据
- 重复 upsert 同一天不产生重复行
- 各指标端点正确剔除 NULL 行

tests/test_analysis.py (34 通过):
- 5 条建议规则的阈值边界逐条固化（8000 步 / 7 小时 / 50 压力 /
  65 静息心率 / 40 HRV）
- 均值按窗口计算而非逐日；只取最近 14 天
- 无睡眠记录的日子不被当作 0 拉低均值
- metric 名走白名单，SQL 注入串不生效
- 建议按 high/medium/low 排序

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-23 12:34:18 +08:00
parent a71c5438ce
commit 8e37e5a551
6 changed files with 745 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,98 @@
"""
Shared pytest fixtures.
Environment must be configured BEFORE `config` is imported, because config.py
reads os.environ at import time. Each test then gets its own SQLite file so
tests never share state.
"""
import os
import sys
import tempfile
import pytest
_BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _BACKEND_DIR)
os.environ.setdefault("DB_TYPE", "sqlite")
os.environ.setdefault("JWT_SECRET", "test_secret_at_least_32_bytes_long_ok")
os.environ.setdefault("CORS_ORIGIN", "http://localhost:3000")
# Point at a throwaway path; the db_path fixture overrides it per test.
os.environ.setdefault(
"DATABASE_PATH", os.path.join(tempfile.mkdtemp(), "bootstrap.db")
)
import db as db_module # noqa: E402
from app import create_app # noqa: E402
@pytest.fixture
def db(tmp_path, monkeypatch):
"""A freshly initialized, isolated SQLite database for one test."""
path = str(tmp_path / "test.db")
monkeypatch.setattr(db_module, "SQLITE_PATH", path)
db_module.init_db()
return db_module
@pytest.fixture
def app(db):
application = create_app()
application.config.update(TESTING=True)
return application
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def user(client):
"""A registered user: returns {id, email, token, password}."""
password = "secret123"
resp = client.post(
"/api/auth/register",
json={
"email": "tester@example.com",
"garminEmail": "gm@example.com",
"garminPassword": password,
},
)
assert resp.status_code == 201, resp.get_data(as_text=True)
body = resp.get_json()
return {**body, "password": password}
@pytest.fixture
def auth(user):
"""Authorization headers for the registered user."""
return {"Authorization": f"Bearer {user['token']}"}
@pytest.fixture
def seed_health(db, user):
"""Insert daily health rows. Returns the inserted records."""
def _seed(records):
for r in records:
db.execute(
"INSERT INTO health_data (id, user_id, date, steps, heart_rate, "
"heart_rate_variability, sleep_duration, sleep_quality, stress, "
"calories_burned) VALUES (?,?,?,?,?,?,?,?,?,?)",
[
f"{user['id']}-{r['date']}",
user["id"],
r["date"],
r.get("steps"),
r.get("heart_rate"),
r.get("hrv"),
r.get("sleep_duration"),
r.get("sleep_quality"),
r.get("stress"),
r.get("calories"),
],
)
return records
return _seed

213
backend/tests/test_analysis.py Normal file
View File

@@ -0,0 +1,213 @@
"""Unit tests for the trends query and the rule-based recommendation engine."""
import pytest
from services import analysis as analysis_svc
def ids(recs):
return {r["id"] for r in recs}
def days(n, **metrics):
"""n consecutive days that all carry the same metric values."""
return [
{"date": f"2026-08-{d:02d}", **metrics} for d in range(1, n + 1)
]
# --- trends -----------------------------------------------------------------
class TestTrends:
def test_empty_without_data(self, db, user):
assert analysis_svc.get_trends("steps", user["id"]) == []
def test_returns_date_value_pairs(self, seed_health, user):
seed_health([{"date": "2026-08-20", "steps": 5000}])
assert analysis_svc.get_trends("steps", user["id"]) == [
{"date": "2026-08-20", "value": 5000}
]
@pytest.mark.parametrize(
"metric,column_value",
[
("steps", {"steps": 5000}),
("heart_rate", {"heart_rate": 60}),
("sleep_duration", {"sleep_duration": 7}),
("stress", {"stress": 30}),
("calories_burned", {"calories": 250}),
],
)
def test_each_supported_metric(self, seed_health, user, metric, column_value):
seed_health([{"date": "2026-08-20", **column_value}])
rows = analysis_svc.get_trends(metric, user["id"])
assert len(rows) == 1 and rows[0]["value"] is not None
def test_unknown_metric_falls_back_to_steps(self, seed_health, user):
seed_health([{"date": "2026-08-20", "steps": 5000}])
assert analysis_svc.get_trends("bogus", user["id"]) == [
{"date": "2026-08-20", "value": 5000}
]
def test_unknown_metric_cannot_inject_sql(self, seed_health, user):
"""The metric name indexes a whitelist; it is never interpolated raw."""
seed_health([{"date": "2026-08-20", "steps": 5000}])
rows = analysis_svc.get_trends(
"steps FROM health_data; DROP TABLE health_data --", user["id"]
)
assert rows == [{"date": "2026-08-20", "value": 5000}]
def test_nulls_excluded(self, seed_health, user):
seed_health([
{"date": "2026-08-20", "steps": 5000},
{"date": "2026-08-21"},
])
assert len(analysis_svc.get_trends("steps", user["id"])) == 1
def test_date_range_filter(self, seed_health, user):
seed_health([
{"date": "2026-08-20", "steps": 1},
{"date": "2026-08-21", "steps": 2},
{"date": "2026-08-22", "steps": 3},
])
rows = analysis_svc.get_trends(
"steps", user["id"], "2026-08-21", "2026-08-21"
)
assert rows == [{"date": "2026-08-21", "value": 2}]
# --- recommendations --------------------------------------------------------
class TestNoData:
def test_returns_a_single_guidance_item(self, db, user):
recs = analysis_svc.get_recommendations(user["id"])
assert len(recs) == 1
assert recs[0]["id"] == "no-data"
assert recs[0]["priority"] == "low"
class TestStepsRule:
def test_fires_below_8000(self, seed_health, user):
seed_health(days(5, steps=6000))
assert "steps" in ids(analysis_svc.get_recommendations(user["id"]))
def test_silent_at_or_above_8000(self, seed_health, user):
seed_health(days(5, steps=8000))
assert "steps" not in ids(analysis_svc.get_recommendations(user["id"]))
def test_priority_is_medium(self, seed_health, user):
seed_health(days(5, steps=6000))
rec = next(
r for r in analysis_svc.get_recommendations(user["id"]) if r["id"] == "steps"
)
assert rec["priority"] == "medium"
assert rec["basedOn"] == ["steps"]
def test_averages_across_days_not_per_day(self, seed_health, user):
# 4000 and 12000 average to 8000 -> rule must NOT fire.
seed_health([
{"date": "2026-08-01", "steps": 4000},
{"date": "2026-08-02", "steps": 12000},
])
assert "steps" not in ids(analysis_svc.get_recommendations(user["id"]))
class TestSleepRule:
def test_fires_below_7_hours(self, seed_health, user):
seed_health(days(5, sleep_duration=6))
assert "sleep" in ids(analysis_svc.get_recommendations(user["id"]))
def test_silent_at_7_hours(self, seed_health, user):
seed_health(days(5, sleep_duration=7))
assert "sleep" not in ids(analysis_svc.get_recommendations(user["id"]))
def test_priority_is_high(self, seed_health, user):
seed_health(days(5, sleep_duration=5))
rec = next(
r for r in analysis_svc.get_recommendations(user["id"]) if r["id"] == "sleep"
)
assert rec["priority"] == "high"
def test_days_without_sleep_do_not_drag_the_average_down(self, seed_health, user):
"""Nights with no sleep record must be excluded, not counted as zero."""
seed_health([
{"date": "2026-08-01", "sleep_duration": 8},
{"date": "2026-08-02", "steps": 5000}, # no sleep recorded
])
assert "sleep" not in ids(analysis_svc.get_recommendations(user["id"]))
class TestStressRule:
def test_fires_above_50(self, seed_health, user):
seed_health(days(5, stress=60))
assert "stress" in ids(analysis_svc.get_recommendations(user["id"]))
def test_silent_at_50(self, seed_health, user):
seed_health(days(5, stress=50))
assert "stress" not in ids(analysis_svc.get_recommendations(user["id"]))
class TestRestingHeartRateRule:
def test_fires_above_65(self, seed_health, user):
seed_health(days(5, heart_rate=70))
assert "rhr" in ids(analysis_svc.get_recommendations(user["id"]))
def test_silent_at_65(self, seed_health, user):
seed_health(days(5, heart_rate=65))
assert "rhr" not in ids(analysis_svc.get_recommendations(user["id"]))
class TestHrvRule:
def test_fires_below_40(self, seed_health, user):
seed_health(days(5, hrv=30))
assert "hrv" in ids(analysis_svc.get_recommendations(user["id"]))
def test_silent_at_40(self, seed_health, user):
seed_health(days(5, hrv=40))
assert "hrv" not in ids(analysis_svc.get_recommendations(user["id"]))
class TestHealthyUser:
def test_all_good_returns_the_positive_message(self, seed_health, user):
seed_health(days(5, steps=10000, sleep_duration=8, stress=30,
heart_rate=55, hrv=60))
recs = analysis_svc.get_recommendations(user["id"])
assert ids(recs) == {"good"}
class TestOrderingAndWindow:
def test_sorted_high_medium_low(self, seed_health, user):
seed_health(days(5, steps=6000, sleep_duration=5, hrv=30))
order = {"high": 0, "medium": 1, "low": 2}
priorities = [r["priority"] for r in analysis_svc.get_recommendations(user["id"])]
assert priorities == sorted(priorities, key=lambda p: order[p])
assert priorities[0] == "high"
def test_only_the_last_14_days_count(self, seed_health, user):
"""20 lazy days then 14 active ones: the old days must not pull it down."""
old = [{"date": f"2026-07-{d:02d}", "steps": 1000} for d in range(1, 21)]
recent = [{"date": f"2026-08-{d:02d}", "steps": 12000} for d in range(1, 15)]
seed_health(old + recent)
assert "steps" not in ids(analysis_svc.get_recommendations(user["id"]))
def test_multiple_rules_can_fire_together(self, seed_health, user):
seed_health(days(5, steps=5000, sleep_duration=5, stress=70, heart_rate=75))
assert {"steps", "sleep", "stress", "rhr"} <= ids(
analysis_svc.get_recommendations(user["id"])
)
class TestEndpoints:
def test_trends_requires_auth(self, client):
assert client.get("/api/analysis/trends").status_code == 401
def test_recommendations_requires_auth(self, client):
assert client.get("/api/analysis/recommendations").status_code == 401
def test_trends_endpoint(self, client, auth, seed_health):
seed_health(days(3, steps=5000))
r = client.get("/api/analysis/trends?metricType=steps", headers=auth)
assert r.status_code == 200 and len(r.get_json()) == 3
def test_recommendations_endpoint(self, client, auth, seed_health):
seed_health(days(3, steps=5000))
r = client.get("/api/analysis/recommendations", headers=auth)
assert r.status_code == 200
assert "steps" in {x["id"] for x in r.get_json()}

233
backend/tests/test_auth.py Normal file
View File

@@ -0,0 +1,233 @@
"""Unit tests for password hashing, JWT handling, and the auth endpoints."""
import datetime
import hashlib
import os
import jwt
import pytest
import auth
from config import JWT_SECRET
# --- password hashing -------------------------------------------------------
class TestPasswordHashing:
def test_hash_is_verifiable(self):
stored = auth.hash_password("correct horse")
assert auth.verify_password("correct horse", stored) is True
def test_wrong_password_rejected(self):
stored = auth.hash_password("correct horse")
assert auth.verify_password("wrong horse", stored) is False
def test_salt_makes_hashes_unique(self):
a = auth.hash_password("same")
b = auth.hash_password("same")
assert a != b, "identical passwords must not produce identical hashes"
def test_hash_format_is_self_describing(self):
stored = auth.hash_password("pw")
prefix, iterations, salt, digest = stored.split("$")
assert prefix == auth.PBKDF2_PREFIX
assert int(iterations) == auth.PBKDF2_ITERATIONS
assert len(bytes.fromhex(salt)) == 16
assert len(bytes.fromhex(digest)) == 64
def test_password_never_appears_in_hash(self):
stored = auth.hash_password("supersecret")
assert "supersecret" not in stored
@pytest.mark.parametrize(
"stored",
["", None, "garbage", "pbkdf2_sha256$notanint$aa$bb", "nothex:nothex"],
)
def test_malformed_hashes_rejected_not_raised(self, stored):
assert auth.verify_password("anything", stored) is False
def test_hashing_does_not_require_scrypt(self, monkeypatch):
"""Regression: macOS system Python (LibreSSL) has no hashlib.scrypt.
Registration used to raise AttributeError -> HTTP 500 on those builds.
"""
monkeypatch.delattr(hashlib, "scrypt", raising=False)
stored = auth.hash_password("pw")
assert auth.verify_password("pw", stored) is True
def test_unicode_password(self):
stored = auth.hash_password("密码🔒")
assert auth.verify_password("密码🔒", stored) is True
assert auth.verify_password("密码", stored) is False
@pytest.mark.skipif(
not hasattr(hashlib, "scrypt"), reason="interpreter built without scrypt"
)
def test_legacy_scrypt_hash_still_verifies(self):
salt = os.urandom(16)
digest = hashlib.scrypt(
b"legacy", salt=salt, n=16384, r=8, p=1, dklen=64
).hex()
assert auth.verify_password("legacy", f"{salt.hex()}:{digest}") is True
assert auth.verify_password("nope", f"{salt.hex()}:{digest}") is False
# --- JWT --------------------------------------------------------------------
class TestTokens:
def test_sign_and_verify_roundtrip(self):
token = auth.sign_token("user-123")
assert auth.verify_token(token) == {"user_id": "user-123"}
def test_expired_token_rejected(self):
past = datetime.datetime.utcnow() - datetime.timedelta(days=1)
token = jwt.encode(
{"sub": "u", "iat": past, "exp": past}, JWT_SECRET, algorithm="HS256"
)
with pytest.raises(jwt.ExpiredSignatureError):
auth.verify_token(token)
def test_token_signed_with_other_secret_rejected(self):
token = jwt.encode({"sub": "u"}, "a-different-secret", algorithm="HS256")
with pytest.raises(jwt.InvalidTokenError):
auth.verify_token(token)
def test_tampered_token_rejected(self):
token = auth.sign_token("user-123")
head, payload, sig = token.split(".")
with pytest.raises(jwt.InvalidTokenError):
auth.verify_token(f"{head}.{payload}.{sig[:-2]}xx")
# --- register ---------------------------------------------------------------
class TestRegister:
def test_returns_201_and_token(self, client):
r = client.post(
"/api/auth/register",
json={
"email": "a@example.com",
"garminEmail": "g@example.com",
"garminPassword": "pw123456",
},
)
assert r.status_code == 201
body = r.get_json()
assert body["email"] == "a@example.com"
assert auth.verify_token(body["token"])["user_id"] == body["id"]
def test_duplicate_email_conflicts(self, client, user):
r = client.post(
"/api/auth/register",
json={
"email": user["email"],
"garminEmail": "other@example.com",
"garminPassword": "pw123456",
},
)
assert r.status_code == 409
@pytest.mark.parametrize(
"payload",
[
{},
{"email": "a@example.com"},
{"email": "a@example.com", "garminEmail": "g@example.com"},
{"email": "", "garminEmail": "g@example.com", "garminPassword": "x"},
],
)
def test_missing_fields_rejected(self, client, payload):
assert client.post("/api/auth/register", json=payload).status_code == 400
def test_password_stored_only_as_hash(self, client, db):
client.post(
"/api/auth/register",
json={
"email": "h@example.com",
"garminEmail": "g@example.com",
"garminPassword": "plaintext-secret",
},
)
row = db.query_one(
"SELECT garmin_password_hash FROM users WHERE email = ?", ["h@example.com"]
)
assert "plaintext-secret" not in row["garmin_password_hash"]
assert auth.verify_password("plaintext-secret", row["garmin_password_hash"])
# --- login ------------------------------------------------------------------
class TestLogin:
def test_valid_credentials(self, client, user):
r = client.post(
"/api/auth/login",
json={"email": user["email"], "password": user["password"]},
)
assert r.status_code == 200
assert r.get_json()["id"] == user["id"]
def test_wrong_password(self, client, user):
r = client.post(
"/api/auth/login", json={"email": user["email"], "password": "nope"}
)
assert r.status_code == 401
def test_unknown_email(self, client):
r = client.post(
"/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
)
assert r.status_code == 401
def test_error_does_not_reveal_which_field_was_wrong(self, client, user):
unknown = client.post(
"/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
).get_json()
bad_pw = client.post(
"/api/auth/login", json={"email": user["email"], "password": "nope"}
).get_json()
assert unknown == bad_pw, "responses must not distinguish the two cases"
# --- require_auth -----------------------------------------------------------
class TestRequireAuth:
def test_missing_header(self, client):
assert client.get("/api/health/summary").status_code == 401
@pytest.mark.parametrize(
"header",
["", "Bearer", "Token abc", "bearer abc", "Bearer not.a.jwt"],
)
def test_malformed_headers(self, client, header):
r = client.get("/api/health/summary", headers={"Authorization": header})
assert r.status_code == 401
def test_expired_token_gets_401_not_500(self, client):
past = datetime.datetime.utcnow() - datetime.timedelta(days=1)
token = jwt.encode(
{"sub": "u", "iat": past, "exp": past}, JWT_SECRET, algorithm="HS256"
)
r = client.get(
"/api/health/summary", headers={"Authorization": f"Bearer {token}"}
)
assert r.status_code == 401
assert "expired" in r.get_json()["error"]
def test_valid_token_passes(self, client, auth):
assert client.get("/api/health/summary", headers=auth).status_code == 200
class TestLogoutAndRefresh:
def test_logout_clears_stored_token(self, client, auth, user, db):
assert client.post("/api/auth/logout", headers=auth).status_code == 200
row = db.query_one("SELECT jwt_token FROM users WHERE id = ?", [user["id"]])
assert row["jwt_token"] is None
def test_refresh_returns_usable_token(self, client, auth):
r = client.post("/api/auth/refresh", headers=auth)
assert r.status_code == 200
new_token = r.get_json()["token"]
assert (
client.get(
"/api/health/summary",
headers={"Authorization": f"Bearer {new_token}"},
).status_code
== 200
)
def test_refresh_requires_auth(self, client):
assert client.post("/api/auth/refresh").status_code == 401

189
backend/tests/test_health.py Normal file
View File

@@ -0,0 +1,189 @@
"""Unit tests for the health data service and its endpoints."""
import pytest
from services import health as health_svc
DAYS = [
{"date": "2026-08-20", "steps": 6500, "heart_rate": 70, "hrv": 45,
"sleep_duration": 6, "sleep_quality": 80, "stress": 55, "calories": 260},
{"date": "2026-08-21", "steps": 9000, "heart_rate": 62, "hrv": 46,
"sleep_duration": 8, "sleep_quality": 79, "stress": 40, "calories": 360},
{"date": "2026-08-22", "steps": 7500, "heart_rate": 68, "hrv": 47,
"sleep_duration": 7, "sleep_quality": 78, "stress": 48, "calories": 300},
]
class TestGetSummary:
def test_empty_for_new_user(self, db, user):
assert health_svc.get_summary(user["id"]) == []
def test_returns_all_rows_ordered_by_date(self, seed_health, user):
seed_health(DAYS)
rows = health_svc.get_summary(user["id"])
assert [r["date"] for r in rows] == [
"2026-08-20", "2026-08-21", "2026-08-22"
]
def test_maps_to_camel_case(self, seed_health, user):
seed_health(DAYS)
row = health_svc.get_summary(user["id"])[0]
assert row["heartRate"] == 70
assert row["heartRateVariability"] == 45
assert row["caloriesBurned"] == 260
assert row["sleep"] == {"duration": 6, "quality": 80}
def test_sleep_is_none_when_absent(self, seed_health, user):
seed_health([{"date": "2026-08-20", "steps": 100}])
assert health_svc.get_summary(user["id"])[0]["sleep"] is None
def test_start_date_filter_is_inclusive(self, seed_health, user):
seed_health(DAYS)
rows = health_svc.get_summary(user["id"], start="2026-08-21")
assert [r["date"] for r in rows] == ["2026-08-21", "2026-08-22"]
def test_end_date_filter_is_inclusive(self, seed_health, user):
seed_health(DAYS)
rows = health_svc.get_summary(user["id"], end="2026-08-21")
assert [r["date"] for r in rows] == ["2026-08-20", "2026-08-21"]
def test_both_bounds(self, seed_health, user):
seed_health(DAYS)
rows = health_svc.get_summary(user["id"], "2026-08-21", "2026-08-21")
assert len(rows) == 1
def test_range_with_no_matches(self, seed_health, user):
seed_health(DAYS)
assert health_svc.get_summary(user["id"], "2027-01-01") == []
def test_scoped_to_the_requesting_user(self, seed_health, user, db):
seed_health(DAYS)
db.execute(
"INSERT INTO users (id, email, garmin_email, garmin_password_hash) "
"VALUES (?,?,?,?)",
["other-user", "other@example.com", "o@example.com", "x"],
)
db.execute(
"INSERT INTO health_data (id, user_id, date, steps) VALUES (?,?,?,?)",
["other-1", "other-user", "2026-08-20", 99999],
)
rows = health_svc.get_summary(user["id"])
assert all(r["steps"] != 99999 for r in rows)
assert len(rows) == 3
class TestMetricReads:
"""Metric endpoints must drop rows where that metric is NULL."""
def test_steps_excludes_nulls(self, seed_health, user):
seed_health(DAYS + [{"date": "2026-08-23", "heart_rate": 60}])
rows = health_svc.get_steps(user["id"])
assert len(rows) == 3
assert all(r["steps"] is not None for r in rows)
def test_heart_rate_excludes_nulls(self, seed_health, user):
seed_health(DAYS + [{"date": "2026-08-23", "steps": 100}])
rows = health_svc.get_heart_rate(user["id"])
assert len(rows) == 3
def test_sleep_excludes_nulls(self, seed_health, user):
seed_health(DAYS + [{"date": "2026-08-23", "steps": 100}])
rows = health_svc.get_sleep(user["id"])
assert len(rows) == 3
assert rows[0] == {"date": "2026-08-20", "duration": 6, "quality": 80}
class TestUpsert:
def test_insert_creates_row(self, db, user):
health_svc.upsert_health_daily(
user["id"], {"date": "2026-08-20", "steps": 5000}
)
rows = health_svc.get_summary(user["id"])
assert len(rows) == 1 and rows[0]["steps"] == 5000
def test_second_upsert_updates_instead_of_duplicating(self, db, user):
health_svc.upsert_health_daily(
user["id"], {"date": "2026-08-20", "steps": 5000}
)
health_svc.upsert_health_daily(
user["id"], {"date": "2026-08-20", "steps": 8000}
)
rows = health_svc.get_summary(user["id"])
assert len(rows) == 1, "re-syncing a day must not duplicate it"
assert rows[0]["steps"] == 8000
def test_upsert_is_deterministic_by_user_and_date(self, db, user):
a = health_svc.upsert_health_daily(user["id"], {"date": "2026-08-20"})
b = health_svc.upsert_health_daily(user["id"], {"date": "2026-08-20"})
assert a == b
def test_missing_metrics_stored_as_null(self, db, user):
health_svc.upsert_health_daily(
user["id"], {"date": "2026-08-20", "steps": 100}
)
row = health_svc.get_summary(user["id"])[0]
assert row["heartRate"] is None
assert row["sleep"] is None
class TestActivities:
def test_insert_and_read_back(self, db, user):
health_svc.insert_activity(
user["id"],
{
"activityType": "running",
"startTime": "2026-08-20T07:00:00",
"endTime": "2026-08-20T07:30:00",
"duration": 1800,
"distance": 5.0,
"calories": 320,
"heartRateAverage": 140,
"heartRateMax": 165,
},
)
rows = health_svc.get_activities(user["id"])
assert len(rows) == 1
assert rows[0]["activity_type"] == "running"
assert rows[0]["distance"] == 5.0
def test_ids_are_unique_per_insert(self, db, user):
base = {
"activityType": "running",
"startTime": "2026-08-20T07:00:00",
"endTime": "2026-08-20T07:30:00",
}
assert health_svc.insert_activity(user["id"], base) != health_svc.insert_activity(
user["id"], base
)
class TestEndpoints:
@pytest.mark.parametrize(
"path",
["/api/health/summary", "/api/health/steps", "/api/health/heart-rate",
"/api/health/sleep", "/api/health/activities"],
)
def test_require_authentication(self, client, path):
assert client.get(path).status_code == 401
@pytest.mark.parametrize(
"path",
["/api/health/summary", "/api/health/steps", "/api/health/heart-rate",
"/api/health/sleep", "/api/health/activities"],
)
def test_return_json_list_when_authenticated(self, client, auth, path):
r = client.get(path, headers=auth)
assert r.status_code == 200
assert isinstance(r.get_json(), list)
def test_summary_reflects_seeded_data(self, client, auth, seed_health):
seed_health(DAYS)
body = client.get("/api/health/summary", headers=auth).get_json()
assert len(body) == 3
def test_query_params_are_applied(self, client, auth, seed_health):
seed_health(DAYS)
body = client.get(
"/api/health/summary?startDate=2026-08-22", headers=auth
).get_json()
assert len(body) == 1