690 lines
27 KiB
Python
690 lines
27 KiB
Python
import asyncio
|
|
from collections.abc import AsyncIterator
|
|
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,
|
|
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"] == 1
|
|
assert recalc.json()["data"]["items"][0]["subscription_amount_yuan"] == "0"
|
|
|
|
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_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(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(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))
|