yxx
5c2441464d
实现 Export Service 完整生命周期：mistune AST → Document AST → HtmlExporter 渲染完整 HTML5，异步任务注册表 + 取消 + 24h 产物过期。新增 5 个 /api/exports 端点与 15 项测试；pdf/docx 与函数图像静态渲染留待后续 PR。
304 lines
10 KiB
Python
304 lines
10 KiB
Python
"""Export Service 的单元与端到端测试。
|
||
|
||
沿用 conftest 隔离机制：APP_DATA_DIR / DB / Vault / exports 目录都落在临时目录，
|
||
不读写真实数据。导出采用「创建即 queued + 后台 Task 执行」的异步模型，测试在同一
|
||
事件循环内创建并等待后台任务结束，得到终态 ExportJob 后再断言。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from app.config import get_settings
|
||
from app.contracts import (
|
||
ExportFormat,
|
||
ExportOptions,
|
||
ExportRequest,
|
||
ExportSource,
|
||
ExportSourceType,
|
||
ExportStatus,
|
||
)
|
||
from app.errors import ApiError
|
||
from app.export import service as export_service
|
||
from app.export.exporters.html import HtmlExporter
|
||
from app.export.markdown import parse_document
|
||
|
||
MD = """# 进程调度
|
||
|
||
一些 **加粗** 和 *斜体*，[链接](https://a.b) 与 `code`。
|
||
|
||
- 项目一
|
||
- 项目二
|
||
|
||
```python
|
||
print(1)
|
||
```
|
||
|
||
```mermaid
|
||
graph LR
|
||
```
|
||
|
||
```function_plot
|
||
y = x
|
||
```
|
||
|
||
| a | b |
|
||
|---|---|
|
||
| 1 | 2 |
|
||
|
||
行内 $x^2$ 与块级
|
||
$$
|
||
y = mx + b
|
||
$$
|
||
"""
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_export_state():
|
||
"""清空内存注册表，避免跨用例的任务/取消标志互相污染。"""
|
||
export_service._jobs.clear()
|
||
export_service._tasks.clear()
|
||
export_service._cancel_flags.clear()
|
||
yield
|
||
export_service._jobs.clear()
|
||
export_service._tasks.clear()
|
||
export_service._cancel_flags.clear()
|
||
|
||
|
||
def _create_and_wait(request: ExportRequest) -> object:
|
||
"""创建导出并在同一事件循环内等待后台任务结束，返回终态 ExportJob。"""
|
||
|
||
async def _execute():
|
||
job = await export_service.create_export(request)
|
||
return await export_service.wait_for_export(job.job_id)
|
||
|
||
return asyncio.run(_execute())
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# markdown → Document AST
|
||
# --------------------------------------------------------------------------- #
|
||
def _types(nodes) -> list[str]:
|
||
return [n.type for n in nodes]
|
||
|
||
|
||
def test_parse_document_heading_and_inline() -> None:
|
||
doc = parse_document("# 标题\n\n一段 **加粗** 和 [链接](https://a.b)。")
|
||
|
||
assert doc.type == "document"
|
||
heading = doc.children[0]
|
||
assert heading.type == "heading"
|
||
assert heading.attributes["level"] == 1
|
||
|
||
para = doc.children[1]
|
||
assert para.type == "paragraph"
|
||
kinds = _types(para.children)
|
||
assert "text" in kinds
|
||
assert "strong" in kinds
|
||
assert "link" in kinds
|
||
|
||
link = next(c for c in para.children if c.type == "link")
|
||
assert link.attributes["href"] == "https://a.b"
|
||
|
||
|
||
def test_parse_document_list_and_code_fencing() -> None:
|
||
doc = parse_document("- a\n- b\n\n```mermaid\ngraph LR\n```\n\n```function_plot\ny=x\n```\n\n```python\nx\n```")
|
||
|
||
kinds = [c.type for c in doc.children]
|
||
assert kinds[0] == "list"
|
||
assert kinds[1] == "mermaid"
|
||
assert kinds[2] == "function_plot"
|
||
assert kinds[3] == "code_block"
|
||
|
||
code = doc.children[3]
|
||
assert code.attributes["language"] == "python"
|
||
assert code.text == "x"
|
||
|
||
|
||
def test_parse_document_table_and_math() -> None:
|
||
doc = parse_document("| a | b |\n|---|---|\n| 1 | 2 |\n\n$x^2$\n\n$$\ny=mx\n$$")
|
||
|
||
table = doc.children[0]
|
||
assert table.type == "table"
|
||
assert table.children[0].type == "table_row"
|
||
assert table.children[0].children[0].attributes["head"] is True
|
||
|
||
# 表格后是「行内数学所在段落」与「块级数学」
|
||
kinds = [c.type for c in doc.children[1:]]
|
||
assert "paragraph" in kinds
|
||
assert "math_block" in kinds
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# HtmlExporter
|
||
# --------------------------------------------------------------------------- #
|
||
async def _render(markdown: str, *, title: str = "") -> str:
|
||
doc = parse_document(markdown)
|
||
doc.attributes["title"] = title
|
||
result = await HtmlExporter().export(doc, ExportOptions())
|
||
return result.content.decode("utf-8")
|
||
|
||
|
||
def test_html_exporter_renders_basic_nodes_and_escapes() -> None:
|
||
html = asyncio.run(_render("# 标题\n\n**加粗** [链接](https://a.b) 与 <b>原始</b>。"))
|
||
|
||
assert "<h1>标题</h1>" in html
|
||
assert "<strong>加粗</strong>" in html
|
||
assert '<a href="https://a.b">链接</a>' in html
|
||
# 原始 HTML 必须被转义，不能注入文档
|
||
assert "<b>原始</b>" in html
|
||
assert "<b>原始</b>" not in html
|
||
|
||
|
||
def test_html_exporter_marks_mermaid_and_function_plot() -> None:
|
||
result = asyncio.run(HtmlExporter().export(parse_document("```mermaid\ngraph LR\n```"), ExportOptions()))
|
||
|
||
html = result.content.decode("utf-8")
|
||
assert '<pre class="mermaid">graph LR</pre>' in html
|
||
assert any("mermaid" in w for w in result.warnings)
|
||
|
||
|
||
def test_html_exporter_include_title_and_metadata() -> None:
|
||
doc = parse_document("正文")
|
||
doc.attributes["title"] = "操作系统复习"
|
||
doc.attributes["metadata"] = {"tags": ["os", "复习"]}
|
||
|
||
opts = ExportOptions(include_title=True, include_metadata=True)
|
||
result = asyncio.run(HtmlExporter().export(doc, opts))
|
||
html = result.content.decode("utf-8")
|
||
|
||
assert '<h1 class="title">操作系统复习</h1>' in html
|
||
assert "os, 复习" in html
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# ExportService
|
||
# --------------------------------------------------------------------------- #
|
||
def _markdown_request(markdown: str, *, format: ExportFormat = ExportFormat.html) -> ExportRequest:
|
||
return ExportRequest(
|
||
source=ExportSource(type=ExportSourceType.markdown, markdown=markdown),
|
||
format=format,
|
||
)
|
||
|
||
|
||
def test_export_markdown_source_completes_and_writes_file() -> None:
|
||
finished = _create_and_wait(_markdown_request(MD))
|
||
|
||
assert finished.status == ExportStatus.completed
|
||
assert finished.file is not None
|
||
assert finished.file.mime_type == "text/html"
|
||
assert finished.file.size > 0
|
||
assert len(finished.file.sha256) == 64
|
||
|
||
path = get_settings().exports_path / f"{finished.job_id}.html"
|
||
assert path.exists()
|
||
content = path.read_text(encoding="utf-8")
|
||
assert "进程调度" in content
|
||
|
||
|
||
def test_export_note_source_resolves_title_and_metadata() -> None:
|
||
from app.services import note_service
|
||
|
||
async def _go():
|
||
note = await note_service.create_note(
|
||
title="操作系统复习", markdown="# 进程调度\n\n内容。", folder="导出", tags=["os"]
|
||
)
|
||
request = ExportRequest(
|
||
source=ExportSource(type=ExportSourceType.note, note_id=note.note_id),
|
||
format=ExportFormat.html,
|
||
options=ExportOptions(include_metadata=True),
|
||
)
|
||
job = await export_service.create_export(request)
|
||
return await export_service.wait_for_export(job.job_id)
|
||
|
||
finished = asyncio.run(_go())
|
||
assert finished.status == ExportStatus.completed
|
||
assert finished.file is not None
|
||
assert finished.file.file_name == "操作系统复习.html"
|
||
content = (get_settings().exports_path / f"{finished.job_id}.html").read_text(encoding="utf-8")
|
||
assert "操作系统复习" in content
|
||
assert "进程调度" in content
|
||
|
||
|
||
def test_export_pdf_unsupported() -> None:
|
||
with pytest.raises(ApiError) as exc:
|
||
asyncio.run(
|
||
export_service.create_export(_markdown_request("# x", format=ExportFormat.pdf))
|
||
)
|
||
assert exc.value.status_code == 400
|
||
assert exc.value.code == "EXPORT_FORMAT_UNSUPPORTED"
|
||
|
||
|
||
def test_export_unknown_note_404() -> None:
|
||
request = ExportRequest(
|
||
source=ExportSource(type=ExportSourceType.note, note_id="note_missing"),
|
||
format=ExportFormat.html,
|
||
)
|
||
with pytest.raises(ApiError) as exc:
|
||
asyncio.run(export_service.create_export(request))
|
||
assert exc.value.status_code == 404
|
||
assert exc.value.code == "EXPORT_SOURCE_NOT_FOUND"
|
||
|
||
|
||
def test_export_empty_markdown_invalid() -> None:
|
||
with pytest.raises(ApiError) as exc:
|
||
asyncio.run(export_service.create_export(_markdown_request(" ")))
|
||
assert exc.value.status_code == 400
|
||
assert exc.value.code == "EXPORT_OPTIONS_INVALID"
|
||
|
||
|
||
def test_export_cancel_queued_job() -> None:
|
||
async def _go():
|
||
job = await export_service.create_export(_markdown_request("# x"))
|
||
cancelled = export_service.cancel_export(job.job_id)
|
||
assert cancelled is not None
|
||
return await export_service.wait_for_export(job.job_id)
|
||
|
||
finished = asyncio.run(_go())
|
||
assert finished.status == ExportStatus.cancelled
|
||
assert finished.file is None
|
||
|
||
|
||
def test_export_file_expired_410() -> None:
|
||
async def _go():
|
||
job = await export_service.create_export(_markdown_request("# x"))
|
||
finished = await export_service.wait_for_export(job.job_id)
|
||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||
export_service._jobs[job.job_id] = finished.model_copy(
|
||
update={"file": finished.file.model_copy(update={"expires_at": past})}
|
||
)
|
||
return job.job_id
|
||
|
||
job_id = asyncio.run(_go())
|
||
with pytest.raises(ApiError) as exc:
|
||
export_service.get_export_file(job_id)
|
||
assert exc.value.status_code == 410
|
||
assert exc.value.code == "EXPORT_FILE_EXPIRED"
|
||
|
||
|
||
def test_export_list_and_get() -> None:
|
||
finished = _create_and_wait(_markdown_request("# 列表测试"))
|
||
|
||
items, total = export_service.list_exports(limit=50, offset=0)
|
||
assert total == 1
|
||
assert items[0].job_id == finished.job_id
|
||
|
||
got = export_service.get_export(finished.job_id)
|
||
assert got is not None and got.status == ExportStatus.completed
|
||
|
||
assert export_service.get_export("export_missing") is None
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 契约校验
|
||
# --------------------------------------------------------------------------- #
|
||
def test_export_source_requires_matching_field() -> None:
|
||
with pytest.raises(ValidationError):
|
||
ExportSource(type=ExportSourceType.note, note_id=None)
|
||
with pytest.raises(ValidationError):
|
||
ExportSource(type=ExportSourceType.markdown, markdown=None)
|