384 lines
14 KiB
Python
384 lines
14 KiB
Python
"""场外附件原始文件预览接口的集成测试。
|
|
|
|
覆盖：PDF 申购单内联预览、JPG 等图片内联预览、media_type 缺失时按后缀兜底、
|
|
非预览类型（Excel、SVG）强制下载、附件不存在、文件越界或缺失、权限拦截，
|
|
以及预览动作必须留下审计记录。
|
|
|
|
统一在单事件循环内调用应用（httpx.ASGITransport），避免连接池里的连接被
|
|
跨事件循环复用。
|
|
"""
|
|
|
|
from collections.abc import AsyncIterator
|
|
from datetime import UTC, date, datetime
|
|
from pathlib import Path
|
|
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.config import get_settings
|
|
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, OffsiteFundMail
|
|
|
|
PREVIEW_PATH = "/api/v1/offsite-fund/attachments/{attachment_id}/file"
|
|
PDF_BYTES = b"%PDF-1.4\n% offsite subscription preview\n"
|
|
JPEG_BYTES = b"\xff\xd8\xff\xe0offsite-jpeg-preview\xff\xd9"
|
|
WORKBOOK_BYTES = b"PK\x03\x04offsite-workbook"
|
|
SVG_BYTES = b"<svg xmlns='http://www.w3.org/2000/svg'><script/></svg>"
|
|
|
|
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="1",
|
|
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_preview(attachment_id: str, **params: str) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.get(
|
|
PREVIEW_PATH.format(attachment_id=attachment_id), params=params or None
|
|
)
|
|
|
|
|
|
def _use_storage_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""把附件存储根切到临时目录，避免触碰真实收件目录。"""
|
|
root = tmp_path / "offsite-mail"
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
settings = get_settings().model_copy(update={"offsite_mail_storage_dir": str(root)})
|
|
monkeypatch.setattr("app.service.offsite_fund_service.get_settings", lambda: settings)
|
|
return root
|
|
|
|
|
|
async def _seed_attachment(
|
|
file_path: Path,
|
|
filename: str,
|
|
media_type: str,
|
|
document_type: str = "subscription",
|
|
) -> tuple[str, str]:
|
|
mail_id = f"M{uuid4().hex[:12]}"
|
|
attachment_id = f"{mail_id}-A01"
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
session.add(OffsiteFundMail(
|
|
mail_id=mail_id,
|
|
imap_uid=f"uid-{mail_id}",
|
|
message_id=f"<{mail_id}@integration.local>",
|
|
received_date=date(2026, 9, 10),
|
|
sender="15008108550@163.com",
|
|
return_path="15008108550@163.com",
|
|
auth_result={"spf": "pass", "dkim": "pass"},
|
|
original_eml_path=str(file_path.parent / "message.eml"),
|
|
status="recognized",
|
|
retry_count=0,
|
|
created_at=now,
|
|
updated_at=now,
|
|
))
|
|
session.add(OffsiteFundAttachment(
|
|
attachment_id=attachment_id,
|
|
mail_id=mail_id,
|
|
filename=filename,
|
|
file_hash=uuid4().hex,
|
|
media_type=media_type,
|
|
size_bytes=file_path.stat().st_size if file_path.exists() else 0,
|
|
document_type=document_type,
|
|
original_file_path=str(file_path),
|
|
ocr_text="识别文本",
|
|
extracted_fields={"基金代码": "159511"},
|
|
field_confidence={"基金代码": "0.99"},
|
|
page_evidence={"基金代码": [1]},
|
|
status="recognized",
|
|
created_at=now,
|
|
))
|
|
return mail_id, attachment_id
|
|
|
|
|
|
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(mail_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 mail_id:
|
|
await session.execute(
|
|
delete(OffsiteFundAttachment).where(OffsiteFundAttachment.mail_id == mail_id)
|
|
)
|
|
await session.execute(
|
|
delete(OffsiteFundMail).where(OffsiteFundMail.mail_id == mail_id)
|
|
)
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_pdf_attachment_previews_inline(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.pdf"
|
|
file_path.write_bytes(PDF_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path, "申购申请单.pdf", "application/pdf"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/pdf"
|
|
assert response.headers["content-disposition"].startswith("inline")
|
|
assert response.headers["x-content-type-options"] == "nosniff"
|
|
assert response.content == PDF_BYTES
|
|
assert await _count_audit("offsite.attachment_viewed") == 1
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_image_attachment_previews_inline(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""扫描件图片（JPG）必须能直接内联预览。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.jpg"
|
|
file_path.write_bytes(JPEG_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path, "申购申请单扫描件.jpg", "image/jpeg"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "image/jpeg"
|
|
assert response.headers["content-disposition"].startswith("inline")
|
|
assert response.content == JPEG_BYTES
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_pdf_media_type_falls_back_to_file_suffix(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""识别时 media_type 存成通用二进制，PDF 申购单仍然要能直接预览。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.pdf"
|
|
file_path.write_bytes(PDF_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path, "赎回申请单.pdf", "application/octet-stream"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/pdf"
|
|
assert response.headers["content-disposition"].startswith("inline")
|
|
assert response.content == PDF_BYTES
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_non_previewable_attachment_forces_download(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.xlsx"
|
|
file_path.write_bytes(WORKBOOK_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path,
|
|
"赎回申请单.xlsx",
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
"redemption",
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id, disposition="inline")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/octet-stream"
|
|
assert response.headers["content-disposition"].startswith("attachment")
|
|
assert response.content == WORKBOOK_BYTES
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_svg_attachment_is_not_inlined(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""SVG 可能携带脚本，必须降级为下载而不是内联渲染。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.svg"
|
|
file_path.write_bytes(SVG_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path, "申购申请单.svg", "image/svg+xml"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id, disposition="inline")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/octet-stream"
|
|
assert response.headers["content-disposition"].startswith("attachment")
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_unknown_attachment_returns_not_found(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
_use_storage_root(tmp_path, monkeypatch)
|
|
try:
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview("M000000000000-A01")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == 404
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_file_outside_storage_root_is_rejected(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""数据库路径指向存储根之外时必须拒绝，阻断路径穿越。"""
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
_use_storage_root(tmp_path, monkeypatch)
|
|
outside = tmp_path / "outside.pdf"
|
|
outside.write_bytes(PDF_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
outside, "申购申请单.pdf", "application/pdf"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == 404
|
|
assert PDF_BYTES not in response.content
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_missing_file_on_disk_returns_not_found(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
root / "A01_missing.pdf", "申购申请单.pdf", "application/pdf"
|
|
)
|
|
_install_context(("operator",), ("offsite:read",))
|
|
response = await _get_preview(attachment_id)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == 404
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_preview_requires_offsite_permission(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
global TRACE_ID
|
|
TRACE_ID = f"trace-attachment-{uuid4()}"
|
|
root = _use_storage_root(tmp_path, monkeypatch)
|
|
file_path = root / "A01.pdf"
|
|
file_path.write_bytes(PDF_BYTES)
|
|
mail_id = ""
|
|
try:
|
|
mail_id, attachment_id = await _seed_attachment(
|
|
file_path, "申购申请单.pdf", "application/pdf"
|
|
)
|
|
_install_context(("operator",), ())
|
|
missing_permission = await _get_preview(attachment_id)
|
|
_install_context(("customer",), ("offsite:read",))
|
|
wrong_role = await _get_preview(attachment_id)
|
|
|
|
assert missing_permission.json()["code"] == 403
|
|
assert wrong_role.json()["code"] == 403
|
|
finally:
|
|
await _cleanup(mail_id)
|
|
app.dependency_overrides.clear()
|
|
TRACE_ID = ""
|