from __future__ import annotations import asyncio from collections.abc import AsyncIterator from pathlib import Path from uuid import uuid4 import pytest from fastapi.testclient import TestClient from sqlalchemy import delete, select 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.core.offsite_fund_contracts import ReceiveRecognizedMailRequest from app.infrastructure.db import SessionFactory from app.main import app from app.model.audit import InteractionAudit from app.model.offsite_fund import ( OffsiteExecutionPlanTask, OffsiteFundAttachment, OffsiteFundDocument, OffsiteFundMail, OffsiteNotification, OffsiteQueryRecord, OffsiteRuleResult, ) from app.service.offsite_fund_service import OffsiteFundService from app.service.offsite_smtp_adapter import SmtpSendResult TRACE_ID = "" async def override_context() -> RequestContext: return RequestContext( user_id="1", trace_id=TRACE_ID or str(uuid4()), roles=("operator",), permissions=("offsite:write",), data_scope="all", ) async def override_session() -> AsyncIterator[AsyncSession]: async with SessionFactory() as session: yield session async def _ingest_recognized_mail_async(payload: dict[str, object]) -> dict[str, object]: """直接调用同一业务入口写入已识别邮件：HTTP 写入接口已下线。""" context_factory = app.dependency_overrides.get(build_request_context, override_context) async with SessionFactory() as session: return await OffsiteFundService(session).receive_recognized_mail( ReceiveRecognizedMailRequest(**payload), await context_factory() ) def _ingest_recognized_mail( client: TestClient, payload: dict[str, object] ) -> dict[str, object]: """通过 TestClient 的事件循环调用，保证连接池与后续 HTTP 请求在同一个循环里。""" assert client.portal is not None return client.portal.call(_ingest_recognized_mail_async, payload) @pytest.mark.integration def test_successful_notification_send_is_idempotent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TRACE_ID TRACE_ID = f"trace-send-success-{uuid4()}" payload = _subscription_payload() attachment_path = tmp_path / "申购申请单.pdf" attachment_path.write_bytes(b"original-pdf") payload["attachments"][0]["original_file_path"] = str(attachment_path) sent_requests: list[object] = [] class FakeSuccessSender: def __init__(self, _settings: object) -> None: pass def send_reply(self, request: object) -> SmtpSendResult: sent_requests.append(request) return SmtpSendResult( status="发送成功", dry_run=False, provider_message_id="provider-test-001", failure_reason=None, retry_count=0, request_summary={"provider": "test"}, ) monkeypatch.setattr("app.service.offsite_fund_service.OffsiteSmtpSender", FakeSuccessSender) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" notification_id = 0 try: with TestClient(app) as client: data = _ingest_recognized_mail(client, payload)["data"] mail_id = data["mail_id"] task_id = data["documents"][0]["task_id"] _confirm_and_create_notice(client, task_id) notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={"notification_type": "normal_return", "operator_id": "operator-001"}, ) notification_id = int(notice.json()["data"]["notification_id"]) first = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={"operator_id": "operator-001", "operator_confirmed": True}, ) second = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={"operator_id": "operator-001", "operator_confirmed": True}, ) recalculate = client.post( "/api/v1/offsite-fund/settlement-statistics/recalculate", json={"fund_code": "000001", "application_date": "2026-09-10"}, ) assert first.json()["data"]["status"] == "发送成功" assert first.json()["data"]["provider_message_id"] == "provider-test-001" assert second.json()["data"]["status"] == "发送成功" assert len(sent_requests) == 1 assert recalculate.json()["data"]["subscription_amount_yuan"] == "10000.0000" asyncio.run(_assert_notification(notification_id, "发送成功", "provider-test-001")) asyncio.run(_assert_mail_status(mail_id, "normal_return_sent")) finally: asyncio.run(_cleanup(mail_id, task_id, notification_id, TRACE_ID)) app.dependency_overrides.clear() TRACE_ID = "" @pytest.mark.integration def test_failed_notification_send_persists_failure_and_retry_count( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TRACE_ID TRACE_ID = f"trace-send-failure-{uuid4()}" payload = _subscription_payload() attachment_path = tmp_path / "申购申请单.pdf" attachment_path.write_bytes(b"original-pdf") payload["attachments"][0]["original_file_path"] = str(attachment_path) class FakeFailureSender: def __init__(self, _settings: object) -> None: pass def send_reply(self, request: object) -> SmtpSendResult: del request return SmtpSendResult( status="发送失败", dry_run=False, provider_message_id=None, failure_reason="SMTPException", retry_count=1, request_summary={"provider": "test"}, ) monkeypatch.setattr("app.service.offsite_fund_service.OffsiteSmtpSender", FakeFailureSender) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" notification_id = 0 try: with TestClient(app) as client: data = _ingest_recognized_mail(client, payload)["data"] mail_id = data["mail_id"] task_id = data["documents"][0]["task_id"] _confirm_and_create_notice(client, task_id) notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={"notification_type": "mail_return", "operator_id": "operator-001"}, ) notification_id = int(notice.json()["data"]["notification_id"]) sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={"operator_id": "operator-001", "operator_confirmed": True}, ) assert sent.json()["data"]["status"] == "发送失败" assert sent.json()["data"]["failure_reason"] == "SMTPException" assert sent.json()["data"]["retry_count"] == 1 asyncio.run(_assert_notification(notification_id, "发送失败", None)) finally: asyncio.run(_cleanup(mail_id, task_id, notification_id, TRACE_ID)) app.dependency_overrides.clear() TRACE_ID = "" @pytest.mark.integration def test_mixed_mail_separates_normal_and_exception_returns( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TRACE_ID TRACE_ID = f"trace-send-mixed-{uuid4()}" payload = _subscription_payload() subscription_path = tmp_path / "申购申请单.pdf" redemption_path = tmp_path / "赎回申请单.pdf" subscription_path.write_bytes(b"subscription-pdf") redemption_path.write_bytes(b"redemption-pdf") payload["attachments"][0]["original_file_path"] = str(subscription_path) payload["attachments"].append({ "filename": "赎回申请单.pdf", "file_hash": f"hash-{uuid4().hex}", "original_file_path": str(redemption_path), "media_type": "application/pdf", "size_bytes": 2048, "document_type": "redemption", "ocr_text": "基金代码 000001 赎回份额 300000", "extracted_fields": { "基金代码": "000001", "基金名称": "测试基金", "账户标识": "ACCT-002", "投资者名称": "测试客户", "申请编号": f"RED-{uuid4().hex[:8]}", "申请日期": "2026-09-10", "代销机构": "测试代销", "赎回份额": "300000", }, "field_confidence": {"基金代码": "0.99", "赎回份额": "0.99"}, "page_evidence": {"基金代码": [1], "赎回份额": [1]}, }) class FakeSuccessSender: def __init__(self, _settings: object) -> None: pass def send_reply(self, request: object) -> SmtpSendResult: del request return SmtpSendResult( status="发送成功", dry_run=False, provider_message_id=f"provider-{uuid4().hex}", failure_reason=None, retry_count=0, request_summary={"provider": "test"}, ) monkeypatch.setattr("app.service.offsite_fund_service.OffsiteSmtpSender", FakeSuccessSender) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_ids: list[str] = [] notification_ids: list[int] = [] try: with TestClient(app) as client: data = _ingest_recognized_mail(client, payload)["data"] mail_id = data["mail_id"] task_ids = [item["task_id"] for item in data["documents"]] for task_id, decision, notification_type in ( (task_ids[0], "确认正常", "normal_return"), (task_ids[1], "确认异常", "exception_return"), ): confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": decision, "operator_id": "operator-001"}, ) assert confirmed.status_code == 200 notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={ "notification_type": notification_type, "operator_id": "operator-001", }, ) assert notice.status_code == 200 notification_id = int(notice.json()["data"]["notification_id"]) notification_ids.append(notification_id) sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={"operator_id": "operator-001", "operator_confirmed": True}, ) assert sent.json()["data"]["status"] == "发送成功" recalculate = client.post( "/api/v1/offsite-fund/settlement-statistics/recalculate", json={"fund_code": "000001", "application_date": "2026-09-10"}, ) assert recalculate.json()["data"]["subscription_amount_yuan"] == "10000.0000" assert recalculate.json()["data"]["redemption_shares"] == "0" asyncio.run(_assert_mail_status(mail_id, "completed")) finally: asyncio.run(_cleanup_many(mail_id, task_ids, notification_ids, TRACE_ID)) app.dependency_overrides.clear() TRACE_ID = "" def _confirm_and_create_notice(client: TestClient, task_id: str) -> None: confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": "确认正常", "operator_id": "operator-001"}, ) assert confirmed.status_code == 200 def _subscription_payload() -> dict[str, object]: suffix = uuid4().hex[:8] return { "imap_uid": f"offsite-{suffix}", "message_id": f"<{suffix}@integration.local>", "sender": "15008108550@163.com", "return_path": "15008108550@163.com", "auth_result": {"spf": "pass", "dkim": "pass"}, "eml_path": "mock/offsite.eml", "attachments": [ { "filename": "申购申请单.pdf", "file_hash": f"hash-{uuid4().hex}", "original_file_path": "mock/申购申请单.pdf", "media_type": "application/pdf", "size_bytes": 2048, "document_type": "subscription", "ocr_text": "基金代码 000001 申购金额 10000", "extracted_fields": { "基金代码": "000001", "基金名称": "测试基金", "账户标识": "ACCT-001", "投资者名称": "测试客户", "申请编号": f"SUB-{suffix}", "申请日期": "2026-09-10", "代销机构": "测试代销", "申购金额": "10000", "金额单位": "元", "最新净值": "1.0000", "基金最新总份额": "1000000", "申请前持有份额": "1000", }, "field_confidence": {"基金代码": "0.99", "申购金额": "0.98"}, "page_evidence": {"基金代码": [1], "申购金额": [1]}, } ], } async def _assert_notification( notification_id: int, status: str, provider_message_id: str | None ) -> None: async with SessionFactory() as session: notice = await session.scalar(select(OffsiteNotification).where( OffsiteNotification.id == notification_id )) assert notice is not None assert notice.status == status assert notice.provider_message_id == provider_message_id if status == "发送成功": assert notice.sent_at is not None else: assert notice.sent_at is None async def _assert_mail_status(mail_id: str, status: str) -> None: async with SessionFactory() as session: mail = await session.scalar(select(OffsiteFundMail).where( OffsiteFundMail.mail_id == mail_id )) assert mail is not None assert mail.status == status async def _cleanup(mail_id: str, task_id: str, notification_id: int, trace_id: str) -> None: await _cleanup_many(mail_id, [task_id] if task_id else [], [notification_id], trace_id) async def _cleanup_many( mail_id: str, task_ids: list[str], notification_ids: list[int], trace_id: 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 notification_ids: await session.execute(delete(OffsiteNotification).where( OffsiteNotification.id.in_(notification_ids) )) for task_id in task_ids: await session.execute(delete(OffsiteQueryRecord).where( OffsiteQueryRecord.task_id == task_id )) await session.execute(delete(OffsiteRuleResult).where( OffsiteRuleResult.task_id == task_id )) await session.execute(delete(OffsiteExecutionPlanTask).where( OffsiteExecutionPlanTask.task_id == task_id )) await session.execute(delete(OffsiteFundDocument).where( OffsiteFundDocument.task_id == task_id )) if mail_id: await session.execute(delete(OffsiteFundAttachment).where( OffsiteFundAttachment.mail_id == mail_id )) await session.execute(delete(OffsiteFundMail).where( OffsiteFundMail.mail_id == mail_id ))