6dd070ec9b
生产上 trends 队列里积了 36 个任务，subject 是 2026-09-01:1033、:1039、 :1044……一路涨。这台账号当时正在补历史，get_summary 的行数每隔几分钟就变， 而我把 len(rows) 写进了 subject——subject 同时是缓存键和任务队列的键，一变 就是一条全新的任务，轮询几次就刷出十几条。 subject 该回答的是「这条解读是关于什么的」，不是「当时有多少行数据」。 数据变化本来就由 fingerprint 负责。 - trends 的 subject 改成快照日期；sleep 用配置的窗口常量而不是实际夜数 （缺一晚也不该换键）；challenges 用固定键 - 加了不变量测试：补一天历史数据后 subject 不许变；任何 subject 段都不许 长得像行数 顺带加一层兜底 jobs.supersede()：单实例 scope 只该有一个在跑的 subject， 队列里同 kind 的其它 pending 任务是关于已经不存在的快照的，跑完也没人看。 per_item 的 daily / activity 不受影响——它们本来就一天一条、一次运动一条。 兜底不是机制，机制是 subject 稳定；它存在只是因为这次 subject 不稳定，而 36 条任务堆在那里之前没人发现。 顺带按要求把 AiPanel 改成默认精简：只显示标题、来源和一句话结论，点「展开 详细」才出要点/建议/依据，可再收起——和今日晨报卡片一致。这些面板压在本来 就很密的图表页上面，全部默认展开会把真正的数据一次性挤到屏幕外。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1056 lines
45 KiB
Python
1056 lines
45 KiB
Python
"""
|
|
Unit tests for the AI coach: feature engineering, prompt parsing, and the
|
|
briefing / trend-insight / Copilot endpoints.
|
|
|
|
Every model call is mocked. The suite never reaches the ai-gateway, so it is
|
|
neither slow nor dependent on that box being up.
|
|
"""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from services import ai as ai_svc
|
|
from services import analysis as analysis_svc
|
|
from services import coach
|
|
from services import insights
|
|
from services import jobs
|
|
from services import scopes
|
|
|
|
|
|
def day(date, **metrics):
|
|
"""One row in the shape `health.get_summary` returns."""
|
|
sleep = metrics.pop("sleep", None)
|
|
row = {"date": date, **metrics}
|
|
row["sleep"] = sleep
|
|
return row
|
|
|
|
|
|
def flat_days(n, start=1, **series):
|
|
"""`n` consecutive days from 2026-08-01, each metric a constant or list."""
|
|
rows = []
|
|
for i in range(n):
|
|
values = {}
|
|
for key, value in series.items():
|
|
values[key] = value[i] if isinstance(value, list) else value
|
|
rows.append(day(f"2026-08-{start + i:02d}", **values))
|
|
return rows
|
|
|
|
|
|
# --- feature engineering ----------------------------------------------------
|
|
class TestFlatten:
|
|
def test_sleep_stages_become_percentages_of_time_asleep(self):
|
|
row = day("2026-08-01", sleep={
|
|
"duration": 8.0, "quality": 80, "deepSeconds": 3600,
|
|
"remSeconds": 7200, "lightSeconds": None, "awakeSeconds": None,
|
|
})
|
|
flat = insights._flatten(row)
|
|
assert flat["sleepDeepPct"] == 12.5
|
|
assert flat["sleepRemPct"] == 25.0
|
|
|
|
def test_missing_stage_is_absent_not_zero(self):
|
|
flat = insights._flatten(day("2026-08-01", sleep={"duration": 7.0}))
|
|
assert "sleepDeepPct" not in flat
|
|
|
|
def test_no_sleep_record_leaves_no_sleep_fields(self):
|
|
flat = insights._flatten(day("2026-08-01", steps=100))
|
|
assert "sleepDuration" not in flat
|
|
|
|
def test_sedentary_seconds_become_hours(self):
|
|
flat = insights._flatten(day("2026-08-01", sedentarySeconds=5400))
|
|
assert flat["sedentaryHours"] == 1.5
|
|
|
|
|
|
class TestDeviations:
|
|
def test_z_score_measures_departure_from_the_personal_baseline(self):
|
|
# A baseline that varies, as real data does: mean 1000, sd 100.
|
|
baseline = [900, 1000, 1100, 900, 1000, 1100, 900, 1000, 1100, 1000]
|
|
history = [
|
|
insights._flatten(r) for r in flat_days(11, steps=baseline + [1800])
|
|
]
|
|
result = {d["metric"]: d for d in insights.deviations(history, history[-1])}
|
|
assert result["steps"]["baselineMean"] == 1000
|
|
assert result["steps"]["z"] > 3
|
|
|
|
def test_today_is_excluded_from_its_own_baseline(self):
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
8, heartRate=[60, 60, 60, 60, 60, 60, 60, 70]
|
|
)]
|
|
result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
|
|
# Including today would pull the mean up to 61.25 and shrink the z.
|
|
assert result["heartRate"]["baselineMean"] == 60
|
|
|
|
def test_too_little_history_reports_insufficient_baseline(self):
|
|
rows = [insights._flatten(r) for r in flat_days(3, steps=5000)]
|
|
result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
|
|
assert result["steps"]["verdict"] == "基线不足"
|
|
assert result["steps"]["z"] is None
|
|
|
|
def test_a_flat_baseline_reports_no_z_rather_than_a_fabricated_zero(self):
|
|
"""Dividing by a zero standard deviation is undefined; calling the day
|
|
'z = 0' would label a genuine departure as perfectly typical."""
|
|
rows = [insights._flatten(r) for r in flat_days(8, steps=[5000] * 7 + [9000])]
|
|
result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
|
|
assert result["steps"]["z"] is None
|
|
assert result["steps"]["verdict"] == "基线无波动"
|
|
|
|
def test_a_flat_baseline_matched_exactly_is_just_normal(self):
|
|
rows = [insights._flatten(r) for r in flat_days(8, steps=5000)]
|
|
result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
|
|
assert result["steps"]["verdict"] == "正常"
|
|
|
|
def test_direction_is_judged_per_metric_not_by_sign(self):
|
|
wobble = [58, 60, 62, 58, 60, 62, 60]
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
8,
|
|
heartRate=wobble + [75],
|
|
heartRateVariability=[38, 40, 42, 38, 40, 42, 40] + [55],
|
|
)]
|
|
result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
|
|
# Both moved up; only one of them is good news.
|
|
assert result["heartRate"]["verdict"] == "偏差"
|
|
assert result["heartRateVariability"]["verdict"] == "偏好"
|
|
|
|
def test_largest_departure_comes_first(self):
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
8, steps=[5000] * 7 + [5100], heartRate=[60] * 7 + [80]
|
|
)]
|
|
result = insights.deviations(rows, rows[-1])
|
|
assert result[0]["metric"] == "heartRate"
|
|
|
|
def test_metrics_absent_today_are_omitted(self):
|
|
rows = [insights._flatten(r) for r in flat_days(8, steps=5000)]
|
|
assert all(d["metric"] == "steps" for d in insights.deviations(rows, rows[-1]))
|
|
|
|
|
|
class TestTrends:
|
|
def test_slope_is_reported_per_thirty_days(self):
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
30, heartRateVariability=[40 + i for i in range(30)]
|
|
)]
|
|
entry = {t["metric"]: t for t in insights.trends(rows)}
|
|
# One unit a day is 30 per 30 days.
|
|
assert entry["heartRateVariability"]["slopePer30d"] == pytest.approx(30, abs=0.5)
|
|
|
|
def test_endpoint_windows_do_not_overlap_on_short_histories(self):
|
|
"""A 30-day history must not report delta 0 for a line that moved."""
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
30, steps=[1000 + i * 100 for i in range(30)]
|
|
)]
|
|
entry = {t["metric"]: t for t in insights.trends(rows)}["steps"]
|
|
assert entry["firstMean"] < entry["lastMean"]
|
|
assert entry["delta"] > 0
|
|
|
|
def test_direction_respects_which_way_is_better(self):
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
30, heartRate=[80 - i for i in range(30)]
|
|
)]
|
|
entry = {t["metric"]: t for t in insights.trends(rows)}["heartRate"]
|
|
assert entry["direction"] == "改善"
|
|
|
|
def test_metrics_with_too_few_samples_are_skipped(self):
|
|
rows = [insights._flatten(r) for r in flat_days(5, steps=1000)]
|
|
assert insights.trends(rows) == []
|
|
|
|
def test_gaps_do_not_compress_the_x_axis(self):
|
|
"""Ordinal dates, not indices: the same rise spread over a longer span
|
|
is a gentler slope, and indices would score the two identically."""
|
|
values = [1000 + i * 100 for i in range(10)]
|
|
dense = [(f"2026-08-{1 + i:02d}", v) for i, v in enumerate(values)]
|
|
# Same ten readings, but the last five sit a month later.
|
|
gapped = dense[:5] + [
|
|
(f"2026-09-{6 + i:02d}", v) for i, v in enumerate(values[5:])
|
|
]
|
|
assert insights._slope_per_30d(gapped) < insights._slope_per_30d(dense)
|
|
|
|
|
|
class TestActivityShift:
|
|
def test_compares_the_last_week_with_the_weeks_before_it(self):
|
|
rows = [insights._flatten(r) for r in flat_days(
|
|
20, steps=[10000] * 13 + [5000] * 7
|
|
)]
|
|
shift = insights.activity_shift(rows)
|
|
assert shift["steps"]["recentMean"] == 5000
|
|
assert shift["steps"]["priorMean"] == 10000
|
|
assert shift["steps"]["changePct"] == -50.0
|
|
|
|
def test_absent_when_there_is_not_enough_history(self):
|
|
rows = [insights._flatten(r) for r in flat_days(6, steps=8000)]
|
|
assert "steps" not in insights.activity_shift(rows)
|
|
|
|
|
|
class TestWindowContext:
|
|
def test_unknown_metric_returns_nothing(self, db, user):
|
|
assert insights.window_context(
|
|
user["id"], "notAMetric", "2026-08-01", "2026-08-30"
|
|
) is None
|
|
|
|
def test_empty_span_returns_nothing(self, db, user):
|
|
assert insights.window_context(
|
|
user["id"], "steps", "2020-01-01", "2020-01-31"
|
|
) is None
|
|
|
|
|
|
class TestBuildContext:
|
|
def test_returns_nothing_without_data(self, db, user):
|
|
assert insights.build_context(user["id"]) is None
|
|
|
|
def test_defaults_to_the_newest_recorded_day(self, db, user, seed_health):
|
|
seed_health([
|
|
{"date": "2026-08-01", "steps": 5000},
|
|
{"date": "2026-08-02", "steps": 6000},
|
|
])
|
|
context = insights.build_context(user["id"])
|
|
assert context["snapshotDate"] == "2026-08-02"
|
|
|
|
def test_an_unknown_date_is_not_silently_replaced(self, db, user, seed_health):
|
|
seed_health([{"date": "2026-08-01", "steps": 5000}])
|
|
assert insights.build_context(user["id"], "2026-08-09") is None
|
|
|
|
def test_stays_small_enough_to_prompt_with(self, db, user, seed_health):
|
|
seed_health([
|
|
{"date": f"2026-08-{d:02d}", "steps": 8000 + d, "heart_rate": 60,
|
|
"hrv": 45, "sleep_duration": 7, "stress": 30}
|
|
for d in range(1, 31)
|
|
])
|
|
context = insights.build_context(user["id"])
|
|
blob = json.dumps(context, ensure_ascii=False)
|
|
# A month of history has to cost thousands of characters, not tens of
|
|
# thousands — the whole point of computing features server-side.
|
|
assert len(blob) < 20_000
|
|
|
|
|
|
# --- reply parsing ----------------------------------------------------------
|
|
GOOD_BRIEFING = {
|
|
"status": "中等偏上",
|
|
"headline": "恢复尚可，睡眠偏短。",
|
|
"diagnosis": [{"title": "睡眠结构", "detail": "睡眠 6 小时，低于目标。"}],
|
|
"shortfall": "睡眠不足",
|
|
"prescription": {
|
|
"intensity": "中等", "hrZone": "Zone 2~Zone 3",
|
|
"suggestion": "40 分钟慢跑", "durationMin": 40, "avoid": "高强度间歇",
|
|
},
|
|
"actions": ["提前 30 分钟入睡", "午后避免咖啡因"],
|
|
}
|
|
|
|
|
|
class TestParseBriefing:
|
|
def test_plain_json(self):
|
|
out = coach.parse_briefing(json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
assert out["status"] == "中等偏上"
|
|
assert out["prescription"]["durationMin"] == 40
|
|
assert out["actions"] == ["提前 30 分钟入睡", "午后避免咖啡因"]
|
|
|
|
def test_answer_is_taken_from_after_a_reasoning_preamble(self):
|
|
"""The gateway's primary upstream narrates its thinking first."""
|
|
reply = (
|
|
'The user wants a briefing. Let me consider {"draft": true} first.\n'
|
|
"Actually I should output the final object now:\n"
|
|
+ json.dumps(GOOD_BRIEFING, ensure_ascii=False)
|
|
)
|
|
assert coach.parse_briefing(reply)["status"] == "中等偏上"
|
|
|
|
def test_markdown_fences_are_tolerated(self):
|
|
reply = "```json\n" + json.dumps(GOOD_BRIEFING, ensure_ascii=False) + "\n```"
|
|
assert coach.parse_briefing(reply)["headline"] == "恢复尚可，睡眠偏短。"
|
|
|
|
def test_missing_prescription_does_not_raise(self):
|
|
payload = {k: v for k, v in GOOD_BRIEFING.items() if k != "prescription"}
|
|
out = coach.parse_briefing(json.dumps(payload, ensure_ascii=False))
|
|
assert out["prescription"]["suggestion"] is None
|
|
|
|
def test_non_numeric_duration_becomes_none(self):
|
|
payload = json.loads(json.dumps(GOOD_BRIEFING))
|
|
payload["prescription"]["durationMin"] = "四十分钟"
|
|
assert coach.parse_briefing(json.dumps(payload))["prescription"]["durationMin"] is None
|
|
|
|
def test_diagnosis_written_as_plain_strings_is_accepted(self):
|
|
payload = json.loads(json.dumps(GOOD_BRIEFING))
|
|
payload["diagnosis"] = ["睡眠偏短。"]
|
|
out = coach.parse_briefing(json.dumps(payload, ensure_ascii=False))
|
|
assert out["diagnosis"][0]["detail"] == "睡眠偏短。"
|
|
|
|
def test_an_empty_briefing_is_rejected_rather_than_rendered_blank(self):
|
|
with pytest.raises(ai_svc.AIError):
|
|
coach.parse_briefing(json.dumps({"status": "好"}))
|
|
|
|
def test_prose_without_json_raises(self):
|
|
with pytest.raises(ai_svc.AIError):
|
|
coach.parse_briefing("今天状态不错，可以正常训练。")
|
|
|
|
|
|
class TestParseTrendInsight:
|
|
def test_valid_reply(self):
|
|
reply = json.dumps({
|
|
"summary": "HRV 稳步上升。",
|
|
"drivers": [{"factor": "有氧负荷", "detail": "区间内 8 次有氧。"}],
|
|
"caution": None, "confidence": "high",
|
|
}, ensure_ascii=False)
|
|
out = coach.parse_trend_insight(reply)
|
|
assert out["confidence"] == "high"
|
|
assert out["drivers"][0]["factor"] == "有氧负荷"
|
|
|
|
def test_unknown_confidence_falls_back_to_medium(self):
|
|
reply = json.dumps({"summary": "上升。", "confidence": "很高"}, ensure_ascii=False)
|
|
assert coach.parse_trend_insight(reply)["confidence"] == "medium"
|
|
|
|
def test_empty_reply_raises(self):
|
|
with pytest.raises(ai_svc.AIError):
|
|
coach.parse_trend_insight(json.dumps({"confidence": "high"}))
|
|
|
|
|
|
class TestExtractJson:
|
|
def test_last_object_wins_over_an_earlier_draft(self):
|
|
assert ai_svc.extract_json('{"a": 1} then {"a": 2}') == {"a": 2}
|
|
|
|
def test_braces_inside_strings_do_not_break_the_scan(self):
|
|
assert ai_svc.extract_json('思考 } 中。{"t": "含 } 的文本"}')["t"] == "含 } 的文本"
|
|
|
|
def test_escaped_quote_inside_a_string(self):
|
|
assert ai_svc.extract_json(r'x {"t": "a \" b"}')["t"] == 'a " b'
|
|
|
|
def test_arrays_are_extracted_too(self):
|
|
assert ai_svc.extract_json("preamble [1, 2, 3]") == [1, 2, 3]
|
|
|
|
def test_empty_reply_raises(self):
|
|
with pytest.raises(ai_svc.AIError):
|
|
ai_svc.extract_json(" ")
|
|
|
|
|
|
# --- prompt assembly --------------------------------------------------------
|
|
class TestPrompts:
|
|
def test_system_prompt_forbids_inventing_numbers(self):
|
|
assert "禁止编造" in coach.SYSTEM
|
|
|
|
def test_system_prompt_disclaims_medical_diagnosis(self):
|
|
assert "不做医疗诊断" in coach.SYSTEM
|
|
|
|
def test_model_is_told_not_to_recompute_the_z_scores(self):
|
|
assert "不要自行重算" in coach.SYSTEM
|
|
|
|
def test_briefing_prompt_carries_the_context_as_json(self):
|
|
messages = coach.briefing_messages({"snapshotDate": "2026-08-01", "x": 1})
|
|
assert messages[0]["role"] == "system"
|
|
assert '"snapshotDate":"2026-08-01"' in messages[1]["content"]
|
|
|
|
def test_context_is_not_ascii_escaped(self):
|
|
"""Escaping Chinese to \\uXXXX roughly triples its token cost."""
|
|
assert "睡眠" in coach._payload({"label": "睡眠"})
|
|
|
|
def test_copilot_keeps_the_context_out_of_the_visible_transcript(self):
|
|
messages = coach.copilot_messages(
|
|
{"snapshotDate": "2026-08-01"}, [], "我今天能练吗？"
|
|
)
|
|
assert [m["role"] for m in messages] == ["system", "system", "user"]
|
|
assert messages[-1]["content"] == "我今天能练吗？"
|
|
|
|
def test_copilot_history_is_capped_and_role_filtered(self):
|
|
history = [{"role": "user", "content": f"q{i}"} for i in range(20)]
|
|
history.append({"role": "tool", "content": "ignored"})
|
|
messages = coach.copilot_messages({}, history, "最后一问")
|
|
turns = [m for m in messages if m["role"] != "system"]
|
|
assert len(turns) == 9 # eight remembered turns plus the new question
|
|
assert "ignored" not in json.dumps(messages, ensure_ascii=False)
|
|
|
|
def test_copilot_asks_for_markdown_not_json(self):
|
|
assert "Markdown" in coach.COPILOT_SYSTEM
|
|
assert "最后出现的 JSON" not in coach.COPILOT_SYSTEM
|
|
|
|
|
|
# --- rule-based counterparts ------------------------------------------------
|
|
def context_with(**today):
|
|
base = {
|
|
"snapshotDate": "2026-08-30",
|
|
"todayMetrics": {
|
|
"sleep": None,
|
|
"autonomicNervous": {},
|
|
"recovery": {},
|
|
"activityToday": {},
|
|
},
|
|
"deviations": [],
|
|
"trends": [],
|
|
"activityShift": {},
|
|
}
|
|
base["todayMetrics"].update(today)
|
|
return base
|
|
|
|
|
|
class TestRuleBriefing:
|
|
def test_readiness_drives_the_prescription(self):
|
|
low = coach.rule_briefing(context_with(recovery={"trainingReadiness": 30}))
|
|
high = coach.rule_briefing(context_with(recovery={"trainingReadiness": 85}))
|
|
assert low["prescription"]["intensity"] == "低"
|
|
assert high["prescription"]["intensity"] == "高"
|
|
|
|
def test_short_sleep_is_named_as_the_shortfall(self):
|
|
out = coach.rule_briefing(context_with(
|
|
sleep={"durationHours": 5.0, "targetHours": 7.0}
|
|
))
|
|
assert "睡眠" in out["shortfall"]
|
|
|
|
def test_no_shortfall_is_stated_explicitly(self):
|
|
out = coach.rule_briefing(context_with(
|
|
sleep={"durationHours": 8.0, "targetHours": 7.0}
|
|
))
|
|
assert out["shortfall"] == "无明显短板"
|
|
|
|
def test_it_never_invents_a_metric_the_watch_did_not_record(self):
|
|
out = coach.rule_briefing(context_with())
|
|
assert out["diagnosis"] == []
|
|
assert out["actions"]
|
|
|
|
def test_the_headline_names_the_largest_departure(self):
|
|
context = context_with(autonomicNervous={"restingHr": 80})
|
|
context["deviations"] = [{
|
|
"metric": "heartRate", "label": "静息心率", "unit": "bpm",
|
|
"value": 80, "baselineMean": 60, "sd": 5, "baselineDays": 28,
|
|
"z": 4.0, "verdict": "偏差",
|
|
}]
|
|
assert "静息心率" in coach.rule_briefing(context)["headline"]
|
|
|
|
def test_a_sustained_drop_in_steps_becomes_an_action(self):
|
|
context = context_with()
|
|
context["activityShift"] = {
|
|
"steps": {"label": "步数", "recentMean": 4000,
|
|
"priorMean": 10000, "changePct": -60.0}
|
|
}
|
|
assert any("60" in a for a in coach.rule_briefing(context)["actions"])
|
|
|
|
|
|
# --- orchestration ----------------------------------------------------------
|
|
@pytest.fixture
|
|
def gateway(monkeypatch):
|
|
monkeypatch.setenv("AI_GATEWAY_TOKEN", "test-token")
|
|
monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gateway.test/v1")
|
|
monkeypatch.setenv("AI_MODEL_CHAIN", "gateway")
|
|
|
|
|
|
@pytest.fixture
|
|
def month(db, user, seed_health):
|
|
seed_health([
|
|
{"date": f"2026-08-{d:02d}", "steps": 8000, "heart_rate": 60, "hrv": 45,
|
|
"sleep_duration": 7, "sleep_quality": 80, "stress": 30}
|
|
for d in range(1, 31)
|
|
])
|
|
return user
|
|
|
|
|
|
def answer(monkeypatch, text):
|
|
"""Make every model reply with `text`, and count the calls."""
|
|
calls = []
|
|
|
|
def fake_chat(self, messages, timeout=None, max_tokens=None):
|
|
calls.append(messages)
|
|
return ai_svc.Completion(text, "nvidia")
|
|
|
|
monkeypatch.setattr(ai_svc.Provider, "chat", fake_chat, raising=False)
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", fake_chat)
|
|
return calls
|
|
|
|
|
|
class TestGetBriefing:
|
|
def test_no_data_says_so_rather_than_guessing(self, db, user, gateway):
|
|
out = analysis_svc.get_briefing(user["id"])
|
|
assert out["meta"]["source"] == "none"
|
|
assert out["briefing"] is None
|
|
|
|
def test_blocking_mode_returns_the_model_answer(self, month, gateway, monkeypatch):
|
|
answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
out = analysis_svc.get_briefing(month["id"], wait=True)
|
|
assert out["meta"]["source"] == "ai"
|
|
assert out["briefing"]["status"] == "中等偏上"
|
|
|
|
def test_a_stored_answer_is_reused(self, month, gateway, monkeypatch):
|
|
calls = answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
analysis_svc.get_briefing(month["id"], wait=True)
|
|
out = analysis_svc.get_briefing(month["id"])
|
|
assert out["meta"]["cached"] is True
|
|
assert len(calls) == 1, "the cached answer must not trigger a second call"
|
|
|
|
def test_new_health_data_expires_the_stored_answer(
|
|
self, month, gateway, monkeypatch, seed_health
|
|
):
|
|
answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
analysis_svc.get_briefing(month["id"], wait=True)
|
|
seed_health([{"date": "2026-08-31", "steps": 12000}])
|
|
out = analysis_svc.get_briefing(month["id"])
|
|
assert out["meta"].get("cached") is not True
|
|
|
|
def test_a_failing_model_degrades_to_the_rule_engine(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
def boom(self, messages, timeout=None, max_tokens=None):
|
|
raise ai_svc.AIError("upstream down")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom)
|
|
out = analysis_svc.get_briefing(month["id"], wait=True)
|
|
assert out["meta"]["source"] == "rules"
|
|
assert out["briefing"] is not None
|
|
|
|
def test_the_non_blocking_path_answers_without_calling_a_model(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
"""It enqueues instead. No worker runs in the suite, so a model call
|
|
here would mean the request generated inline."""
|
|
calls = answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
out = analysis_svc.get_briefing(month["id"])
|
|
assert out["meta"]["pending"] is True
|
|
assert out["briefing"]["status"]
|
|
assert calls == []
|
|
assert jobs.pending_count(month["id"]) == 1
|
|
|
|
def test_polling_does_not_queue_a_job_per_poll(self, month, gateway):
|
|
"""The screen polls every few seconds; a generation takes minutes."""
|
|
for _ in range(5):
|
|
analysis_svc.get_briefing(month["id"])
|
|
assert jobs.pending_count(month["id"]) == 1
|
|
|
|
|
|
class TestGetTrendInsight:
|
|
def test_empty_span_is_reported_not_analysed(self, month, gateway):
|
|
out = analysis_svc.get_trend_insight(
|
|
month["id"], "steps", "2020-01-01", "2020-01-31"
|
|
)
|
|
assert out["meta"]["source"] == "none"
|
|
|
|
def test_model_answer_is_returned_and_cached(self, month, gateway, monkeypatch):
|
|
reply = json.dumps({
|
|
"summary": "步数稳定。", "drivers": [], "caution": None,
|
|
"confidence": "medium",
|
|
}, ensure_ascii=False)
|
|
calls = answer(monkeypatch, reply)
|
|
first = analysis_svc.get_trend_insight(
|
|
month["id"], "steps", "2026-08-01", "2026-08-30"
|
|
)
|
|
second = analysis_svc.get_trend_insight(
|
|
month["id"], "steps", "2026-08-01", "2026-08-30"
|
|
)
|
|
assert first["insight"]["summary"] == "步数稳定。"
|
|
assert second["meta"]["cached"] is True
|
|
assert len(calls) == 1
|
|
|
|
def test_a_failing_model_degrades_to_the_rule_engine(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
def boom(self, messages, timeout=None, max_tokens=None):
|
|
raise ai_svc.AIError("down")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom)
|
|
out = analysis_svc.get_trend_insight(
|
|
month["id"], "steps", "2026-08-01", "2026-08-30"
|
|
)
|
|
assert out["meta"]["source"] == "rules"
|
|
assert out["insight"]["summary"]
|
|
|
|
|
|
class TestStreamChat:
|
|
"""The gateway's streaming path is measurably less reliable than its
|
|
blocking one, so a stream that produces nothing must not end the attempt."""
|
|
|
|
def test_deltas_are_forwarded(self, gateway, monkeypatch):
|
|
def fake_stream(self, messages, timeout=None, max_tokens=None):
|
|
yield ai_svc.Completion("你好", "nvidia")
|
|
yield ai_svc.Completion("，世界", "nvidia")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
|
|
text = "".join(d.text for d in ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
|
|
assert text == "你好，世界"
|
|
|
|
def test_a_failed_stream_retries_the_same_model_without_streaming(
|
|
self, gateway, monkeypatch
|
|
):
|
|
def fake_stream(self, messages, timeout=None, max_tokens=None):
|
|
raise ai_svc.AIError("所有模型均不可用")
|
|
yield # pragma: no cover - generator marker
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
|
|
answer(monkeypatch, "完整回答")
|
|
deltas = list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
|
|
assert "".join(d.text for d in deltas) == "完整回答"
|
|
|
|
def test_no_model_switch_once_text_has_been_sent(self, gateway, monkeypatch):
|
|
def fake_stream(self, messages, timeout=None, max_tokens=None):
|
|
yield ai_svc.Completion("半句", "nvidia")
|
|
raise ai_svc.AIError("断流")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
|
|
with pytest.raises(ai_svc.AIError):
|
|
list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
|
|
|
|
|
|
class TestCopilotStream:
|
|
def test_no_data_yields_an_error_event(self, db, user, gateway):
|
|
events = list(analysis_svc.copilot_stream(user["id"], "我今天能练吗"))
|
|
assert events[0][0] == "error"
|
|
|
|
def test_a_successful_answer_is_framed_start_delta_done(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
def fake_stream(self, messages, timeout=None, max_tokens=None):
|
|
yield ai_svc.Completion("可以。", "nvidia")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
|
|
events = list(analysis_svc.copilot_stream(month["id"], "我今天能练吗"))
|
|
assert [e for e, _ in events] == ["start", "delta", "done"]
|
|
assert events[-1][1]["upstream"] == "nvidia"
|
|
|
|
|
|
# --- endpoints --------------------------------------------------------------
|
|
class TestEndpoints:
|
|
def test_briefing_requires_auth(self, client):
|
|
assert client.get("/api/analysis/briefing").status_code == 401
|
|
|
|
def test_trend_insight_requires_auth(self, client):
|
|
assert client.get("/api/analysis/trend-insight").status_code == 401
|
|
|
|
def test_copilot_requires_auth(self, client):
|
|
assert client.post("/api/analysis/copilot", json={}).status_code == 401
|
|
|
|
def test_briefing_answers_even_with_no_model_configured(self, client, auth, month):
|
|
resp = client.get("/api/analysis/briefing", headers=auth)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["briefing"] is not None
|
|
|
|
def test_trend_insight_rejects_an_unknown_metric(self, client, auth, month):
|
|
resp = client.get(
|
|
"/api/analysis/trend-insight",
|
|
query_string={"metric": "nope", "startDate": "2026-08-01",
|
|
"endDate": "2026-08-30"},
|
|
headers=auth,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "supported" in resp.get_json()
|
|
|
|
def test_trend_insight_requires_a_range(self, client, auth, month):
|
|
resp = client.get(
|
|
"/api/analysis/trend-insight",
|
|
query_string={"metric": "steps"}, headers=auth,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_copilot_requires_a_question(self, client, auth, month):
|
|
resp = client.post("/api/analysis/copilot", json={}, headers=auth)
|
|
assert resp.status_code == 400
|
|
|
|
def test_copilot_streams_server_sent_events(
|
|
self, client, auth, month, gateway, monkeypatch
|
|
):
|
|
def fake_stream(self, messages, timeout=None, max_tokens=None):
|
|
yield ai_svc.Completion("可以，注意强度。", "nvidia")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
|
|
resp = client.post(
|
|
"/api/analysis/copilot",
|
|
json={"question": "我今天能练吗"}, headers=auth,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.mimetype == "text/event-stream"
|
|
body = resp.get_data(as_text=True)
|
|
assert "event: delta" in body
|
|
assert "可以，注意强度。" in body
|
|
|
|
def test_briefing_never_leaks_the_gateway_token(
|
|
self, client, auth, month, gateway
|
|
):
|
|
body = client.get("/api/analysis/briefing", headers=auth).get_data(as_text=True)
|
|
assert "test-token" not in body
|
|
|
|
|
|
class TestStreamRetryScope:
|
|
"""The blind non-streaming retry is only worth doing for endpoints that
|
|
actually have a separate streaming transport."""
|
|
|
|
def test_a_provider_without_streaming_is_not_called_twice(
|
|
self, monkeypatch
|
|
):
|
|
monkeypatch.setenv("GEMINI_API_KEY", "k")
|
|
monkeypatch.setenv("AI_MODEL_CHAIN", "gemini-flash")
|
|
calls = []
|
|
|
|
def boom(self, messages, timeout=None, max_tokens=None):
|
|
calls.append(1)
|
|
raise ai_svc.AIError("down")
|
|
|
|
monkeypatch.setattr(ai_svc.GeminiProvider, "chat", boom)
|
|
with pytest.raises(ai_svc.AIError):
|
|
list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
|
|
assert len(calls) == 1
|
|
|
|
def test_the_gateway_declares_a_streaming_transport(self):
|
|
assert ai_svc.CATALOG["gateway"].streaming is True
|
|
assert ai_svc.CATALOG["gemini-flash"].streaming is False
|
|
|
|
|
|
class TestRegenerate:
|
|
def test_refresh_evicts_the_stored_answer_so_the_poll_can_see_the_new_one(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
"""Without eviction the poll after 重新生成 reads the row it was asked
|
|
to replace, reports `cached`, and stops — leaving the old text on
|
|
screen."""
|
|
answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
|
|
analysis_svc.get_briefing(month["id"], wait=True)
|
|
assert analysis_svc.get_briefing(month["id"])["meta"]["cached"] is True
|
|
|
|
analysis_svc.get_briefing(month["id"], refresh=True)
|
|
|
|
after = analysis_svc.get_briefing(month["id"])
|
|
assert after["meta"].get("cached") is not True
|
|
assert after["meta"]["pending"] is True
|
|
|
|
|
|
# --- the producer/consumer queue --------------------------------------------
|
|
class TestJobQueue:
|
|
def test_enqueue_then_claim(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "2026-08-30:14")
|
|
claimed = jobs._claim_next()
|
|
assert claimed["kind"] == "sleep"
|
|
assert claimed["status"] == "pending", "the row read is the pre-claim one"
|
|
assert jobs.status_of(user["id"], "sleep", "2026-08-30:14")["status"] == "running"
|
|
|
|
def test_the_same_work_queued_twice_is_one_row(self, db, user):
|
|
for _ in range(6):
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
assert jobs.pending_count(user["id"]) == 1
|
|
|
|
def test_opening_a_screen_promotes_it_ahead_of_the_backfill(self, db, user):
|
|
for scope in ("health", "exercise", "trends"):
|
|
jobs.enqueue(user["id"], scope, "s", priority=jobs.PRIORITY_PREFETCH)
|
|
jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_PREFETCH)
|
|
# The user opens 睡眠 while the backfill is still queued.
|
|
jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_INTERACTIVE)
|
|
assert jobs._claim_next()["kind"] == "sleep"
|
|
|
|
def test_priority_is_never_demoted(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_INTERACTIVE)
|
|
jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_PREFETCH)
|
|
assert jobs.status_of(user["id"], "sleep", "s")["priority"] == \
|
|
jobs.PRIORITY_INTERACTIVE
|
|
|
|
def test_equal_priority_runs_oldest_first(self, db, user):
|
|
jobs.enqueue(user["id"], "health", "s")
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
assert jobs._claim_next()["kind"] == "health"
|
|
|
|
def test_a_finished_job_is_not_run_again(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc")
|
|
jobs._finish(jobs.job_id(user["id"], "sleep", "s"))
|
|
assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "done"
|
|
assert jobs._claim_next() is None
|
|
|
|
def test_changed_data_requeues_a_finished_job(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc")
|
|
jobs._finish(jobs.job_id(user["id"], "sleep", "s"))
|
|
assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="xyz") == "pending"
|
|
|
|
def test_a_running_job_is_not_restarted_by_a_poll(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
jobs._claim_next()
|
|
assert jobs.enqueue(user["id"], "sleep", "s") == "running"
|
|
|
|
def test_a_claim_left_by_a_dead_worker_is_retried(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
jobs._claim_next()
|
|
assert jobs._claim_next() is None, "still within the claim window"
|
|
# Backdate the claim past its timeout, as a killed worker would leave it.
|
|
stale = (jobs._now() - jobs.datetime.timedelta(
|
|
seconds=jobs.CLAIM_TIMEOUT_SECONDS + 60))
|
|
db.execute("UPDATE ai_jobs SET claimed_at = ?", [jobs._iso(stale)])
|
|
assert jobs._claim_next() is not None
|
|
|
|
def test_restart_releases_whatever_was_running(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
jobs._claim_next()
|
|
jobs.reset_stale_claims()
|
|
assert jobs.status_of(user["id"], "sleep", "s")["status"] == "pending"
|
|
|
|
def test_a_job_that_keeps_failing_is_eventually_dropped(self, db, user):
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
for _ in range(jobs.MAX_ATTEMPTS):
|
|
row = jobs._claim_next()
|
|
assert row is not None
|
|
jobs._finish(row["id"], "boom")
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
assert jobs._claim_next() is None
|
|
|
|
def test_the_runner_failing_does_not_stop_the_queue(self, db, user, monkeypatch):
|
|
monkeypatch.setattr(jobs, "_runner", lambda *a: (_ for _ in ()).throw(RuntimeError("x")))
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
assert jobs.run_once() is True
|
|
assert jobs.status_of(user["id"], "sleep", "s")["status"] == "failed"
|
|
|
|
def test_run_once_dispatches_to_the_registered_runner(self, db, user, monkeypatch):
|
|
seen = []
|
|
monkeypatch.setattr(jobs, "_runner", lambda u, k, s: seen.append((u, k, s)))
|
|
jobs.enqueue(user["id"], "sleep", "2026-08-30:14")
|
|
assert jobs.run_once() is True
|
|
assert seen == [(user["id"], "sleep", "2026-08-30:14")]
|
|
assert jobs.status_of(user["id"], "sleep", "2026-08-30:14")["status"] == "done"
|
|
|
|
def test_run_once_is_a_no_op_on_an_empty_queue(self, db, user, monkeypatch):
|
|
monkeypatch.setattr(jobs, "_runner", lambda *a: None)
|
|
assert jobs.run_once() is False
|
|
|
|
def test_jobs_are_per_account(self, db, user, make_user):
|
|
other = make_user("other@example.com")
|
|
jobs.enqueue(user["id"], "sleep", "s")
|
|
assert jobs.pending_count(other["id"]) == 0
|
|
|
|
|
|
# --- per-screen scopes ------------------------------------------------------
|
|
class TestScopes:
|
|
def test_every_registered_scope_builds_or_declines(self, month):
|
|
"""No builder may raise on a real account — an empty screen returns
|
|
None so the caller can stay quiet."""
|
|
for name in scopes.SCOPES:
|
|
out = scopes.build(month["id"], name, subject="1" if name == "activity" else None)
|
|
assert out is None or (isinstance(out, tuple) and len(out) == 2)
|
|
|
|
def test_a_scope_with_data_carries_highlights_and_a_subject(self, month):
|
|
subject, context = scopes.build(month["id"], "sleep")
|
|
assert subject
|
|
assert context["scope"] == "sleep"
|
|
assert context["highlights"]
|
|
|
|
def test_contexts_stay_small_enough_to_prompt_with(self, month):
|
|
for name in ("health", "sleep", "exercise", "trends", "daily"):
|
|
built = scopes.build(month["id"], name)
|
|
if not built:
|
|
continue
|
|
blob = json.dumps(built[1], ensure_ascii=False)
|
|
assert len(blob) < 30_000, f"{name} context is {len(blob)} chars"
|
|
|
|
def test_an_empty_screen_declines_rather_than_inventing_one(self, db, user):
|
|
assert scopes.build(user["id"], "body") is None
|
|
assert scopes.build(user["id"], "race") is None
|
|
|
|
def test_per_item_scopes_need_a_subject(self, month):
|
|
assert scopes.build(month["id"], "activity") is None
|
|
|
|
def test_unknown_scope_raises(self, month):
|
|
with pytest.raises(KeyError):
|
|
scopes.build(month["id"], "nope")
|
|
|
|
def test_prefetch_list_excludes_per_item_screens(self):
|
|
"""Pre-warming `activity` would queue one job per session, not one."""
|
|
assert "activity" not in scopes.PREFETCH_SCOPES
|
|
assert "daily" not in scopes.PREFETCH_SCOPES
|
|
assert set(scopes.PREFETCH_SCOPES) <= set(scopes.SCOPES)
|
|
|
|
def test_health_scope_carries_the_bands_the_ui_shows(self, month):
|
|
_, context = scopes.build(month["id"], "health")
|
|
assert context["referenceBands"], "the model must not contradict the card"
|
|
|
|
|
|
class TestScopeInsight:
|
|
def test_unknown_scope_raises(self, month):
|
|
with pytest.raises(KeyError):
|
|
analysis_svc.get_scope_insight(month["id"], "nope")
|
|
|
|
def test_an_empty_screen_reports_it(self, db, user, gateway):
|
|
out = analysis_svc.get_scope_insight(user["id"], "race")
|
|
assert out["meta"]["source"] == "none"
|
|
|
|
def test_first_visit_answers_from_rules_and_queues_the_model(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
calls = answer(monkeypatch, "{}")
|
|
out = analysis_svc.get_scope_insight(month["id"], "sleep")
|
|
assert out["meta"]["pending"] is True
|
|
# The computed facts, with the first one leading; a screen with a
|
|
# single fact has a headline and no remaining points.
|
|
assert out["insight"]["headline"]
|
|
assert calls == [], "nothing may be generated inside the request"
|
|
assert jobs.pending_count(month["id"]) == 1
|
|
|
|
def test_the_stored_answer_is_served_once_generated(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
reply = json.dumps({
|
|
"headline": "睡眠偏短。", "points": [{"title": "时长", "detail": "6.5 小时。"}],
|
|
"actions": ["提前入睡"], "caution": None, "confidence": "medium",
|
|
}, ensure_ascii=False)
|
|
answer(monkeypatch, reply)
|
|
analysis_svc.generate_scope_insight(month["id"], "sleep")
|
|
out = analysis_svc.get_scope_insight(month["id"], "sleep")
|
|
assert out["meta"]["source"] == "ai"
|
|
assert out["meta"]["cached"] is True
|
|
assert out["insight"]["headline"] == "睡眠偏短。"
|
|
|
|
def test_rules_fallback_says_it_has_not_been_interpreted(self, month, gateway):
|
|
out = analysis_svc.get_scope_insight(month["id"], "sleep")
|
|
assert "尚未经过模型解读" in out["insight"]["caution"]
|
|
|
|
def test_prefetch_queues_every_screen_that_has_data(self, month, gateway):
|
|
queued = analysis_svc.prefetch_insights(month["id"])
|
|
assert "briefing" in queued
|
|
assert "sleep" in queued
|
|
assert jobs.pending_count(month["id"]) == len(queued)
|
|
|
|
def test_the_queue_runner_reaches_a_scope(self, month, gateway, monkeypatch):
|
|
reply = json.dumps({"headline": "ok", "points": [], "actions": [],
|
|
"confidence": "low"}, ensure_ascii=False)
|
|
answer(monkeypatch, reply)
|
|
subject, _ = scopes.build(month["id"], "sleep")
|
|
analysis_svc._run_job(month["id"], "sleep", subject)
|
|
assert analysis_svc.get_scope_insight(month["id"], "sleep")["meta"]["source"] == "ai"
|
|
|
|
|
|
class TestParseScopeInsight:
|
|
def test_valid_reply(self):
|
|
out = coach.parse_scope_insight(json.dumps({
|
|
"headline": "睡眠不足。",
|
|
"points": [{"title": "时长", "detail": "平均 6 小时。"}],
|
|
"actions": ["提前入睡"], "caution": "样本偏少", "confidence": "high",
|
|
}, ensure_ascii=False))
|
|
assert out["confidence"] == "high"
|
|
assert out["actions"] == ["提前入睡"]
|
|
|
|
def test_points_written_as_plain_strings(self):
|
|
out = coach.parse_scope_insight(json.dumps(
|
|
{"headline": "h", "points": ["纯文本要点"]}, ensure_ascii=False))
|
|
assert out["points"][0]["detail"] == "纯文本要点"
|
|
|
|
def test_blank_reply_is_rejected(self):
|
|
with pytest.raises(ai_svc.AIError):
|
|
coach.parse_scope_insight(json.dumps({"confidence": "high"}))
|
|
|
|
def test_reasoning_preamble_is_skipped(self):
|
|
reply = ("Let me think. Maybe {\"headline\": \"draft\"} ... final:\n"
|
|
+ json.dumps({"headline": "定稿", "points": []}, ensure_ascii=False))
|
|
assert coach.parse_scope_insight(reply)["headline"] == "定稿"
|
|
|
|
|
|
class TestScopeEndpoint:
|
|
def test_requires_auth(self, client):
|
|
assert client.get("/api/analysis/insight?scope=sleep").status_code == 401
|
|
|
|
def test_unknown_scope_lists_the_supported_ones(self, client, auth, month):
|
|
resp = client.get("/api/analysis/insight?scope=nope", headers=auth)
|
|
assert resp.status_code == 400
|
|
assert "sleep" in resp.get_json()["supported"]
|
|
|
|
def test_answers_immediately(self, client, auth, month):
|
|
resp = client.get("/api/analysis/insight?scope=sleep", headers=auth)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["insight"]["headline"]
|
|
|
|
def test_queue_endpoint_reports_outstanding_work(self, client, auth, month):
|
|
client.get("/api/analysis/insight?scope=sleep", headers=auth)
|
|
assert client.get("/api/analysis/insight/queue", headers=auth).get_json()["pending"] >= 1
|
|
|
|
|
|
class TestOutageRecovery:
|
|
"""A gateway outage must delay an insight, not retire it."""
|
|
|
|
def test_a_given_up_job_is_retried_after_the_cooldown(self, db, user):
|
|
jid = jobs.job_id(user["id"], "sleep", "s")
|
|
jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc")
|
|
for _ in range(jobs.MAX_ATTEMPTS):
|
|
row = jobs._claim_next()
|
|
jobs._finish(row["id"], "connection refused")
|
|
jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc")
|
|
assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "failed"
|
|
|
|
# Past the cooldown, as it would be once the upstream is back.
|
|
old = jobs._now() - jobs.datetime.timedelta(
|
|
seconds=jobs.FAILED_RETRY_SECONDS + 60)
|
|
db.execute("UPDATE ai_jobs SET updated_at = ? WHERE id = ?",
|
|
[jobs._iso(old), jid])
|
|
assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "pending"
|
|
assert jobs.status_of(user["id"], "sleep", "s")["attempts"] == 0
|
|
assert jobs._claim_next() is not None
|
|
|
|
def test_the_screen_stops_polling_once_the_queue_gives_up(
|
|
self, month, gateway, monkeypatch
|
|
):
|
|
"""`pending` must go false, or the card spins for minutes waiting on an
|
|
answer that is not coming."""
|
|
def boom(self, messages, timeout=None, max_tokens=None):
|
|
raise ai_svc.AIError("upstream unreachable")
|
|
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom)
|
|
monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", boom)
|
|
|
|
subject, _ = scopes.build(month["id"], "sleep")
|
|
for _ in range(jobs.MAX_ATTEMPTS):
|
|
analysis_svc.get_scope_insight(month["id"], "sleep")
|
|
jobs.run_once()
|
|
|
|
out = analysis_svc.get_scope_insight(month["id"], "sleep")
|
|
assert out["meta"]["queue"] == "failed"
|
|
assert out["meta"]["pending"] is False
|
|
assert out["meta"]["reason"], "the card should be able to say why"
|
|
assert out["insight"]["headline"], "and still show the computed facts"
|
|
|
|
|
|
class TestSubjectStability:
|
|
"""The subject keys the cache *and* the job queue, so it must identify what
|
|
the insight is about — never how much data happened to be there.
|
|
|
|
A count in the key made every poll mint a fresh job: production
|
|
accumulated a dozen `trends` jobs in minutes, all for the same screen."""
|
|
|
|
def test_the_subject_does_not_move_when_the_data_grows(
|
|
self, db, user, seed_health
|
|
):
|
|
seed_health([
|
|
{"date": f"2026-08-{d:02d}", "steps": 8000, "heart_rate": 60,
|
|
"hrv": 45, "sleep_duration": 7, "sleep_quality": 80, "stress": 30}
|
|
for d in range(1, 29)
|
|
])
|
|
before = {
|
|
name: scopes.build(user["id"], name)[0]
|
|
for name in ("health", "sleep", "exercise", "trends")
|
|
if scopes.build(user["id"], name)
|
|
}
|
|
|
|
# A backfilled day, as a history sync would add: older than the newest,
|
|
# so what the screen is about has not changed.
|
|
seed_health([{"date": "2026-07-31", "steps": 7000, "heart_rate": 61,
|
|
"hrv": 44, "sleep_duration": 7, "sleep_quality": 79,
|
|
"stress": 31}])
|
|
|
|
after = {name: scopes.build(user["id"], name)[0] for name in before}
|
|
assert after == before
|
|
|
|
def test_no_subject_encodes_a_row_count(self, month):
|
|
for name in ("health", "sleep", "exercise", "trends", "challenges"):
|
|
built = scopes.build(month["id"], name)
|
|
if not built:
|
|
continue
|
|
subject = built[0]
|
|
# A count would grow without bound; a date or a window constant
|
|
# will not. This catches the shape of the mistake, not one instance.
|
|
for part in str(subject).split(":"):
|
|
assert not (part.isdigit() and int(part) > 400), \
|
|
f"{name} subject {subject!r} looks like a row count"
|
|
|
|
|
|
class TestSupersede:
|
|
def test_opening_a_screen_clears_queued_work_for_an_older_snapshot(
|
|
self, month, gateway
|
|
):
|
|
jobs.enqueue(month["id"], "trends", "2026-08-01")
|
|
jobs.enqueue(month["id"], "trends", "2026-08-15")
|
|
analysis_svc.get_scope_insight(month["id"], "trends")
|
|
rows = analysis_svc.query_all(
|
|
"SELECT subject FROM ai_jobs WHERE kind = 'trends'")
|
|
assert len(rows) == 1, "only the current snapshot should be queued"
|
|
|
|
def test_per_item_screens_keep_one_job_each(self, month, gateway):
|
|
"""每日 and 运动详情 legitimately have one entry per date / session."""
|
|
jobs.enqueue(month["id"], "daily", "2026-08-01")
|
|
jobs.enqueue(month["id"], "daily", "2026-08-02")
|
|
analysis_svc.get_scope_insight(month["id"], "daily", subject="2026-08-03")
|
|
rows = analysis_svc.query_all(
|
|
"SELECT subject FROM ai_jobs WHERE kind = 'daily'")
|
|
assert len(rows) == 3
|
|
|
|
def test_running_work_is_not_dropped_from_under_the_worker(self, month, gateway):
|
|
jobs.enqueue(month["id"], "trends", "2026-08-01")
|
|
jobs._claim_next()
|
|
analysis_svc.get_scope_insight(month["id"], "trends")
|
|
statuses = {r["subject"]: r["status"] for r in analysis_svc.query_all(
|
|
"SELECT subject, status FROM ai_jobs WHERE kind = 'trends'")}
|
|
assert statuses.get("2026-08-01") == "running"
|