327 lines
12 KiB
Python
327 lines
12 KiB
Python
"""单据规则判定结果查询与重新判定接口的集成测试。
|
|
|
|
覆盖：申购三条规则与赎回两条规则（含大额赎回）的结论返回、汇总统计、
|
|
用已落库查询结果重新判定后结论正确、没有查询记录时按"无法判断"并给出原因、
|
|
单据不存在、越权，以及两个动作都必须留下审计记录。
|
|
|
|
统一在单事件循环内调用应用（httpx.ASGITransport），避免连接池里的连接被
|
|
跨事件循环复用。
|
|
"""
|
|
|
|
from collections.abc import AsyncIterator
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
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.infrastructure.db import SessionFactory, engine
|
|
from app.main import app
|
|
from app.model.audit import InteractionAudit
|
|
from app.model.offsite_fund import (
|
|
OffsiteFundAttachment,
|
|
OffsiteFundDocument,
|
|
OffsiteQueryRecord,
|
|
OffsiteRuleResult,
|
|
)
|
|
|
|
RESULTS_PATH = "/api/v1/offsite-fund/documents/{task_id}/rule-results"
|
|
RECALC_PATH = "/api/v1/offsite-fund/documents/{task_id}/rule-results/recalculations"
|
|
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def _dispose_engine_after_test() -> AsyncIterator[None]:
|
|
"""用例结束后释放连接池，避免连接绑定在已关闭的事件循环上。"""
|
|
yield
|
|
await engine.dispose()
|
|
|
|
|
|
async def _override_session() -> AsyncIterator[AsyncSession]:
|
|
async with SessionFactory() as session:
|
|
yield session
|
|
|
|
|
|
def _install_context(roles: tuple[str, ...], permissions: tuple[str, ...]) -> None:
|
|
async def override_context() -> RequestContext:
|
|
return RequestContext(
|
|
user_id="990911000",
|
|
trace_id=TRACE_ID,
|
|
roles=roles,
|
|
permissions=permissions,
|
|
data_scope="all",
|
|
)
|
|
|
|
app.dependency_overrides[build_request_context] = override_context
|
|
app.dependency_overrides[get_session] = _override_session
|
|
|
|
|
|
async def _get_results(task_id: str) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.get(RESULTS_PATH.format(task_id=task_id))
|
|
|
|
|
|
async def _recalculate(task_id: str) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.post(
|
|
RECALC_PATH.format(task_id=task_id), json={"operator_id": "990911000"}
|
|
)
|
|
|
|
|
|
async def _seed_document(
|
|
document_type: str,
|
|
extracted_fields: dict[str, object],
|
|
document_status: str = "planned",
|
|
) -> tuple[str, str]:
|
|
"""写入单据与原始附件识别字段，返回 (task_id, attachment_id)。"""
|
|
task_id = f"T{uuid4().hex[:14]}"
|
|
attachment_id = f"A{uuid4().hex[:14]}"
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
session.add(OffsiteFundDocument(
|
|
task_id=task_id,
|
|
mail_id=f"M{uuid4().hex[:12]}",
|
|
attachment_id=attachment_id,
|
|
document_type=document_type,
|
|
fund_code="15911",
|
|
account_identifier="10001",
|
|
status=document_status,
|
|
operator_decision="未处理",
|
|
created_at=now,
|
|
updated_at=now,
|
|
))
|
|
session.add(OffsiteFundAttachment(
|
|
attachment_id=attachment_id,
|
|
mail_id=f"M{uuid4().hex[:12]}",
|
|
filename="规则判定测试附件.pdf",
|
|
file_hash=uuid4().hex,
|
|
media_type="application/pdf",
|
|
size_bytes=1024,
|
|
document_type=document_type,
|
|
original_file_path="mock/规则判定测试附件.pdf",
|
|
ocr_text="规则判定测试",
|
|
extracted_fields=extracted_fields,
|
|
field_confidence={},
|
|
page_evidence={},
|
|
status="recognized",
|
|
created_at=now,
|
|
))
|
|
return task_id, attachment_id
|
|
|
|
|
|
async def _seed_query_record(
|
|
task_id: str, rule_code: str, rows: list[dict[str, object]]
|
|
) -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
session.add(OffsiteQueryRecord(
|
|
task_id=task_id,
|
|
rule_code=rule_code,
|
|
natural_language_request=f"{rule_code} 的只读核对查询",
|
|
script_path="nl2sql_yc.py",
|
|
result_summary={"status": "success", "data": {"total": len(rows), "rows": rows}},
|
|
status="success",
|
|
error_message=None,
|
|
created_at=now,
|
|
))
|
|
|
|
|
|
async def _count_audit(action_type: str) -> int:
|
|
async with SessionFactory() as session:
|
|
rows = await session.execute(select(InteractionAudit).where(
|
|
InteractionAudit.action_type == action_type,
|
|
InteractionAudit.detail["trace_id"].as_string() == TRACE_ID,
|
|
))
|
|
return len(rows.scalars().all())
|
|
|
|
|
|
async def _cleanup(task_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 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(OffsiteFundAttachment).where(
|
|
OffsiteFundAttachment.mail_id.in_(
|
|
select(OffsiteFundDocument.mail_id).where(
|
|
OffsiteFundDocument.task_id == task_id
|
|
)
|
|
)
|
|
))
|
|
await session.execute(delete(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == task_id
|
|
))
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_subscription_rules_return_three_results() -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
task_id = ""
|
|
try:
|
|
task_id, _ = await _seed_document(
|
|
"subscription", {"申购金额": "200000000", "金额单位": "元"}
|
|
)
|
|
await _seed_query_record(task_id, "subscription_holding_ratio", [{
|
|
"nav": "1.250000",
|
|
"total_fund_shares": "10000000000.0000",
|
|
"total_quantity": "100000000.0000",
|
|
}])
|
|
await _seed_query_record(task_id, "subscription_single_share_limit", [{
|
|
"nav": "1.250000",
|
|
"total_fund_shares": "10000000000.0000",
|
|
}])
|
|
_install_context(("operator",), ("offsite:read", "offsite:write"))
|
|
|
|
recalculated = await _recalculate(task_id)
|
|
assert recalculated.status_code == 200
|
|
assert recalculated.json()["code"] == 0
|
|
data = recalculated.json()["data"]
|
|
assert [item["rule_name"] for item in data["rules"]] == [
|
|
"申购最低金额", "申购后单一投资者持有比例", "申购单笔份额上限",
|
|
]
|
|
assert data["summary"] == {"total": 3, "normal": 3, "abnormal": 0, "unknown": 0}
|
|
assert all(item["status"] == "success" for item in data["query_snapshot"])
|
|
|
|
queried = await _get_results(task_id)
|
|
assert queried.json()["data"]["summary"]["normal"] == 3
|
|
assert queried.json()["data"]["rules"][1]["database_value"]["最新净值"] == "1.250000"
|
|
assert await _count_audit("offsite.rule_results_recalculated") == 1
|
|
assert await _count_audit("offsite.rule_results_viewed") == 1
|
|
finally:
|
|
await _cleanup(task_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_redemption_rules_flag_large_redemption() -> None:
|
|
"""赎回 30% 超过 20% 阈值时必须判定为异常，重新判定接口要能给出结论。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
task_id = ""
|
|
try:
|
|
task_id, _ = await _seed_document("redemption", {"赎回份额": "300000"})
|
|
await _seed_query_record(task_id, "redemption_large_ratio", [{
|
|
"total_fund_shares": "1000000",
|
|
}])
|
|
await _seed_query_record(task_id, "redemption_available_quantity", [{
|
|
"available_quantity": "500000",
|
|
}])
|
|
_install_context(("operator",), ("offsite:write",))
|
|
|
|
recalculated = await _recalculate(task_id)
|
|
data = recalculated.json()["data"]
|
|
rules = {item["rule_code"]: item for item in data["rules"]}
|
|
assert rules["redemption_large_ratio"]["result"] == "异常"
|
|
assert rules["redemption_large_ratio"]["calculation"]["赎回比例"] == "0.3"
|
|
assert rules["redemption_available_quantity"]["result"] == "正常"
|
|
assert data["summary"] == {"total": 2, "normal": 1, "abnormal": 1, "unknown": 0}
|
|
finally:
|
|
await _cleanup(task_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_recalculation_without_query_records_marks_unknown() -> None:
|
|
"""没有查询记录时不得猜测结论，必须给出无法判断和具体原因。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
task_id = ""
|
|
try:
|
|
task_id, _ = await _seed_document("redemption", {"赎回份额": "300000"})
|
|
_install_context(("operator",), ("offsite:write",))
|
|
|
|
recalculated = await _recalculate(task_id)
|
|
data = recalculated.json()["data"]
|
|
assert data["summary"] == {"total": 2, "normal": 0, "abnormal": 0, "unknown": 2}
|
|
assert data["document_status"] == "query_failed"
|
|
assert all(
|
|
item["reason"] == "尚未执行该规则的 NL2SQL 查询"
|
|
for item in data["query_snapshot"]
|
|
)
|
|
finally:
|
|
await _cleanup(task_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_recalculation_keeps_recognition_status() -> None:
|
|
"""识别异常属于识别阶段状态，重新判定不得把它改成查询失败。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
task_id = ""
|
|
try:
|
|
task_id, _ = await _seed_document(
|
|
"subscription",
|
|
{"申购金额": "200000000", "金额单位": "元"},
|
|
document_status="recognition_exception",
|
|
)
|
|
_install_context(("operator",), ("offsite:write",))
|
|
|
|
recalculated = await _recalculate(task_id)
|
|
data = recalculated.json()["data"]
|
|
assert data["document_status"] == "recognition_exception"
|
|
# 金额已识别，所以最低金额规则仍可判定；两条依赖查询的规则必须为无法判断。
|
|
assert data["summary"] == {"total": 3, "normal": 1, "abnormal": 0, "unknown": 2}
|
|
finally:
|
|
await _cleanup(task_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_rule_results_unknown_document_returns_not_found() -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
try:
|
|
_install_context(("operator",), ("offsite:read", "offsite:write"))
|
|
queried = await _get_results("T000000000000")
|
|
recalculated = await _recalculate("T000000000000")
|
|
|
|
assert queried.json()["code"] == 404
|
|
assert recalculated.json()["code"] == 404
|
|
assert recalculated.json()["message"] == "单据不存在"
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_rule_results_requires_offsite_permission() -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-rule-results-{uuid4()}"
|
|
task_id = ""
|
|
try:
|
|
task_id, _ = await _seed_document("redemption", {"赎回份额": "300000"})
|
|
_install_context(("operator",), ())
|
|
read_denied = await _get_results(task_id)
|
|
recalculate_denied = await _recalculate(task_id)
|
|
_install_context(("customer",), ("offsite:read", "offsite:write"))
|
|
wrong_role = await _get_results(task_id)
|
|
|
|
assert read_denied.json()["code"] == 403
|
|
assert recalculate_denied.json()["code"] == 403
|
|
assert wrong_role.json()["code"] == 403
|
|
finally:
|
|
await _cleanup(task_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|