zikai
729c77e98b
- pdf.html 移除页脚"管理页"链接：管理页入口对普通用户隐藏， 仅管理员知晓 /pdf-admin 直达路径（权限同 /files 文件浏览页， 复用 require_docs_auth Basic Auth） - test_pdf_service.py 同步测试凭据为 a/66511315 注：实际凭据存放于 config.yaml 的 docs 段（git-ignored，仅本机 root 持有）， 本次仅隐藏入口；管理页鉴权机制（require_docs_auth）此前已与 文件浏览页 /files 完全一致，无需改动。
285 lines
12 KiB
Python
285 lines
12 KiB
Python
"""PDF 转换服务测试（pytest + TestClient）。
|
||
|
||
用 SQLite 内存库覆盖 get_db，免依赖 MySQL；用临时目录覆盖上传根目录。
|
||
覆盖完整链路：上传 epub -> 轮询至 done -> 下载有效 PDF -> 用户软删 ->
|
||
管理页仍可见 -> 管理员硬删 -> 真正删除。另测大小超限与格式校验。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import time
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
from app import config as config_module
|
||
from app import database as db_module
|
||
from app.database import Base, get_db
|
||
from app.models.uploaded_file import UploadedFile # noqa: F401 注册映射
|
||
|
||
|
||
# ---------------- fixtures ----------------
|
||
|
||
|
||
def _make_epub_bytes() -> bytes:
|
||
"""构造一个最小的合法 epub（含 1 章节 + 1 PNG 图片 + CSS）。"""
|
||
from PIL import Image # 已是 weasyprint 依赖间接项，环境可用
|
||
|
||
png = io.BytesIO()
|
||
Image.new("RGBA", (8, 8), (26, 95, 180, 255)).save(png, format="PNG")
|
||
png_bytes = png.getvalue()
|
||
|
||
ch = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<html xmlns="http://www.w3.org/1999/xhtml"><head><title>Ch1</title>'
|
||
'<link rel="stylesheet" type="text/css" href="style.css"/></head>'
|
||
'<body><h1>Hello PDF</h1><p>Test paragraph for conversion.</p>'
|
||
'<p><img src="img.png" alt="pic"/></p></body></html>'
|
||
)
|
||
opf = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<package xmlns="http://www.idpf.org/2007/opf" version="2.0" unique-identifier="Bid">'
|
||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
||
'<dc:title>T</dc:title><dc:identifier id="Bid">urn:uuid:t</dc:identifier>'
|
||
'<dc:language>en</dc:language></metadata>'
|
||
'<manifest>'
|
||
'<item id="ncx" href="toc.ncx" media-type="application/x-dtbncx+xml"/>'
|
||
'<item id="css" href="style.css" media-type="text/css"/>'
|
||
'<item id="img" href="img.png" media-type="image/png"/>'
|
||
'<item id="ch1" href="ch1.xhtml" media-type="application/xhtml+xml"/>'
|
||
'</manifest><spine toc="ncx"><itemref idref="ch1"/></spine></package>'
|
||
)
|
||
ncx = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<ncx xmlns="http://www.daisy.org/z3986/2005/ncx/" version="2005-1">'
|
||
'<head><meta name="dtb:uid" content="urn:uuid:t"/></head>'
|
||
'<docTitle><text>T</text></docTitle>'
|
||
'<navMap><navPoint id="n1" playOrder="1"><navLabel><text>Ch1</text></navLabel>'
|
||
'<content src="ch1.xhtml"/></navPoint></navMap></ncx>'
|
||
)
|
||
css = "h1{color:#1a5fb4} p{line-height:1.6}"
|
||
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as z:
|
||
z.writestr("mimetype", "application/epub+zip", compress_type=zipfile.ZIP_STORED)
|
||
z.writestr("OEBPS/content.opf", opf, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/toc.ncx", ncx, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/style.css", css, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/img.png", png_bytes, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/ch1.xhtml", ch, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("META-INF/container.xml",
|
||
'<?xml version="1.0"?><container version="1.0" '
|
||
'xmlns="urn:oasis:names:tc:opendocument:xmlns:container">'
|
||
'<rootfiles><rootfile full-path="OEBPS/content.opf" '
|
||
'media-type="application/oebps-package+xml"/></rootfiles></container>',
|
||
compress_type=zipfile.ZIP_DEFLATED)
|
||
return buf.getvalue()
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path, monkeypatch):
|
||
"""构造用隔离 MySQL 测试库 + 临时上传目录的 TestClient。
|
||
|
||
用独立库 zikai_filesvc_test（与生产库隔离），每个 fixture 重建表，准确测试生产
|
||
行为（MySQL BigInteger 自增、线程安全连接）。后台转换经 asyncio.to_thread 在
|
||
独立线程执行，MySQL 连接池天然支持跨线程。
|
||
"""
|
||
real_settings = config_module.get_settings()
|
||
# 用独立测试库 zikai_filesvc_test（与生产库隔离），准确测试生产行为
|
||
test_settings = real_settings.model_copy(deep=True)
|
||
test_settings.database.database = "zikai_filesvc_test"
|
||
test_settings.pdf.convert_timeout_seconds = 60
|
||
|
||
# 全局 get_settings 被 lru_cache；config 与 database 两个模块各自 import 了它，
|
||
# 都需 patch，使后台线程（经 database.get_session_local）也读到测试库
|
||
monkeypatch.setattr(config_module, "get_settings", lambda: test_settings)
|
||
monkeypatch.setattr(db_module, "get_settings", lambda: test_settings)
|
||
# 清掉已建的全局引擎/SessionLocal，下次 get_session_local() 用测试库重建
|
||
db_module.dispose_engine()
|
||
|
||
upload_dir = tmp_path / "uploads"
|
||
upload_dir.mkdir()
|
||
|
||
# 复用全局 SessionLocal（指向测试库），保证请求路径与后台线程用同一库
|
||
SessionLocal = db_module.get_session_local()
|
||
|
||
# 覆盖 get_db（仍用全局 SessionLocal，但确保请求结束关闭）
|
||
def override_get_db():
|
||
db = SessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
# 上传目录指向临时目录
|
||
monkeypatch.setattr(test_settings, "storage", test_settings.storage)
|
||
test_settings.storage.upload_dir = str(upload_dir)
|
||
# upload_service / pdf_service 内部 import 了 get_settings，需 patch 其引用
|
||
import app.services.upload_service as us
|
||
import app.services.pdf_service as ps
|
||
import app.services.pdf_converter as pc # noqa: F401
|
||
monkeypatch.setattr(us, "get_settings", lambda: test_settings)
|
||
monkeypatch.setattr(ps, "get_settings", lambda: test_settings)
|
||
|
||
# 每次测试前清空重建表，保证隔离
|
||
Base.metadata.drop_all(bind=db_module.get_engine())
|
||
Base.metadata.create_all(bind=db_module.get_engine())
|
||
|
||
from app.main import app
|
||
app.dependency_overrides[get_db] = override_get_db
|
||
with TestClient(app) as c:
|
||
yield c
|
||
app.dependency_overrides.clear()
|
||
Base.metadata.drop_all(bind=db_module.get_engine())
|
||
db_module.dispose_engine()
|
||
|
||
|
||
def _wait_done(client, job_id, cookie, timeout=60):
|
||
"""轮询任务状态直到 done/failed 或超时。"""
|
||
deadline = time.time() + timeout
|
||
last = None
|
||
while time.time() < deadline:
|
||
r = client.get(f"/api/pdf/jobs/{job_id}", cookies=cookie)
|
||
assert r.status_code == 200, r.text
|
||
last = r.json()
|
||
if last["status"] in ("done", "failed"):
|
||
return last
|
||
time.sleep(0.3)
|
||
raise AssertionError(f"任务未在 {timeout}s 内完成，最后状态: {last}")
|
||
|
||
|
||
# 管理页 Basic Auth 凭据（与 config.yaml 的 docs 段保持一致）
|
||
ADMIN_AUTH = ("a", "66511315")
|
||
|
||
|
||
# ---------------- 测试 ----------------
|
||
|
||
|
||
class TestSubmitAndConvert:
|
||
def test_upload_epub_converts_and_downloads(self, client):
|
||
epub = _make_epub_bytes()
|
||
# 首次上传：无 cookie，应下发新 cookie
|
||
r = client.post(
|
||
"/api/pdf/jobs",
|
||
files={"file": ("test.epub", epub, "application/epub+zip")},
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["set_cookie"] is True
|
||
assert "zk_pdf" in r.headers.get("set-cookie", "")
|
||
job = body["job"]
|
||
assert job["status"] == "pending"
|
||
job_id = job["id"]
|
||
cookie = {"zk_pdf": r.cookies.get("zk_pdf")}
|
||
|
||
# 轮询至完成
|
||
final = _wait_done(client, job_id, cookie)
|
||
assert final["status"] == "done", final
|
||
assert final["progress"] == 100
|
||
|
||
# 下载产物：应为有效 PDF
|
||
d = client.get(f"/api/pdf/jobs/{job_id}/download", cookies=cookie)
|
||
assert d.status_code == 200, d.text
|
||
assert d.headers["content-type"] == "application/pdf"
|
||
assert d.content[:5] == b"%PDF-"
|
||
assert len(d.content) > 100
|
||
|
||
def test_user_list_shows_own_jobs(self, client):
|
||
epub = _make_epub_bytes()
|
||
r = client.post("/api/pdf/jobs", files={"file": ("a.epub", epub, "application/epub+zip")})
|
||
cookie = {"zk_pdf": r.cookies.get("zk_pdf")}
|
||
lst = client.get("/api/pdf/jobs", cookies=cookie)
|
||
assert lst.status_code == 200
|
||
assert lst.json()["total"] == 1
|
||
|
||
# 另一用户（清空 client cookie jar 模拟全新浏览器，应下发新 cookie）
|
||
client.cookies.clear()
|
||
r2 = client.post("/api/pdf/jobs", files={"file": ("b.epub", epub, "application/epub+zip")})
|
||
cookie2 = {"zk_pdf": r2.cookies.get("zk_pdf")}
|
||
assert cookie2["zk_pdf"] != cookie["zk_pdf"] # 确是不同用户
|
||
lst2 = client.get("/api/pdf/jobs", cookies=cookie2)
|
||
assert lst2.json()["total"] == 1 # 只有自己的 1 个
|
||
|
||
def test_rejects_non_epub(self, client):
|
||
r = client.post(
|
||
"/api/pdf/jobs",
|
||
files={"file": ("note.txt", b"hello world", "text/plain")},
|
||
)
|
||
assert r.status_code == 400
|
||
|
||
def test_rejects_oversize(self, client, monkeypatch):
|
||
# 把上限调到极小，避免真的构造 250MB 文件
|
||
import app.services.pdf_service as ps
|
||
real = ps.get_settings()
|
||
small = real.model_copy(deep=True)
|
||
small.pdf.max_size_bytes = 100
|
||
monkeypatch.setattr(ps, "get_settings", lambda: small)
|
||
monkeypatch.setattr(ps, "get_settings", lambda: small)
|
||
# controller 的 _resolve_cookie 也读 get_settings，但 submit 内 service 用 ps.get_settings
|
||
epub = _make_epub_bytes()
|
||
r = client.post("/api/pdf/jobs", files={"file": ("big.epub", epub, "application/epub+zip")})
|
||
assert r.status_code == 413
|
||
|
||
|
||
class TestDeleteSemantics:
|
||
def test_user_soft_delete_admin_still_visible_then_admin_hard_delete(self, client):
|
||
epub = _make_epub_bytes()
|
||
r = client.post("/api/pdf/jobs", files={"file": ("del.epub", epub, "application/epub+zip")})
|
||
cookie = {"zk_pdf": r.cookies.get("zk_pdf")}
|
||
job_id = r.json()["job"]["id"]
|
||
_wait_done(client, job_id, cookie)
|
||
|
||
# 用户软删
|
||
d = client.delete(f"/api/pdf/jobs/{job_id}", cookies=cookie)
|
||
assert d.status_code == 200
|
||
assert d.json()["deleted"] is True
|
||
|
||
# 用户列表不再可见
|
||
lst = client.get("/api/pdf/jobs", cookies=cookie)
|
||
assert lst.json()["total"] == 0
|
||
|
||
# 管理页仍可见且标注已删除
|
||
al = client.get("/api/admin/pdf/jobs", auth=ADMIN_AUTH)
|
||
assert al.status_code == 200
|
||
items = al.json()["items"]
|
||
assert any(it["id"] == job_id and it["user_deleted"] is True for it in items)
|
||
assert any(it["id"] == job_id and it["deleted_at"] is not None for it in items)
|
||
|
||
# 用户已无法下载（已软删）
|
||
dl = client.get(f"/api/pdf/jobs/{job_id}/download", cookies=cookie)
|
||
assert dl.status_code == 404
|
||
|
||
# 管理员硬删
|
||
ad = client.delete(f"/api/admin/pdf/jobs/{job_id}", auth=ADMIN_AUTH)
|
||
assert ad.status_code == 200
|
||
assert ad.json()["deleted"] is True
|
||
|
||
# 管理页也不再可见
|
||
al2 = client.get("/api/admin/pdf/jobs", auth=ADMIN_AUTH)
|
||
assert not any(it["id"] == job_id for it in al2.json()["items"])
|
||
|
||
def test_admin_requires_auth(self, client):
|
||
r = client.get("/api/admin/pdf/jobs")
|
||
assert r.status_code == 401
|
||
r = client.get("/api/admin/pdf/jobs", auth=("a", "wrong"))
|
||
assert r.status_code == 401
|
||
r = client.get("/api/admin/pdf/jobs", auth=ADMIN_AUTH)
|
||
assert r.status_code == 200
|
||
|
||
def test_user_cannot_access_others_job(self, client):
|
||
epub = _make_epub_bytes()
|
||
r1 = client.post("/api/pdf/jobs", files={"file": ("x.epub", epub, "application/epub+zip")})
|
||
job_id = r1.json()["job"]["id"]
|
||
# 另一用户访问（清空 cookie jar 模拟全新浏览器）
|
||
client.cookies.clear()
|
||
r2 = client.post("/api/pdf/jobs", files={"file": ("y.epub", epub, "application/epub+zip")})
|
||
cookie2 = {"zk_pdf": r2.cookies.get("zk_pdf")}
|
||
assert client.get(f"/api/pdf/jobs/{job_id}", cookies=cookie2).status_code == 404
|
||
assert client.get(f"/api/pdf/jobs/{job_id}/download", cookies=cookie2).status_code == 404
|
||
assert client.delete(f"/api/pdf/jobs/{job_id}", cookies=cookie2).status_code == 404
|