404 lines
16 KiB
Python
404 lines
16 KiB
Python
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": "1"},
|
|
)
|
|
notification_id = int(notice.json()["data"]["notification_id"])
|
|
|
|
first = client.post(
|
|
f"/api/v1/offsite-fund/notifications/{notification_id}/send",
|
|
json={"operator_id": "1", "operator_confirmed": True},
|
|
)
|
|
second = client.post(
|
|
f"/api/v1/offsite-fund/notifications/{notification_id}/send",
|
|
json={"operator_id": "1", "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": "1"},
|
|
)
|
|
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},
|
|
)
|
|
|
|
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": "1"},
|
|
)
|
|
assert confirmed.status_code == 200
|
|
notice = client.post(
|
|
f"/api/v1/offsite-fund/documents/{task_id}/notifications",
|
|
json={
|
|
"notification_type": notification_type,
|
|
"operator_id": "1",
|
|
},
|
|
)
|
|
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": "1", "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": "1"},
|
|
)
|
|
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
|
|
))
|