from __future__ import annotations import asyncio import io import shutil from collections.abc import AsyncIterator from pathlib import Path from uuid import uuid4 import pytest from fastapi.testclient import TestClient from PIL import Image from sqlalchemy import delete, text from sqlalchemy.ext.asyncio import AsyncSession from app.api.dependencies.auth import build_request_context from app.api.dependencies.database import get_session from app.core.contracts import RequestContext from app.infrastructure.db import SessionFactory from app.main import app from app.model.audit import InteractionAudit from app.model.promotion_material import ( PromotionAttachment, PromotionComplianceCheck, PromotionDeliveryRecord, PromotionInputSnapshot, PromotionMaterialTask, PromotionMaterialVersion, PromotionReviewRecord, ) from app.service.promotion_material_service import PromotionMaterialService CURRENT_CONTEXT = RequestContext( user_id="1", trace_id="promotion-integration", roles=("operator",), permissions=( "promotion:read", "promotion:write", "promotion:review", "promotion:deliver", ), data_scope="all", ) async def override_context() -> RequestContext: return CURRENT_CONTEXT async def override_session() -> AsyncIterator[AsyncSession]: async with SessionFactory() as session: yield session @pytest.mark.integration def test_promotion_material_http_workflow_and_advisor_scope(tmp_path: Path) -> None: global CURRENT_CONTEXT trace_id = f"trace-promotion-{uuid4()}" CURRENT_CONTEXT = RequestContext( user_id="1", trace_id=trace_id, roles=("operator",), permissions=( "promotion:read", "promotion:write", "promotion:review", "promotion:deliver", ), data_scope="all", ) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session task_no = "" idempotency_keys: list[str] = [] try: with TestClient(app) as client: create_key = _key("create") idempotency_keys.append(create_key) created = client.post( "/api/v1/fund-promotion-materials", headers={"Idempotency-Key": create_key}, json={ "product_name": "南方稳健配置测试基金", "product_code": "PROMO-001", "material_title": "南方稳健配置测试基金产品推介材料", "style_code": "balanced_allocation", "output_formats": ["pptx", "poster"], }, ) assert created.status_code == 200 assert created.json()["code"] == 0 task_no = created.json()["data"]["task_no"] replayed = client.post( "/api/v1/fund-promotion-materials", headers={"Idempotency-Key": create_key}, json={ "product_name": "南方稳健配置测试基金", "product_code": "PROMO-001", "material_title": "南方稳健配置测试基金产品推介材料", "style_code": "balanced_allocation", "output_formats": ["pptx", "poster"], }, ) assert replayed.status_code == 200 assert replayed.json()["data"] == created.json()["data"] input_key = _key("inputs") idempotency_keys.append(input_key) updated = client.put( f"/api/v1/fund-promotion-materials/{task_no}/inputs", headers={"Idempotency-Key": input_key}, json=_inputs_payload(), ) assert updated.status_code == 200 assert updated.json()["data"]["status"] == "input_ready" photo_key = _key("photo") idempotency_keys.append(photo_key) photo = client.post( f"/api/v1/fund-promotion-materials/{task_no}/attachments", params={"attachment_type": "manager_photo"}, headers={"Idempotency-Key": photo_key}, files={ "file": ( "manager-photo.jpg", _manager_photo_bytes(), # 模拟部分 Windows 浏览器/代理上传图片时给出的通用媒体类型。 "application/octet-stream", ) }, ) assert photo.status_code == 200 assert photo.json()["data"]["attachment_type"] == "manager_photo" performance_key = _key("performance") idempotency_keys.append(performance_key) performance = client.post( f"/api/v1/fund-promotion-materials/{task_no}/attachments", params={"attachment_type": "performance_data"}, headers={"Idempotency-Key": performance_key}, files={ "file": ( "performance.csv", _performance_csv_bytes(), "text/csv", ) }, ) assert performance.status_code == 200 assert performance.json()["data"]["attachment_type"] == "performance_data" assert performance.json()["data"]["performance_summary"] == { "initial_date": "2025-01-31", "as_of_date": "2025-08-31", "history_months": 7, "product_return": "7.1%", "max_drawdown": "0%", } generation_key = _key("generation") idempotency_keys.append(generation_key) generated = client.post( f"/api/v1/fund-promotion-materials/{task_no}/generations", headers={"Idempotency-Key": generation_key}, json={"output_formats": ["pptx", "poster"]}, ) assert generated.status_code == 200 generated_data = generated.json()["data"] assert generated_data["status"] == "pending_review" version_id = int(generated_data["material_version_id"]) assert Path(generated_data["pptx_path"]).is_file() assert Path(generated_data["poster_path"]).is_file() assert len(generated_data["chart_paths"]) == 1 assert Path(generated_data["chart_paths"][0]).is_file() checks = client.get( f"/api/v1/fund-promotion-materials/{task_no}/compliance-checks" ) assert checks.status_code == 200 assert any( item["rule_code"] == "overall.pass" for item in checks.json()["data"]["findings"] ) before_review = client.get( f"/api/v1/fund-promotion-materials/{task_no}" ) assert before_review.status_code == 200 assert before_review.json()["data"]["material_version"] is None review_key = _key("review") idempotency_keys.append(review_key) reviewed = client.post( f"/api/v1/fund-promotion-materials/{task_no}/reviews", headers={"Idempotency-Key": review_key}, json={ "material_version_id": version_id, "decision": "approved", "comment": "集成测试审核通过", }, ) assert reviewed.status_code == 200 assert reviewed.json()["data"]["status"] == "approved" after_review = client.get( f"/api/v1/fund-promotion-materials/{task_no}" ) assert after_review.status_code == 200 assert after_review.json()["data"]["material_version"]["id"] == version_id assert after_review.json()["data"]["material_version"]["status"] == "approved" delivery_key = _key("delivery") idempotency_keys.append(delivery_key) delivered = client.post( f"/api/v1/fund-promotion-materials/{task_no}/deliveries", headers={"Idempotency-Key": delivery_key}, json={ "material_version_id": version_id, "advisor_ids": [42], "delivery_channel": "internal_record", }, ) assert delivered.status_code == 200 assert delivered.json()["data"]["status"] == "sent" CURRENT_CONTEXT = RequestContext( user_id="42", trace_id=trace_id, roles=("advisor",), permissions=("promotion:read",), data_scope="self", ) advisor_view = client.get( f"/api/v1/fund-promotion-materials/{task_no}" ) assert advisor_view.status_code == 200 assert advisor_view.json()["data"]["material_version"]["status"] == "sent" CURRENT_CONTEXT = RequestContext( user_id="43", trace_id=trace_id, roles=("advisor",), permissions=("promotion:read",), data_scope="self", ) other_advisor_view = client.get( f"/api/v1/fund-promotion-materials/{task_no}" ) assert other_advisor_view.status_code == 404 # `docs/05-接口文档.md` §3.6 未定义通用 404 码，非会话资源按同一"资源隐藏" # 语义复用 SESSION_NOT_FOUND（见 app/core/errors.py 的说明）。 assert other_advisor_view.json()["error"]["code"] == "SESSION_NOT_FOUND" finally: asyncio.run(_cleanup_promotion_task(task_no, trace_id, idempotency_keys)) storage_root = Path(PromotionMaterialService().storage_root) if task_no: shutil.rmtree(storage_root / task_no, ignore_errors=True) app.dependency_overrides.clear() CURRENT_CONTEXT = RequestContext( user_id="1", trace_id="promotion-integration", roles=("operator",), permissions=( "promotion:read", "promotion:write", "promotion:review", "promotion:deliver", ), data_scope="all", ) @pytest.mark.integration def test_promotion_material_http_rejects_missing_key_and_permission() -> None: global CURRENT_CONTEXT app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session try: with TestClient(app) as client: CURRENT_CONTEXT = RequestContext( user_id="1", trace_id=f"trace-promotion-validation-{uuid4()}", roles=("operator",), permissions=("promotion:write",), data_scope="all", ) missing_key = client.post( "/api/v1/fund-promotion-materials", json={ "product_name": "缺少幂等键测试", "material_title": "缺少幂等键测试材料", }, ) # 缺少必填幂等键属于请求校验失败：文档 §3.5/§3.6 规定 422 + AGENT_INPUT_INVALID， # 且必须走统一错误信封。该 Header 在 Controller 上声明为可选、由 Service 校验， # 因此走业务异常信封（field_errors 为空，原因在 message 里）； # FastAPI 原生校验错误的信封由 tests/unit/api/test_request_validation_envelope.py 覆盖。 assert missing_key.status_code == 422 assert missing_key.json()["error"]["code"] == "AGENT_INPUT_INVALID" CURRENT_CONTEXT = RequestContext( user_id="1", trace_id=f"trace-promotion-forbidden-{uuid4()}", roles=("operator",), permissions=(), data_scope="all", ) forbidden = client.post( "/api/v1/fund-promotion-materials", headers={"Idempotency-Key": _key("forbidden")}, json={ "product_name": "无权限测试", "material_title": "无权限测试材料", }, ) assert forbidden.status_code == 403 # 文档 §3.6 的权威码是 AGENT_PERMISSION_DENIED（旧用例沿用未定义的 FORBIDDEN）。 assert forbidden.json()["error"]["code"] == "AGENT_PERMISSION_DENIED" finally: app.dependency_overrides.clear() CURRENT_CONTEXT = RequestContext( user_id="1", trace_id="promotion-integration", roles=("operator",), permissions=( "promotion:read", "promotion:write", "promotion:review", "promotion:deliver", ), data_scope="all", ) def _inputs_payload() -> dict[str, object]: return { "product_info": { "fund_type": "混合型", "operation_mode": "开放式", "product_status": "new_product", "investment_objective": "通过股债灵活配置实现长期资产配置目标", "benchmark": "中证全债指数收益率×60%+沪深300指数收益率×40%", "risk_level": "R3", }, "manager_info": { "manager_name": "张三", "management_company": "南方基金管理有限公司", "registration_code": "P10000001", "employment_years": "10年", "investment_management_experience": "8年公募基金投资管理经验", "profile": "负责资产配置和组合管理。", }, "team_info": { "team_description": "具备完整的投资、研究和风险管理分工。", "research_capability": "覆盖宏观、行业和信用研究。", }, "strategy_info": { "investment_scope": "股票、债券、货币市场工具及法律法规允许的其他资产。", "strategy": "通过大类资产配置和基本面研究动态调整组合。", "restrictions": "遵守法律法规、基金合同及监管限制。", "index_tool_attribute": "用于长期资产配置和组合风险管理。", }, "fee_structure": { "subscription_fee": "1.0%", "purchase_fee": "1.0%", "redemption_fee": "0.5%", "sales_service_fee": "0.2%", "management_fee": "1.0%", "custody_fee": "0.2%", "client_maintenance_fee": "不适用", }, "performance_info": { "as_of_date": "2026-08-31", "history_months": 12, "product_return": "8.6%", "max_drawdown": "-5.2%", "volatility": "10.1%", "sharpe_ratio": "0.82", "show_product_performance": True, "show_manager_performance": True, "ranking": { "enabled": False, "institution_name": None, "evaluation_period_years": None, "ranking_text": None, "public_source": None, }, }, "risk_disclosure": { "special_risks": ["市场风险", "流动性风险"], "additional_notes": "本材料仅供专业投顾内部研究使用。", }, "source_notes": {"source": "promotion-integration-test"}, } def _manager_photo_bytes() -> bytes: image = Image.new("RGB", (640, 800), (24, 65, 110)) output = io.BytesIO() image.save(output, format="JPEG", quality=90) return output.getvalue() def _performance_csv_bytes() -> bytes: return ( "日期,产品收益率,业绩比较基准收益率,基金经理代表产品收益率\n" "2025-01-31,1.0%,0.6%,0.8%\n" "2025-02-28,2.5%,1.4%,2.0%\n" "2025-03-31,3.2%,2.0%,2.7%\n" "2025-04-30,4.1%,2.8%,3.5%\n" "2025-05-31,5.0%,3.6%,4.2%\n" "2025-06-30,5.8%,4.2%,5.1%\n" "2025-07-31,6.4%,4.9%,5.9%\n" "2025-08-31,7.1%,5.5%,6.6%\n" ).encode("utf-8-sig") def _key(prefix: str) -> str: return f"promotion-it-{prefix}-{uuid4().hex}" async def _cleanup_promotion_task( task_no: str, trace_id: str, idempotency_keys: list[str] ) -> None: async with SessionFactory() as session, session.begin(): if trace_id: await session.execute( delete(InteractionAudit).where( InteractionAudit.detail["trace_id"].as_string() == trace_id ) ) if task_no: await session.execute( delete(PromotionComplianceCheck).where( PromotionComplianceCheck.task_no == task_no ) ) await session.execute( delete(PromotionReviewRecord).where( PromotionReviewRecord.task_no == task_no ) ) await session.execute( delete(PromotionDeliveryRecord).where( PromotionDeliveryRecord.task_no == task_no ) ) await session.execute( delete(PromotionMaterialVersion).where( PromotionMaterialVersion.task_no == task_no ) ) await session.execute( delete(PromotionAttachment).where(PromotionAttachment.task_no == task_no) ) await session.execute( delete(PromotionInputSnapshot).where( PromotionInputSnapshot.task_no == task_no ) ) await session.execute( delete(PromotionMaterialTask).where(PromotionMaterialTask.task_no == task_no) ) if idempotency_keys: await session.execute( text( "DELETE FROM api_request_receipt " "WHERE idempotency_key IN :keys" ).bindparams(keys=tuple(idempotency_keys)) )