8e37e5a551
测试基建: - 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>
214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
"""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()}
|