"""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 = ( '' 'Ch1' '' '

Hello PDF

Test paragraph for conversion.

' '

pic

' ) opf = ( '' '' '' 'Turn:uuid:t' 'en' '' '' '' '' '' '' ) ncx = ( '' '' '' 'T' 'Ch1' '' ) 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", '' '', 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