import asyncio from collections.abc import AsyncIterator from datetime import UTC, datetime from decimal import Decimal from pathlib import Path from uuid import uuid4 import pytest from fastapi.testclient import TestClient from sqlalchemy import delete, func, 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.config import get_settings 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, OffsiteFieldCorrection, OffsiteFundAttachment, OffsiteFundDocument, OffsiteFundMail, OffsiteNotification, OffsiteQueryRecord, OffsiteRecognitionAttempt, OffsiteRuleResult, ) from app.service.offsite_document_recognition_adapter import StructuredRecognitionResult from app.service.offsite_fund_service import OffsiteFundService from app.service.offsite_smtp_adapter import SmtpSendResult TEST_TRACE_ID = "" async def override_context() -> RequestContext: return RequestContext( user_id="1", trace_id=TEST_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 的 `POST /api/v1/offsite-fund/recognized-mails` 已下线（真实收信链路由 Worker 直接调用 Service），测试改用 Service 入口，业务校验与副作用完全一致。 """ 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_offsite_recognized_mail_persists_workflow_and_notification() -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = { "imap_uid": uid, "message_id": message_id, "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}", "media_type": "application/pdf", "size_bytes": 2048, "document_type": "subscription", "ocr_text": "基金代码 000001 申购金额 10000", "extracted_fields": { "基金代码": "000001", "基金名称": "测试基金", "账户标识": "ACCT-001", "投资者名称": "测试客户", "申请编号": f"SUB-{uuid4().hex[:8]}", "申请日期": "2026-09-10", "代销机构": "测试代销", "申购金额": "10000", "金额单位": "元", "最新净值": "1.0000", "基金最新总份额": "1000000", "申请前持有份额": "1000", }, "field_confidence": {"基金代码": "0.99", "申购金额": "0.98"}, "page_evidence": {"基金代码": [1], "申购金额": [1]}, } ], } app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" try: with TestClient(app) as client: body = _ingest_recognized_mail(client, payload) assert body["code"] == 0 mail_id = body["data"]["mail_id"] task_id = body["data"]["documents"][0]["task_id"] assert body["data"]["documents"][0]["rule_results"] == { "subscription_minimum_amount": "正常", "subscription_holding_ratio": "正常", "subscription_single_share_limit": "正常", } duplicate = _ingest_recognized_mail(client, payload) assert duplicate["data"] == {"mail_id": mail_id} confirm = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": "确认正常", "operator_id": "1"}, ) assert confirm.status_code == 200 assert confirm.json()["code"] == 0 recalc = client.post( "/api/v1/offsite-fund/settlement-statistics/recalculate", json={"application_date": "2026-09-10"}, ) assert recalc.status_code == 200 assert recalc.json()["data"]["fund_count"] == 0 assert recalc.json()["data"]["items"] == [] fund_recalc = client.post( "/api/v1/offsite-fund/settlement-statistics/recalculate", json={"fund_code": "000001", "application_date": "2026-09-10"}, ) assert fund_recalc.status_code == 200 assert fund_recalc.json()["data"] == { "application_date": "2026-09-10", "fund_code": None, "fund_count": 0, "items": [], } notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={"notification_type": "settlement", "operator_id": "1"}, ) assert notice.status_code == 200 assert notice.json()["data"]["notification_id"] asyncio.run(_assert_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_delete_physically_cleans_workflow_files_and_allows_new_same_kind_mail( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-delete-{uuid4()}" settings = get_settings().model_copy( update={ "offsite_allowed_senders": ("15008108550@163.com",), "offsite_mail_storage_dir": str(tmp_path), } ) monkeypatch.setattr( "app.service.offsite_fund_service.get_settings", lambda: settings ) uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" eml_path = tmp_path / "old-message.eml" attachment_path = tmp_path / "old-subscription.pdf" eml_path.write_bytes(b"old eml") attachment_path.write_bytes(b"old attachment") payload = _subscription_payload(uid, message_id) payload["eml_path"] = str(eml_path) payload["attachments"][0]["original_file_path"] = str(attachment_path) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" try: with TestClient(app) as client: created = _ingest_recognized_mail(client, payload) assert created["code"] == 0 mail_id = created["data"]["mail_id"] document = created["data"]["documents"][0] task_id = document["task_id"] attachment_id = f"{mail_id}-A01" now = datetime.now(UTC).replace(tzinfo=None) async def add_extra_rows() -> None: async with SessionFactory() as session, session.begin(): session.add( OffsiteFieldCorrection( target_type="recognition", mail_id=mail_id, attachment_id=attachment_id, task_id=None, operator_id="1", original_fields={"基金代码": "000001"}, corrected_fields={"基金代码": "000002"}, changed_fields=["基金代码"], created_at=now, ) ) session.add( OffsiteRecognitionAttempt( task_id=None, mail_id=None, attachment_id=None, imap_uid=uid, message_id=message_id, filename="申购申请单.pdf", file_hash=payload["attachments"][0]["file_hash"], original_file_path=str(attachment_path), attempt_no=99, source="automatic", operator_id=None, document_type="subscription", ocr_status="success", llm_status="success", extracted_fields={"基金代码": "000001"}, field_confidence={"基金代码": "0.99"}, missing_fields=[], low_confidence_fields=[], page_evidence={}, status="success", error_message=None, started_at=now, finished_at=now, created_at=now, ) ) session.add( OffsiteFieldCorrection( target_type="nl2sql", mail_id=mail_id, attachment_id=None, task_id=task_id, operator_id="1", original_fields={"最新净值": "1.00"}, corrected_fields={"最新净值": "1.25"}, changed_fields=["最新净值"], created_at=now, ) ) session.add( OffsiteQueryRecord( task_id=task_id, rule_code="subscription_holding_ratio", natural_language_request="查询测试数据", script_path="nl2sql_yc.py", result_summary={"rows": [{"nav": "1.25"}]}, status="success", error_message=None, created_at=now, ) ) session.add( OffsiteNotification( notification_type="settlement", business_key=task_id, receiver_id="receiver", operator_id="1", agent_draft="测试通知", final_content="测试通知", payload={"task_id": task_id}, status="待发送", retry_count=0, provider_message_id=None, failure_reason=None, created_at=now, sent_at=None, updated_at=now, ) ) asyncio.run(add_extra_rows()) deleted = client.post( f"/api/v1/offsite-fund/mails/{mail_id}/deletions", json={"operator_id": "1"}, ) assert deleted.status_code == 200 assert deleted.json()["code"] == 0 assert deleted.json()["data"]["status"] == "deleted" assert deleted.json()["data"]["deleted_file_count"] == 2 assert not eml_path.exists() assert not attachment_path.exists() listed = client.get("/api/v1/offsite-fund/mails") assert listed.status_code == 200 assert all(item["mail_id"] != mail_id for item in listed.json()["data"]["items"]) detail = client.get(f"/api/v1/offsite-fund/mails/{mail_id}") assert detail.status_code == 200 assert detail.json()["code"] == 404 async def assert_deleted_rows() -> None: async with SessionFactory() as session: for model, criterion in ( (OffsiteFundMail, OffsiteFundMail.mail_id == mail_id), (OffsiteFundAttachment, OffsiteFundAttachment.mail_id == mail_id), (OffsiteFundDocument, OffsiteFundDocument.task_id == task_id), (OffsiteRecognitionAttempt, OffsiteRecognitionAttempt.mail_id == mail_id), (OffsiteFieldCorrection, OffsiteFieldCorrection.mail_id == mail_id), (OffsiteQueryRecord, OffsiteQueryRecord.task_id == task_id), (OffsiteRuleResult, OffsiteRuleResult.task_id == task_id), (OffsiteExecutionPlanTask, OffsiteExecutionPlanTask.task_id == task_id), (OffsiteNotification, OffsiteNotification.business_key == task_id), ): assert await _count(session, model, criterion) == 0 assert await _count( session, InteractionAudit, InteractionAudit.action_type == "offsite.mail_deleted", ) >= 1 asyncio.run(assert_deleted_rows()) # 删除后重新收到同类邮件：新消息号、新 UID，应能重新建立完整业务链路。 new_eml_path = tmp_path / "new-message.eml" new_attachment_path = tmp_path / "new-subscription.pdf" new_eml_path.write_bytes(b"new eml") new_attachment_path.write_bytes(b"new attachment") new_payload = _subscription_payload( f"offsite-{uuid4()}", f"<{uuid4()}@integration.local>", ) new_payload["eml_path"] = str(new_eml_path) new_payload["attachments"][0]["original_file_path"] = str(new_attachment_path) with TestClient(app) as client: recreated = _ingest_recognized_mail(client, new_payload) assert recreated["code"] == 0 assert recreated["data"]["mail_id"] == mail_id assert recreated["data"]["documents"] finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_trigger_nl2sql_uses_query_dict_adapter() -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-nl2sql-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = _subscription_payload(uid, message_id) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" try: with TestClient(app) as client: body = _ingest_recognized_mail(client, payload) mail_id = body["data"]["mail_id"] task_id = body["data"]["documents"][0]["task_id"] response = client.post( f"/api/tasks/{task_id}/trigger-agent-nl2sql", json={"operator_id": "1", "manual_confirmed": True}, ) assert response.status_code == 200 data = response.json()["data"] assert data["task_id"] == task_id assert len(data["queries"]) == 2 assert {item["status"] for item in data["queries"]} <= { "ready", "success", "need_confirmation", "rejected", "error", "query_failed", } asyncio.run(_assert_nl2sql_rows(task_id)) finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_nl2sql_success_completes_subscription_plan( monkeypatch: pytest.MonkeyPatch, ) -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-nl2sql-success-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = _subscription_payload(uid, message_id) class SuccessfulNl2SqlAdapter: script_path = "nl2sql_yc.py" def query(self, question: str, context: RequestContext) -> dict[str, object]: del context if "申请前持有份额" in question: rows = [{ "total_fund_shares": Decimal("1000000"), "nav": Decimal("1.25"), "total_quantity": Decimal("1000"), }] else: rows = [{ "total_fund_shares": Decimal("1000000"), "nav": Decimal("1.25"), }] return {"status": "success", "data": {"total": 1, "rows": rows}} monkeypatch.setattr( "app.service.offsite_fund_service.OffsiteNl2SqlAdapter", SuccessfulNl2SqlAdapter, ) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" 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"] triggered = client.post( f"/api/tasks/{task_id}/trigger-agent-nl2sql", json={"operator_id": "1", "manual_confirmed": True}, ) assert triggered.status_code == 200 assert triggered.json()["data"]["status"] == "planned" assert all( item["status"] == "success" for item in triggered.json()["data"]["queries"] ) confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": "确认正常", "operator_id": "1"}, ) assert confirmed.status_code == 200 assert confirmed.json()["code"] == 0 asyncio.run(_assert_completed_subscription_plan(task_id)) finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_low_confidence_document_goes_to_recognition_exception() -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-low-confidence-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = _subscription_payload(uid, message_id) payload["attachments"][0]["field_confidence"] = {"基金代码": "0.79", "申购金额": "0.99"} app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" try: with TestClient(app) as client: body = _ingest_recognized_mail(client, payload) mail_id = body["data"]["mail_id"] document = body["data"]["documents"][0] task_id = document["task_id"] assert document["status"] == "recognition_exception" blocked = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": "确认正常", "operator_id": "1"}, ) assert blocked.status_code == 200 assert blocked.json()["code"] == 422 finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_recognition_retry_recovers_document_without_overwriting_attachment( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-recognition-retry-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = _subscription_payload(uid, message_id) attachment_path = tmp_path / "retry-subscription.pdf" attachment_path.write_bytes(b"retry-document") original_fields = dict(payload["attachments"][0]["extracted_fields"]) payload["attachments"][0]["original_file_path"] = str(attachment_path) payload["attachments"][0]["field_confidence"] = {"基金代码": "0.79"} settings = get_settings().model_copy( update={ "offsite_allowed_senders": ("15008108550@163.com",), "offsite_ocr_enabled": True, "offsite_deepseek_enabled": True, } ) assert settings.offsite_ocr_enabled is True assert settings.offsite_deepseek_enabled is True recognized_fields: list[dict[str, object]] = [] class SuccessfulRetryRecognizer: def __init__(self, _settings: object) -> None: pass async def recognize(self, source: object) -> StructuredRecognitionResult: del source result = StructuredRecognitionResult( document_type="subscription", extracted_fields={ **original_fields, "基金代码": "000002", }, field_confidence={"基金代码": "0.99"}, missing_fields=(), low_confidence_fields=(), page_evidence={"基金代码": [{"page": 1}]}, ocr_text="申购申请单", ocr_status="success", llm_status="success", ) recognized_fields.append(result.extracted_fields) return result monkeypatch.setattr("app.service.offsite_fund_service.get_settings", lambda: settings) monkeypatch.setattr( "app.service.offsite_fund_service.OffsiteDocumentRecognitionAdapter", SuccessfulRetryRecognizer, ) app.dependency_overrides[build_request_context] = override_context app.dependency_overrides[get_session] = override_session mail_id = "" task_id = "" 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"] assert data["documents"][0]["status"] == "recognition_exception" retried = client.post( f"/api/v1/offsite-fund/documents/{task_id}/recognition-retries", json={"operator_id": "1"}, ) assert retried.status_code == 200 assert recognized_fields assert retried.json()["data"]["status"] == "planned", retried.json() asyncio.run(_assert_recognition_recovery_rows( mail_id, task_id, original_fields )) finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" @pytest.mark.integration def test_offsite_service_rejects_missing_permission() -> None: async def no_permission_context() -> RequestContext: return RequestContext( user_id="1", trace_id=f"trace-offsite-denied-{uuid4()}", roles=("operator",), permissions=(), ) payload = _subscription_payload(f"offsite-{uuid4()}", f"<{uuid4()}@integration.local>") app.dependency_overrides[build_request_context] = no_permission_context app.dependency_overrides[get_session] = override_session try: with TestClient(app) as client: body = _ingest_recognized_mail(client, payload) assert body["code"] == 403 assert "缺少场外基金操作权限" in body["message"] finally: app.dependency_overrides.clear() @pytest.mark.integration def test_offsite_mail_return_send_updates_notification_without_external_call( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: global TEST_TRACE_ID TEST_TRACE_ID = f"trace-offsite-send-{uuid4()}" uid = f"offsite-{uuid4()}" message_id = f"<{uuid4()}@integration.local>" payload = _subscription_payload(uid, message_id) attachment_path = tmp_path / "申购申请单.pdf" attachment_path.write_bytes(b"pdf-bytes") payload["attachments"][0]["original_file_path"] = str(attachment_path) class FakeDryRunSender: def __init__(self, _settings: object) -> None: pass def send_reply(self, request: object) -> SmtpSendResult: del request return SmtpSendResult( status="待发送", dry_run=True, provider_message_id=None, failure_reason=None, retry_count=0, request_summary={"provider": "test"}, ) monkeypatch.setattr("app.service.offsite_fund_service.OffsiteSmtpSender", FakeDryRunSender) 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"] confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", json={"decision": "确认正常", "operator_id": "1"}, ) assert confirmed.status_code == 200 notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={"notification_type": "mail_return", "operator_id": "1"}, ) assert notice.status_code == 200 notification_id = int(notice.json()["data"]["notification_id"]) sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={ "operator_id": "1", "operator_confirmed": True, "final_content": "运营确认后的回复正文", }, ) assert sent.status_code == 200 assert sent.json()["data"] == { "notification_id": str(notification_id), "status": "待发送", "dry_run": True, "provider_message_id": None, "failure_reason": None, "retry_count": 0, } asyncio.run(_assert_notification_pending(notification_id)) finally: asyncio.run(_cleanup_offsite_rows(mail_id, task_id, TEST_TRACE_ID)) app.dependency_overrides.clear() TEST_TRACE_ID = "" def _subscription_payload(uid: str, message_id: str) -> dict[str, object]: return { "imap_uid": uid, "message_id": message_id, "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-{uuid4().hex[:8]}", "申请日期": "2026-09-10", "代销机构": "测试代销", "申购金额": "10000", "金额单位": "元", "最新净值": "1.0000", "基金最新总份额": "1000000", "申请前持有份额": "1000", }, "field_confidence": {"基金代码": "0.99", "申购金额": "0.98"}, "page_evidence": {"基金代码": [1], "申购金额": [1]}, } ], } async def _assert_offsite_rows(mail_id: str, task_id: str, trace_id: str) -> None: async with SessionFactory() as session: assert await _count(session, OffsiteFundMail, OffsiteFundMail.mail_id == mail_id) == 1 assert await _count( session, OffsiteFundAttachment, OffsiteFundAttachment.mail_id == mail_id ) == 1 assert await _count( session, OffsiteFundDocument, OffsiteFundDocument.task_id == task_id ) == 1 assert await _count(session, OffsiteRuleResult, OffsiteRuleResult.task_id == task_id) == 3 assert await _count( session, OffsiteExecutionPlanTask, OffsiteExecutionPlanTask.task_id == task_id ) == 9 assert await _count( session, OffsiteNotification, OffsiteNotification.business_key == task_id ) == 1 assert await _count( session, InteractionAudit, InteractionAudit.action_type.like("offsite.%") ) >= 1 async def _assert_nl2sql_rows(task_id: str) -> None: async with SessionFactory() as session: assert await _count(session, OffsiteQueryRecord, OffsiteQueryRecord.task_id == task_id) == 2 tasks = (await session.execute(select(OffsiteExecutionPlanTask).where( OffsiteExecutionPlanTask.task_id == task_id, OffsiteExecutionPlanTask.stage == "查询", OffsiteExecutionPlanTask.rule_code != "subscription_minimum_amount", ))).scalars().all() assert {task.status for task in tasks} <= {"已完成", "查询失败", "无法判断"} assert all(task.output_json is not None for task in tasks) async def _assert_completed_subscription_plan(task_id: str) -> None: async with SessionFactory() as session: document = await session.scalar( select(OffsiteFundDocument).where(OffsiteFundDocument.task_id == task_id) ) assert document is not None assert document.status == "operator_confirmed" rules = ( await session.execute( select(OffsiteRuleResult).where(OffsiteRuleResult.task_id == task_id) ) ).scalars().all() assert {item.rule_code: item.result for item in rules} == { "subscription_minimum_amount": "正常", "subscription_holding_ratio": "正常", "subscription_single_share_limit": "正常", } query_records = ( await session.execute( select(OffsiteQueryRecord).where( OffsiteQueryRecord.task_id == task_id ) ) ).scalars().all() assert {record.status for record in query_records} == {"success"} plan_tasks = ( await session.execute( select(OffsiteExecutionPlanTask).where( OffsiteExecutionPlanTask.task_id == task_id ) ) ).scalars().all() assert all(task.status == "已完成" for task in plan_tasks if task.stage != "查询") assert all( task.status in {"已完成", "不适用"} for task in plan_tasks if task.stage == "查询" ) async def _assert_recognition_recovery_rows( mail_id: str, task_id: str, original_fields: dict[str, object] ) -> None: async with SessionFactory() as session: attachment = await session.scalar( select(OffsiteFundAttachment).where( OffsiteFundAttachment.mail_id == mail_id ) ) document = await session.scalar( select(OffsiteFundDocument).where( OffsiteFundDocument.task_id == task_id ) ) attempts = ( await session.execute( select(OffsiteRecognitionAttempt).where( OffsiteRecognitionAttempt.task_id == task_id ) ) ).scalars().all() assert attachment is not None assert document is not None assert attachment.extracted_fields == original_fields assert document.fund_code == "000002" assert document.status == "planned" assert len(attempts) == 1 assert attempts[0].source == "manual" assert attempts[0].status == "success" plan_tasks = ( await session.execute( select(OffsiteExecutionPlanTask).where( OffsiteExecutionPlanTask.task_id == task_id ) ) ).scalars().all() assert any(task.status == "待执行" for task in plan_tasks) assert await _count( session, InteractionAudit, InteractionAudit.action_type == "offsite.document_recognition_recovered", ) >= 1 async def _assert_notification_pending(notification_id: int) -> 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 == "待发送" assert notice.provider_message_id is None assert notice.failure_reason is None assert notice.sent_at is None async def _count(session: AsyncSession, model: type, criterion: object) -> int: return int(await session.scalar(select(func.count()).select_from(model).where(criterion)) or 0) async def _cleanup_offsite_rows(mail_id: str, task_id: str, trace_id: str) -> None: if not mail_id and not task_id: return 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_id: await session.execute( delete(OffsiteFieldCorrection).where( OffsiteFieldCorrection.task_id == task_id ) ) await session.execute(delete(OffsiteRecognitionAttempt).where( OffsiteRecognitionAttempt.task_id == task_id )) await session.execute( delete(OffsiteNotification).where(OffsiteNotification.business_key == task_id) ) 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(OffsiteFieldCorrection).where( OffsiteFieldCorrection.mail_id == mail_id ) ) await session.execute(delete(OffsiteRecognitionAttempt).where( OffsiteRecognitionAttempt.mail_id == mail_id )) await session.execute( delete(OffsiteFundAttachment).where(OffsiteFundAttachment.mail_id == mail_id) ) await session.execute(delete(OffsiteFundMail).where(OffsiteFundMail.mail_id == mail_id))