05f55e695b
队列本来是完全不可见的：页面上一句「排队生成中」说不出自己是下一个、第二十 个，还是已经放弃了——网关挂掉的时候，「还在生成」和「永远不会好」长得一模 一样。今天排查就是这么排的。 - GET /analysis/insight/queue 返回队列（running 在前，其次按优先级和年龄， 和 worker 实际取任务的顺序一致）、已生成的解读、scope 名到中文标签的映射 （前端不必再抄一份），以及消费者的限流配置 - POST /analysis/insight/queue/retry：手动把「已放弃」的重新排队，不等冷却。 自动重试要等冷却是为了不去捶一个正在抽风的上游；人按下重试是他自己判断值得 再试一次 - 页面在 设置 → AI 生成队列。插队的任务标「插队」——这是整个界面最想让人看见 的一件事：为什么是它排在最前面 - 「已生成」单独列：队列空了意味着「没有待办」，不是「什么都没生成过」， 没有这一节这两件事在界面上没法区分 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1169 lines
51 KiB
Python
1169 lines
51 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"
|
||
|
||
|
||
class TestHighlightsReadAlone:
|
||
"""The first highlight becomes the collapsed card's whole text, with no
|
||
title beside it — so it has to name what it is about."""
|
||
|
||
def test_every_first_highlight_names_its_subject(self, month):
|
||
for name in scopes.SCOPES:
|
||
built = scopes.build(month["id"], name)
|
||
if not built:
|
||
continue
|
||
first = built[1]["highlights"][0]
|
||
title = first["title"]
|
||
# Titles that are already summaries rather than names of a metric.
|
||
if title in ("整体", "当前预测") or title.startswith("近 "):
|
||
continue
|
||
assert title in first["detail"], (
|
||
f"{name}: 折叠态只显示 detail，但它没提到 {title!r}: "
|
||
f"{first['detail']!r}"
|
||
)
|
||
|
||
|
||
class TestGatewayCourtesy:
|
||
"""The gateway runs one worker with four threads and is shared with two
|
||
other projects. This consumer must not be able to saturate it."""
|
||
|
||
def test_only_one_job_runs_at_a_time_across_the_deployment(self, db, user):
|
||
jobs.enqueue(user["id"], "health", "a")
|
||
jobs.enqueue(user["id"], "sleep", "b")
|
||
assert jobs._claim_next() is not None
|
||
assert jobs._claim_next() is None, \
|
||
"a second Gunicorn worker must not start a second gateway call"
|
||
|
||
def test_a_finished_job_frees_the_slot(self, db, user):
|
||
jobs.enqueue(user["id"], "health", "a")
|
||
jobs.enqueue(user["id"], "sleep", "b")
|
||
first = jobs._claim_next()
|
||
jobs._finish(first["id"])
|
||
assert jobs._claim_next() is not None
|
||
|
||
def test_an_abandoned_claim_does_not_block_the_queue_forever(self, db, user):
|
||
jobs.enqueue(user["id"], "health", "a")
|
||
jobs.enqueue(user["id"], "sleep", "b")
|
||
jobs._claim_next()
|
||
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_there_is_a_gap_between_jobs(self):
|
||
"""Back-to-back is what saturates a four-thread box."""
|
||
assert jobs.GAP_SECONDS > 0
|
||
|
||
|
||
class TestQueueScreen:
|
||
def test_queue_endpoint_requires_auth(self, client):
|
||
assert client.get("/api/analysis/insight/queue").status_code == 401
|
||
assert client.post("/api/analysis/insight/queue/retry").status_code == 401
|
||
|
||
def test_lists_jobs_labels_and_limits(self, client, auth, month):
|
||
client.get("/api/analysis/insight?scope=sleep", headers=auth)
|
||
body = client.get("/api/analysis/insight/queue", headers=auth).get_json()
|
||
assert body["pending"] >= 1
|
||
assert any(j["kind"] == "sleep" for j in body["jobs"])
|
||
assert body["scopes"]["sleep"] == "睡眠", "the UI must not restate the list"
|
||
assert body["settings"]["concurrency"] == jobs.MAX_CONCURRENT
|
||
|
||
def test_an_open_screen_is_marked_as_having_jumped_the_queue(
|
||
self, client, auth, month
|
||
):
|
||
jobs.enqueue(month["id"], "trends", "2026-08-30", priority=jobs.PRIORITY_PREFETCH)
|
||
client.get("/api/analysis/insight?scope=sleep", headers=auth)
|
||
by_kind = {j["kind"]: j for j in
|
||
client.get("/api/analysis/insight/queue", headers=auth).get_json()["jobs"]}
|
||
assert by_kind["sleep"]["interactive"] is True
|
||
assert by_kind["trends"]["interactive"] is False
|
||
|
||
def test_running_jobs_are_listed_before_waiting_ones(self, client, auth, month):
|
||
jobs.enqueue(month["id"], "health", "a")
|
||
jobs.enqueue(month["id"], "sleep", "b")
|
||
jobs._claim_next()
|
||
listed = client.get("/api/analysis/insight/queue", headers=auth).get_json()["jobs"]
|
||
assert listed[0]["status"] == "running"
|
||
|
||
def test_generated_answers_are_listed_separately(
|
||
self, client, auth, month, gateway, monkeypatch
|
||
):
|
||
"""An empty queue means "nothing left to do", not "nothing was done" —
|
||
the two look the same without this."""
|
||
answer(monkeypatch, json.dumps({
|
||
"headline": "h", "points": [], "actions": [], "confidence": "low",
|
||
}, ensure_ascii=False))
|
||
analysis_svc.generate_scope_insight(month["id"], "sleep")
|
||
body = client.get("/api/analysis/insight/queue", headers=auth).get_json()
|
||
assert any(i["kind"] == "sleep" for i in body["insights"])
|
||
|
||
def test_retry_puts_given_up_jobs_back(self, client, auth, month):
|
||
jobs.enqueue(month["id"], "sleep", "s")
|
||
row = jobs._claim_next()
|
||
for _ in range(jobs.MAX_ATTEMPTS):
|
||
jobs._finish(row["id"], "boom")
|
||
assert jobs.status_of(month["id"], "sleep", "s")["status"] == "failed"
|
||
|
||
resp = client.post("/api/analysis/insight/queue/retry", headers=auth)
|
||
assert resp.status_code == 200
|
||
assert jobs.status_of(month["id"], "sleep", "s")["status"] == "pending"
|
||
assert jobs.status_of(month["id"], "sleep", "s")["attempts"] == 0
|
||
|
||
def test_the_queue_is_per_account(self, client, auth, month, make_user):
|
||
other = make_user("other@example.com")
|
||
jobs.enqueue(other["id"], "sleep", "s")
|
||
body = client.get("/api/analysis/insight/queue", headers=auth).get_json()
|
||
assert body["jobs"] == []
|