"""结论反馈 + 预约留资的端到端断言（T-P3-21 / T-P3-22，迭代 3）。 覆盖验收点：反馈 upsert 幂等（同维度覆盖更新不堆行）、归属校验他人报告 40420、dimension_key 校验（不存在/skipped → 40001）、预约重复 created=false。 报告行直接 INSERT（不走流水线）——反馈/预约断言的是路由与服务口径， 不是生成链路（生成链路由 test_report_flow 覆盖）。 """ from __future__ import annotations import os import uuid from collections.abc import AsyncIterator from typing import Any import pytest def _infra_available() -> bool: return bool(os.getenv("MEIDA_RUN_INTEGRATION")) pytestmark = [ pytest.mark.integration, pytest.mark.skipif( not _infra_available(), reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", ), ] _OK_DIM = "body_shape" _SKIPPED_DIM = "color_season" @pytest.fixture async def client() -> AsyncIterator[Any]: import httpx from apps.api.main import create_app from packages.shared.config import get_settings app = create_app(get_settings()) transport = httpx.ASGITransport(app=app) async with ( httpx.AsyncClient(transport=transport, base_url="http://test") as http, app.router.lifespan_context(app), ): yield http async def _login(client: Any) -> tuple[dict[str, str], int]: resp = await client.post( "/api/v1/auth/dev-token", json={"device_key": f"fb-it-{uuid.uuid4().hex[:8]}"} ) assert resp.json()["code"] == 0 data = resp.json()["data"] return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"]) async def _insert_report(user_id: int) -> int: """直接落一份 succeeded 报告：两个维度，一个 ok 一个 skipped。""" from packages.domain.ids import next_id from packages.shared.db import session_scope from packages.shared.models import RpReport report_id = next_id() async with session_scope() as session: session.add( RpReport( report_id=report_id, user_id=user_id, tier="basic", status="succeeded", version=1, result={ "diagnosis": { "dimensions": [ {"number": 5, "key": _OK_DIM, "status": "ok", "value": "X"}, {"number": 19, "key": _SKIPPED_DIM, "status": "skipped", "value": None}, ], "coverage": {"basic_full": False, "total": 2}, } }, ) ) return report_id async def _count_feedback(report_id: int) -> int: from sqlalchemy import func, select from packages.shared.db import session_scope from packages.shared.models import RpFeedback async with session_scope() as session: result = await session.execute( select(func.count()).select_from(RpFeedback).where(RpFeedback.report_id == report_id) ) return int(result.scalar_one()) # --- 反馈：upsert 幂等 -------------------------------------------------------- async def test_feedback_upsert_overwrites_same_dimension(client: Any) -> None: headers, user_id = await _login(client) report_id = await _insert_report(user_id) first = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": _OK_DIM, "verdict": "accurate"}, ) assert first.status_code == 200 assert first.json()["data"]["updated"] is False # 改判：同维度再反馈 = 覆盖更新（updated=true），不是堆行 second = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": _OK_DIM, "verdict": "inaccurate", "correction": "其实我是梨形"}, ) assert second.status_code == 200 assert second.json()["data"]["updated"] is True assert await _count_feedback(report_id) == 1 listing = (await client.get(f"/api/v1/reports/{report_id}/feedback", headers=headers)).json()[ "data" ] assert len(listing) == 1 assert listing[0]["dimension_key"] == _OK_DIM assert listing[0]["verdict"] == "inaccurate" assert listing[0]["correction"] == "其实我是梨形" # --- 反馈：归属与维度校验 ------------------------------------------------------ async def test_feedback_on_others_report_returns_40420(client: Any) -> None: headers, _ = await _login(client) _, other_user_id = await _login(client) report_id = await _insert_report(other_user_id) resp = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": _OK_DIM, "verdict": "accurate"}, ) assert resp.json()["code"] == 40420 get_resp = await client.get(f"/api/v1/reports/{report_id}/feedback", headers=headers) assert get_resp.json()["code"] == 40420 async def test_feedback_dimension_must_exist_and_be_ok(client: Any) -> None: headers, user_id = await _login(client) report_id = await _insert_report(user_id) # 不存在的维度 resp = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": "no_such_dim", "verdict": "accurate"}, ) assert resp.json()["code"] == 40001 # skipped 维度没有结论，不可反馈 resp2 = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": _SKIPPED_DIM, "verdict": "accurate"}, ) assert resp2.json()["code"] == 40001 # verdict 枚举外 → 40001（pydantic 422 → 统一信封 VALIDATION_FAILED） resp3 = await client.post( f"/api/v1/reports/{report_id}/feedback", headers=headers, json={"dimension_key": _OK_DIM, "verdict": "kind-of"}, ) assert resp3.json()["code"] == 40001 assert await _count_feedback(report_id) == 0 # --- 预约：幂等 --------------------------------------------------------------- async def test_reservation_duplicate_returns_created_false(client: Any) -> None: headers, _ = await _login(client) first = await client.post( "/api/v1/reservations", headers=headers, json={"topic": "full_report"} ) assert first.status_code == 200 assert first.json()["data"]["created"] is True again = await client.post( "/api/v1/reservations", headers=headers, json={"topic": "full_report"} ) assert again.status_code == 200 assert again.json()["data"]["created"] is False listing = (await client.get("/api/v1/reservations?topic=full_report", headers=headers)).json()[ "data" ] assert listing["topics"] == ["full_report"] # topic 参数缺省返回全部 all_topics = (await client.get("/api/v1/reservations", headers=headers)).json()["data"] assert "full_report" in all_topics["topics"]