357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""深度诊断（tier=full）链路端到端：派生 → full_report 日配额 → full 流水线 → 拒绝路径。
|
||
|
||
与 test_report_flow 同一套基建：Fake 阶段经 ``REPORT_PIPELINE_STAGES_JSON`` /
|
||
``REPORT_FULL_PIPELINE_STAGES_JSON`` 动态装配（worker 按 report.tier 分流两
|
||
条阶段清单，两份 env 要同时给）；不启真 celery，worker 内芯直调。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import uuid
|
||
from collections.abc import AsyncIterator, Iterator
|
||
from contextlib import contextmanager
|
||
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",
|
||
),
|
||
]
|
||
|
||
_MODULE = "tests.integration.test_full_report_flow"
|
||
_BASIC_STAGES = json.dumps([f"{_MODULE}:FakeVisionStage", f"{_MODULE}:FakeDiagnosisStage"])
|
||
_FULL_STAGES = json.dumps([f"{_MODULE}:FakeFullDiagnosisStage", f"{_MODULE}:FakeFullCopyStage"])
|
||
|
||
|
||
class FakeVisionStage:
|
||
"""与 test_report_flow 同一契约：derived_features 非空即跳过。"""
|
||
|
||
name = "vision"
|
||
|
||
async def run(self, ctx: Any) -> None:
|
||
features = ctx.report_input.derived_features
|
||
if not features:
|
||
features["vision_runs"] = 1
|
||
ctx.report.result["vision"] = {"features": sorted(features)}
|
||
|
||
|
||
class FakeDiagnosisStage:
|
||
name = "diagnosis"
|
||
|
||
async def run(self, ctx: Any) -> None:
|
||
assert "vision" in ctx.report.result
|
||
ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}}
|
||
|
||
|
||
class FakeFullDiagnosisStage:
|
||
"""full 流水线的诊断阶段替身：前面没有 vision 阶段（full 不含视觉）。"""
|
||
|
||
name = "diagnosis"
|
||
|
||
async def run(self, ctx: Any) -> None:
|
||
assert "vision" not in ctx.report.result
|
||
ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}}
|
||
|
||
|
||
class FakeFullCopyStage:
|
||
"""full 文案阶段替身：copy 与 full 双写（结构与真实 FullCopyStage 一致）。"""
|
||
|
||
name = "copy"
|
||
|
||
async def run(self, ctx: Any) -> None:
|
||
assert "diagnosis" in ctx.report.result
|
||
ctx.report.result["copy"] = {
|
||
"summary": "深度总评",
|
||
"sections": [{"title": "风格", "paragraphs": ["段落"]}],
|
||
"model": "fake",
|
||
"prompt_version": "v1",
|
||
"dropped_keys": [],
|
||
}
|
||
ctx.report.result["full"] = {
|
||
"season": {
|
||
"code": "summer",
|
||
"detailed": "柔夏",
|
||
"name_cn": "柔夏型",
|
||
"analysis": "解析文案",
|
||
},
|
||
"palette": {
|
||
"best": [{"name": "雾霾蓝", "hex": "#8FA8BF"}],
|
||
"neutral": [{"name": "浅灰", "hex": "#D9D9D9"}],
|
||
"avoid": [{"name": "明黄", "hex": "#FFD93B"}],
|
||
},
|
||
"scenes": [
|
||
{
|
||
"key": "commute",
|
||
"title": "日常通勤",
|
||
"outfits": [{"name": "通勤 A", "points": ["要点一"]}],
|
||
}
|
||
],
|
||
"plan30": [{"week": 1, "title": "第 1 周", "tasks": ["任务一"]}],
|
||
}
|
||
|
||
|
||
# --- fixtures ---------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def no_enqueue(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr("apps.api.routers.reports._enqueue_report", lambda task_id: None)
|
||
|
||
|
||
@contextmanager
|
||
def _stage_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
||
from packages.shared.config import get_settings
|
||
|
||
monkeypatch.setenv("REPORT_PIPELINE_STAGES_JSON", _BASIC_STAGES)
|
||
monkeypatch.setenv("REPORT_FULL_PIPELINE_STAGES_JSON", _FULL_STAGES)
|
||
get_settings.cache_clear()
|
||
try:
|
||
yield
|
||
finally:
|
||
monkeypatch.undo()
|
||
get_settings.cache_clear()
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_stages(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
||
with _stage_config(monkeypatch):
|
||
yield
|
||
|
||
|
||
@pytest.fixture
|
||
async def client(no_enqueue: None) -> 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
|
||
|
||
|
||
# --- helpers ----------------------------------------------------------------
|
||
|
||
|
||
async def _login(client: Any) -> tuple[dict[str, str], int]:
|
||
resp = await client.post(
|
||
"/api/v1/auth/dev-token", json={"device_key": f"full-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 _upload_photo(client: Any, headers: dict[str, str]) -> str:
|
||
resp = await client.post(
|
||
"/api/v1/uploads/direct?scene=report_photo",
|
||
headers={**headers, "Content-Type": "image/jpeg"},
|
||
content=b"\xff\xd8\xff\xe0" + b"\x00" * 64,
|
||
)
|
||
assert resp.status_code == 200
|
||
return str(resp.json()["data"]["object_key"])
|
||
|
||
|
||
async def _create_succeeded_basic(client: Any, headers: dict[str, str]) -> dict[str, Any]:
|
||
from apps.worker.report_tasks import _generate_report
|
||
|
||
resp = await client.post(
|
||
"/api/v1/reports",
|
||
headers=headers,
|
||
json={
|
||
"questionnaire": {"style": "通勤", "height_cm": 168},
|
||
"photo_object_key": await _upload_photo(client, headers),
|
||
"request_key": f"rk-{uuid.uuid4().hex}",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()["data"]
|
||
await _generate_report(int(body["task_id"]))
|
||
return body
|
||
|
||
|
||
async def _derive_full(
|
||
client: Any, headers: dict[str, str], report_id: str, *, request_key: str | None = None
|
||
) -> Any:
|
||
return await client.post(
|
||
f"/api/v1/reports/{report_id}/full",
|
||
headers=headers,
|
||
json={"request_key": request_key or f"rk-full-{uuid.uuid4().hex}"},
|
||
)
|
||
|
||
|
||
async def _fetch_report(report_id: int) -> Any:
|
||
from sqlalchemy import select
|
||
|
||
from packages.shared.db import session_scope
|
||
from packages.shared.models import RpReport
|
||
|
||
async with session_scope() as session:
|
||
result = await session.execute(select(RpReport).where(RpReport.report_id == report_id))
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
async def _fetch_input(report_id: int) -> Any:
|
||
from sqlalchemy import select
|
||
|
||
from packages.shared.db import session_scope
|
||
from packages.shared.models import RpInput
|
||
|
||
async with session_scope() as session:
|
||
result = await session.execute(select(RpInput).where(RpInput.report_id == report_id))
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
async def _ledger_rows(user_id: int, quota_key: str) -> list[Any]:
|
||
from sqlalchemy import select
|
||
|
||
from packages.shared.db import session_scope
|
||
from packages.shared.models import Ledger
|
||
|
||
async with session_scope() as session:
|
||
result = await session.execute(
|
||
select(Ledger)
|
||
.where(Ledger.user_id == user_id, Ledger.quota_key == quota_key)
|
||
.order_by(Ledger.created_at.asc(), Ledger.tx_id.asc())
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
# --- 场景 1：派生成功 + full 流水线 + 幂等重放 ----------------------------------
|
||
|
||
|
||
async def test_derive_full_success_runs_full_pipeline(fake_stages: None, client: Any) -> None:
|
||
from apps.worker.report_tasks import _generate_report
|
||
|
||
headers, user_id = await _login(client)
|
||
basic = await _create_succeeded_basic(client, headers)
|
||
|
||
request_key = f"rk-full-{uuid.uuid4().hex}"
|
||
resp = await _derive_full(client, headers, basic["report_id"], request_key=request_key)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
full_report_id = int(data["report_id"])
|
||
assert full_report_id != int(basic["report_id"])
|
||
assert data["version"] == 1 # full 是独立谱系，版本从 1 起算
|
||
|
||
# full_report 日配额预扣（period_key d:YYYY-MM-DD）。
|
||
rows = await _ledger_rows(user_id, "full_report")
|
||
assert len(rows) == 1
|
||
assert rows[0].status == "reserved"
|
||
assert rows[0].period_key.startswith("d:")
|
||
|
||
# 输入三件套复用（derived_features 带着 basic 的视觉中间态，无视觉阶段）。
|
||
full_input = await _fetch_input(full_report_id)
|
||
assert full_input.derived_features == {"vision_runs": 1}
|
||
|
||
await _generate_report(int(data["task_id"]))
|
||
report = await _fetch_report(full_report_id)
|
||
assert report.status == "succeeded"
|
||
assert report.tier == "full"
|
||
# full 流水线：无 vision 键，copy + full + diagnosis 齐全。
|
||
assert set(report.result) == {"diagnosis", "copy", "full"}
|
||
full = report.result["full"]
|
||
assert full["season"]["name_cn"] == "柔夏型"
|
||
assert full["palette"]["best"][0]["hex"] == "#8FA8BF"
|
||
assert full["scenes"][0]["outfits"][0]["points"]
|
||
assert full["plan30"][0]["week"] == 1
|
||
|
||
rows = await _ledger_rows(user_id, "full_report")
|
||
assert [row.status for row in rows] == ["committed"]
|
||
|
||
# 详情视图透出 tier 与 full 结构（App 凭 tier/full 渲染深度区块）。
|
||
detail = (await client.get(f"/api/v1/reports/{full_report_id}", headers=headers)).json()["data"]
|
||
assert detail["tier"] == "full"
|
||
assert detail["result"]["full"]["season"]["detailed"] == "柔夏"
|
||
|
||
# 幂等重放：同 request_key 返回同任务同报告，不二次预扣。
|
||
replay = await _derive_full(client, headers, basic["report_id"], request_key=request_key)
|
||
assert replay.status_code == 200
|
||
assert replay.json()["data"]["report_id"] == str(full_report_id)
|
||
assert len(await _ledger_rows(user_id, "full_report")) == 1
|
||
|
||
|
||
# --- 场景 2：非 SUCCEEDED 源报告拒绝 --------------------------------------------
|
||
|
||
|
||
async def test_derive_full_rejects_pending_source(fake_stages: None, client: Any) -> None:
|
||
headers, _user_id = await _login(client)
|
||
resp = await client.post(
|
||
"/api/v1/reports",
|
||
headers=headers,
|
||
json={
|
||
"questionnaire": {"style": "通勤"},
|
||
"photo_object_key": await _upload_photo(client, headers),
|
||
"request_key": f"rk-{uuid.uuid4().hex}",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
basic = resp.json()["data"] # pending，不跑 worker
|
||
|
||
derived = await _derive_full(client, headers, basic["report_id"])
|
||
assert derived.status_code == 409
|
||
assert derived.json()["code"] == 40900 # STATE_CONFLICT
|
||
|
||
|
||
# --- 场景 3：full 不能再派生 full（防套娃） --------------------------------------
|
||
|
||
|
||
async def test_derive_full_rejects_full_source(fake_stages: None, client: Any) -> None:
|
||
from apps.worker.report_tasks import _generate_report
|
||
|
||
headers, _user_id = await _login(client)
|
||
basic = await _create_succeeded_basic(client, headers)
|
||
derived = await _derive_full(client, headers, basic["report_id"])
|
||
assert derived.status_code == 200
|
||
full = derived.json()["data"]
|
||
await _generate_report(int(full["task_id"]))
|
||
|
||
nested = await _derive_full(client, headers, full["report_id"])
|
||
assert nested.status_code == 409
|
||
assert nested.json()["code"] == 40900
|
||
|
||
|
||
# --- 场景 4：full_report 配额耗尽 ------------------------------------------------
|
||
|
||
|
||
async def test_derive_full_blocked_when_quota_exhausted(fake_stages: None, client: Any) -> None:
|
||
"""开发库 full_report 规则行是测试期放宽值（base_limit=10）；把本用例用户的
|
||
bl_quota 快照 limit_value 收敛到已用量，确定性断言 40301（同 test_report_flow
|
||
场景 4 的做法，只动本用户的快照行，不动全局规则）。"""
|
||
from sqlalchemy import update
|
||
|
||
from packages.shared.db import session_scope
|
||
from packages.shared.models import Quota
|
||
|
||
headers, user_id = await _login(client)
|
||
basic = await _create_succeeded_basic(client, headers)
|
||
|
||
first = await _derive_full(client, headers, basic["report_id"])
|
||
assert first.status_code == 200 # 首次派生占掉 1 次
|
||
|
||
async with session_scope() as session:
|
||
await session.execute(
|
||
update(Quota)
|
||
.where(Quota.user_id == user_id, Quota.quota_key == "full_report")
|
||
.values(limit_value=1)
|
||
)
|
||
|
||
second = await _derive_full(client, headers, basic["report_id"])
|
||
assert second.status_code == 403
|
||
assert second.json()["code"] == 40301 # QUOTA_EXHAUSTED
|