xsl
00ef56b08b
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
- errors.py：IDEMPOTENCY_CONFLICT / TASK_STATE_INVALID 的 hint 改为「首页报告」口径（App 无任务中心页，经 hint 透出会误导） - report.html.j2：head 增加 author/description/generator meta（WeasyPrint 落入 PDF 属性，AI 生成内容标识 CR-05） - test_report_export.py：+2 单测（模板 meta 渲染 + PDF 产物 Info 字典解压断言，不新增依赖） 多 agent 评审 R3/R4 产物；单测 576 passed
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""报告 PDF 导出（Phase A）的单测：渲染器与路由接线，不连库不发网。
|
||
|
||
- 渲染器：构造样例 result（32 维，ok/skipped 混合 + 色卡维度）→ 断言
|
||
输出为合法 PDF 头且非空；产物缺 copy → 40900；url_fetcher 全拒。
|
||
- 路由：参照 test_admin_app.py 的方式——create_app + ASGITransport +
|
||
dependency_overrides 换假服务，只验证接线与信封/错误码透传；
|
||
真实的资格/幂等/配额语义由 E2E 与后续集成测试覆盖。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import zlib
|
||
from collections.abc import AsyncIterator
|
||
from contextlib import asynccontextmanager
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
|
||
from apps.api.deps import current_user_id
|
||
from apps.api.main import create_app
|
||
from apps.api.routers.reports import report_export_service_dep
|
||
from packages.report.dimensions import DIMENSIONS
|
||
from packages.report.export_service import ExportFile, ExportSubmission
|
||
from packages.report.pdf import renderer as pdf_renderer
|
||
from packages.report.pdf.renderer import (
|
||
ReportPdfMeta,
|
||
_reject_all_urls,
|
||
render_report_pdf,
|
||
)
|
||
from packages.shared.errors import AppError, ErrorCode
|
||
|
||
META = ReportPdfMeta(report_id=101, version=1, generated_at=datetime(2026, 8, 29, 8, 0, tzinfo=UTC))
|
||
|
||
|
||
def _sample_result() -> dict[str, Any]:
|
||
"""与流水线产物同构的最小样例：21 项 ok（含色卡维度），11 项 skipped。"""
|
||
dimensions: list[dict[str, Any]] = []
|
||
for spec in DIMENSIONS:
|
||
if spec.basic_coverage == "none":
|
||
dimensions.append(
|
||
{
|
||
"number": spec.number,
|
||
"key": spec.key,
|
||
"value": None,
|
||
"confidence": 0.0,
|
||
"source": "none",
|
||
"evidence": "初级版报告不覆盖该维度",
|
||
"rule_hits": [],
|
||
"status": "skipped",
|
||
}
|
||
)
|
||
continue
|
||
value: Any = f"{spec.name_cn}样例结论"
|
||
if spec.key == "best_colors":
|
||
value = [
|
||
{"hex": "#A8C3A0", "name_cn": "鼠尾草绿"},
|
||
{"hex": "#D8B4A0", "name_cn": "裸粉"},
|
||
]
|
||
dimensions.append(
|
||
{
|
||
"number": spec.number,
|
||
"key": spec.key,
|
||
"value": value,
|
||
"confidence": 0.9,
|
||
"source": spec.source,
|
||
"evidence": "问卷与视觉分析一致",
|
||
"rule_hits": [],
|
||
"status": "ok",
|
||
}
|
||
)
|
||
return {
|
||
"diagnosis": {"dimensions": dimensions, "coverage": {"basic_full": 10, "total": 32}},
|
||
"copy": {
|
||
"summary": "你的整体气质偏向自然优雅，适合柔和明亮的色彩搭配。",
|
||
"sections": [
|
||
{
|
||
"title": "个人风格核心诊断",
|
||
"paragraphs": ["第一段样例文案。", "第二段样例文案。"],
|
||
},
|
||
{"title": "身材分析与穿搭指南", "paragraphs": ["第三段样例文案。"]},
|
||
],
|
||
"model": "unit-test",
|
||
"prompt_version": "v1",
|
||
"dropped_keys": [],
|
||
},
|
||
"validation": {
|
||
"passed": True,
|
||
"checks": {"pejorative": 0, "consistency": 0, "absolute": 0},
|
||
},
|
||
}
|
||
|
||
|
||
# --- 渲染器 ---------------------------------------------------------------
|
||
|
||
|
||
def test_render_produces_pdf_with_cjk_content() -> None:
|
||
pdf = render_report_pdf(_sample_result(), meta=META)
|
||
assert pdf.startswith(b"%PDF-")
|
||
# Noto CJK 子集嵌入后体积不可能只有几 KB；阈值防「渲染出空壳」退化。
|
||
assert len(pdf) > 20_000
|
||
|
||
|
||
def test_render_rejects_result_without_copy() -> None:
|
||
with pytest.raises(AppError) as excinfo:
|
||
render_report_pdf({"diagnosis": {"dimensions": []}}, meta=META)
|
||
assert excinfo.value.code is ErrorCode.STATE_CONFLICT
|
||
|
||
|
||
def test_url_fetcher_rejects_everything_including_file_scheme() -> None:
|
||
# SSRF 面：模板不允许任何外部资源，含 file:// 本地文件。
|
||
for url in ("https://evil.example/x.png", "file:///etc/passwd"):
|
||
with pytest.raises(ValueError, match="不允许加载外部资源"):
|
||
_reject_all_urls(url)
|
||
|
||
|
||
# --- PDF 元数据 AI 声明（CR-05 隐式标识，R4 N2'-①） -----------------------
|
||
|
||
|
||
def _render_template_html() -> str:
|
||
"""用渲染器同款 Jinja 环境+上下文形状渲染 HTML（与 render_report_pdf 内部一致）。"""
|
||
template = pdf_renderer._env.get_template("report.html.j2")
|
||
return template.render(
|
||
meta=META,
|
||
date_text=META.generated_at.strftime("%Y年%m月%d日"),
|
||
summary="",
|
||
sections=[],
|
||
dimensions=[],
|
||
core_dimensions=[],
|
||
ok_count=0,
|
||
)
|
||
|
||
|
||
def _decompressed_streams(pdf: bytes) -> list[bytes]:
|
||
"""解出 PDF 里全部 FlateDecode 流——WeasyPrint 把 Info 字典放进压缩对象流，
|
||
裸字节 grep 不到 /Author 等标记（zlib 为标准库，不新增依赖）。"""
|
||
out: list[bytes] = []
|
||
for m in re.finditer(rb"stream\r?\n", pdf):
|
||
start = m.end()
|
||
end = pdf.find(b"endstream", start)
|
||
if end < 0:
|
||
continue
|
||
try:
|
||
out.append(zlib.decompress(pdf[start:end]))
|
||
except zlib.error:
|
||
continue
|
||
return out
|
||
|
||
|
||
def test_report_html_declares_ai_generation_meta() -> None:
|
||
html = _render_template_html()
|
||
assert '<meta name="author" content="美搭（AI 生成内容）">' in html
|
||
assert 'name="description"' in html
|
||
# description 携带报告号/版本/日期（ReportPdfMeta 现成数据，不新增字段）
|
||
assert "报告 101 v1" in html
|
||
assert "生成于 2026年08月29日" in html
|
||
|
||
|
||
def test_rendered_pdf_carries_ai_metadata() -> None:
|
||
# WeasyPrint 把 meta author/description/generator 映射进 PDF 信息字典
|
||
# （Author/Subject/Creator）；值编码为带 BOM 的 UTF-16BE 十六进制文本串。
|
||
pdf = render_report_pdf(_sample_result(), meta=META)
|
||
joined = b"\n".join(_decompressed_streams(pdf))
|
||
for marker in (b"/Author", b"/Subject", b"/Creator"):
|
||
assert marker in joined, f"PDF 信息字典缺 {marker!r}"
|
||
author_hex = "feff" + "美搭（AI 生成内容）".encode("utf-16-be").hex()
|
||
assert author_hex.encode("ascii") in joined
|
||
desc_hex = "feff" + "本报告由人工智能生成，仅供参考。报告 101 v1".encode("utf-16-be").hex()
|
||
assert desc_hex.encode("ascii") in joined
|
||
|
||
|
||
# --- 路由接线（假服务，不连库） --------------------------------------------
|
||
|
||
|
||
class _FakeExportService:
|
||
def __init__(self) -> None:
|
||
self.create_calls: list[dict[str, Any]] = []
|
||
self.behavior: str = "ok"
|
||
self.export_file: ExportFile | None = None
|
||
|
||
async def create_export(self, **kwargs: Any) -> ExportSubmission:
|
||
self.create_calls.append(kwargs)
|
||
if self.behavior == "not_succeeded":
|
||
raise AppError(ErrorCode.STATE_CONFLICT, message="只有已完成的报告可以导出 PDF")
|
||
if self.behavior == "degraded":
|
||
raise AppError(ErrorCode.LLM_SCENE_DEGRADED)
|
||
return ExportSubmission(
|
||
task_id=987654321098765432, report_id=kwargs["report_id"], status="pending"
|
||
)
|
||
|
||
async def get_export_file(self, **kwargs: Any) -> ExportFile:
|
||
if self.export_file is None:
|
||
raise AppError(ErrorCode.REPORT_EXPORT_NOT_READY)
|
||
return self.export_file
|
||
|
||
|
||
@asynccontextmanager
|
||
async def _client(app: FastAPI) -> AsyncIterator[httpx.AsyncClient]:
|
||
"""不进 lifespan：单测不建 DB engine / LLM 网关，只测装配与信封。"""
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as http:
|
||
yield http
|
||
|
||
|
||
def _build_app(service: _FakeExportService) -> FastAPI:
|
||
app = create_app()
|
||
app.dependency_overrides[current_user_id] = lambda: 42
|
||
app.dependency_overrides[report_export_service_dep] = lambda: service
|
||
return app
|
||
|
||
|
||
_AUTH = {"Authorization": "Bearer unit-test-token"}
|
||
|
||
|
||
async def test_export_endpoint_returns_task_envelope() -> None:
|
||
service = _FakeExportService()
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.post(
|
||
"/api/v1/reports/101/export", json={"request_key": "k-1"}, headers=_AUTH
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
assert body["data"]["task_id"] == "987654321098765432"
|
||
assert body["data"]["status"] == "pending"
|
||
assert service.create_calls == [{"user_id": 42, "report_id": 101, "request_key": "k-1"}]
|
||
|
||
|
||
async def test_export_endpoint_fills_request_key_when_absent() -> None:
|
||
service = _FakeExportService()
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.post("/api/v1/reports/101/export", json={}, headers=_AUTH)
|
||
assert resp.status_code == 200
|
||
key = service.create_calls[0]["request_key"]
|
||
assert str(key).startswith("report-export:42:101:")
|
||
|
||
|
||
async def test_export_endpoint_propagates_state_conflict_as_40900() -> None:
|
||
service = _FakeExportService()
|
||
service.behavior = "not_succeeded"
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.post("/api/v1/reports/101/export", json={}, headers=_AUTH)
|
||
assert resp.status_code == 409
|
||
assert resp.json()["code"] == 40900
|
||
|
||
|
||
async def test_export_endpoint_propagates_scene_degraded_as_60006() -> None:
|
||
# 熔断（tier2 起 hd_export 被拦）经服务透传到统一信封。
|
||
service = _FakeExportService()
|
||
service.behavior = "degraded"
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.post("/api/v1/reports/101/export", json={}, headers=_AUTH)
|
||
assert resp.status_code == 503
|
||
assert resp.json()["code"] == 60006
|
||
|
||
|
||
async def test_export_requires_authentication() -> None:
|
||
async with _client(create_app()) as client:
|
||
resp = await client.post("/api/v1/reports/101/export", json={})
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
resp = await client.get("/api/v1/reports/101/export/download")
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
|
||
|
||
async def test_download_streams_pdf_with_content_disposition(tmp_path: Path) -> None:
|
||
pdf_path = tmp_path / "var" / "exports" / "report_101_v1.pdf"
|
||
pdf_path.parent.mkdir(parents=True)
|
||
pdf_path.write_bytes(b"%PDF-1.7\nunit-test-bytes")
|
||
service = _FakeExportService()
|
||
service.export_file = ExportFile(path=pdf_path, filename="美搭形象报告-v1.pdf")
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.get("/api/v1/reports/101/export/download", headers=_AUTH)
|
||
assert resp.status_code == 200
|
||
assert resp.headers["content-type"] == "application/pdf"
|
||
assert resp.content.startswith(b"%PDF-")
|
||
# 中文文件名必须走 RFC 5987 的 filename* 编码段
|
||
assert "filename*" in resp.headers["content-disposition"]
|
||
|
||
|
||
async def test_download_before_export_is_40421() -> None:
|
||
service = _FakeExportService()
|
||
async with _client(_build_app(service)) as client:
|
||
resp = await client.get("/api/v1/reports/101/export/download", headers=_AUTH)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["code"] == 40421
|