"""report.profile.latest_profile_summary:repo 打桩,验证摘要选取纪律。""" from __future__ import annotations import json from types import SimpleNamespace from typing import Any import pytest from packages.report import repo from packages.report.profile import latest_profile_summary _DIAGNOSIS = { "dimensions": [ { "key": "body_shape", "label": "体型", "status": "ok", "value": "梨形", "confidence": 0.9, "evidence": ["问卷自报肩窄胯宽"], }, {"key": "style_impression", "status": "skipped", "value": None}, ], "rule_hits": [ {"rule_key": "pear_lower_dark", "dimension": "body_shape", "suggestion": "下装选深色"} ], } def _report(report_id: int, *, status: str, result: dict[str, Any]) -> Any: return SimpleNamespace(report_id=report_id, status=status, result=result) @pytest.fixture def stub_list(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: state: dict[str, Any] = {"reports": []} async def fake_list(session: Any, *, user_id: int, limit: int = 20) -> list[Any]: state["limit"] = limit return state["reports"] monkeypatch.setattr(repo, "list_reports", fake_list) return state class TestLatestProfileSummary: async def test_no_reports_returns_none(self, stub_list: dict[str, Any]) -> None: assert await latest_profile_summary(None, user_id=1, max_chars=800) is None # type: ignore[arg-type] async def test_succeeded_report_with_diagnosis(self, stub_list: dict[str, Any]) -> None: stub_list["reports"] = [_report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS})] summary = await latest_profile_summary(None, user_id=1, max_chars=800) # type: ignore[arg-type] assert summary is not None payload = json.loads(summary) values = {d["key"]: d["value"] for d in payload["dimensions"]} assert values["body_shape"] == "梨形" # skipped 维度不进素材(防幻觉第一道闸) assert "style_impression" not in values assert payload["rule_hits"][0]["rule_key"] == "pear_lower_dark" async def test_latest_failed_falls_back_to_earlier_succeeded( self, stub_list: dict[str, Any] ) -> None: stub_list["reports"] = [ _report(2, status="failed", result={}), _report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS}), ] summary = await latest_profile_summary(None, user_id=1, max_chars=800) # type: ignore[arg-type] assert summary is not None and "梨形" in summary async def test_succeeded_without_diagnosis_is_skipped(self, stub_list: dict[str, Any]) -> None: stub_list["reports"] = [_report(1, status="succeeded", result={"vision": {}})] assert await latest_profile_summary(None, user_id=1, max_chars=800) is None # type: ignore[arg-type] async def test_budget_is_forwarded(self, stub_list: dict[str, Any]) -> None: """max_chars 收紧时低优先级维度被裁、硬结论保留(复用 copy 的裁剪纪律)。""" stub_list["reports"] = [_report(1, status="succeeded", result={"diagnosis": _DIAGNOSIS})] summary = await latest_profile_summary(None, user_id=1, max_chars=60) # type: ignore[arg-type] assert summary is not None assert len(summary) > 60 # 硬结论永不裁,载荷可超预算(copy.py 既定语义) assert "梨形" in summary