potaser
649e66f41d
- POST/GET /api/review/compile[/{id}[/pdf|/log]]: isolated workdir,
--untrusted, 60s timeout with process-group kill, single-slot queue
(429 QUEUE_FULL), 7-day retention sweep
- tools/install_tectonic.sh: pinned tectonic 0.17.0 (sha256) + README
- Rename menu/panel title to 学术期刊情报
- Compile UI stays a placeholder per product decision (P3 deferred)
229 lines
7.5 KiB
Python
229 lines
7.5 KiB
Python
"""tests/test_compile.py —— P3a LaTeX 编译单元/集成测试。
|
||
|
||
真实调用 tools/tectonic（宏包已预热缓存）；二进制缺失时集成用例 skipif 跳过，
|
||
校验/队列/清扫等纯逻辑用例照常运行。
|
||
运行: cd review_service && <venv>/bin/python -m pytest -q
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.config import get_settings
|
||
from app.main import app
|
||
from app.services import latex_compile as lc
|
||
|
||
MINIMAL = "\\documentclass[11pt]{article}\n\\begin{document}\nhello p3\n\\end{document}\n"
|
||
|
||
_TECTONIC_OK = get_settings().tectonic_abs.is_file()
|
||
|
||
needs_tectonic = pytest.mark.skipif(
|
||
not _TECTONIC_OK, reason="tools/tectonic 未安装（bash tools/install_tectonic.sh）"
|
||
)
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path, monkeypatch):
|
||
settings = get_settings()
|
||
monkeypatch.setattr(settings, "reformat_dir", str(tmp_path / "jobs"))
|
||
# 缓存目录不注入 tmp：用部署预热好的 data/tectonic-cache（否则每次测试重拉 bundle）
|
||
monkeypatch.setattr(settings, "compile_timeout_s", 60.0)
|
||
monkeypatch.setattr(lc, "_pending", 0)
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
# ================================================================ 校验（无需 tectonic）
|
||
|
||
def test_validation_no_input(client: TestClient):
|
||
r = client.post("/api/review/compile", data={})
|
||
body = r.json()
|
||
assert r.status_code == 200 and body["ok"] is False
|
||
assert body["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_validation_bad_job_id(client: TestClient):
|
||
r = client.post("/api/review/compile", data={"job_id": "../evil"})
|
||
assert r.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_validation_job_not_found(client: TestClient):
|
||
r = client.post("/api/review/compile", data={"job_id": "0123456789ab"})
|
||
assert r.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_validation_path_traversal_filename(client: TestClient):
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("../evil.tex", b"x", "application/x-tex")},
|
||
)
|
||
assert r.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_validation_non_tex(client: TestClient):
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("paper.pdf", b"x", "application/pdf")},
|
||
)
|
||
assert r.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_validation_oversize(client: TestClient, monkeypatch):
|
||
monkeypatch.setattr(get_settings(), "max_tex_bytes", 10)
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("paper.tex", b"01234567890", "application/x-tex")},
|
||
)
|
||
assert r.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
# ================================================================ 编译（真实 tectonic）
|
||
|
||
@needs_tectonic
|
||
def test_compile_happy_path(client: TestClient):
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("paper.tex", MINIMAL.encode(), "application/x-tex")},
|
||
)
|
||
body = r.json()
|
||
assert r.status_code == 200 and body["ok"] is True, body
|
||
cid = body["compile_id"]
|
||
assert len(cid) == 12
|
||
|
||
pdf = client.get(body["pdf_url"])
|
||
assert pdf.status_code == 200
|
||
assert pdf.headers["content-type"].startswith("application/pdf")
|
||
assert pdf.content[:5] == b"%PDF-"
|
||
|
||
log = client.get(body["log_url"])
|
||
assert log.status_code == 200
|
||
assert "LaTeX2e" in log.text and "Output written" in log.text
|
||
|
||
meta = client.get(f"/api/review/compile/{cid}")
|
||
assert meta.json()["ok"] is True
|
||
assert meta.json()["meta"]["pdf_bytes"] > 0
|
||
|
||
|
||
@needs_tectonic
|
||
def test_compile_by_job_id(client: TestClient):
|
||
tex = (
|
||
"\\documentclass[11pt]{article}\n\\begin{document}\nvia job\n\\end{document}\n"
|
||
)
|
||
rf = client.post(
|
||
"/api/review/reformat",
|
||
files={"file": ("paper.tex", tex.encode(), "application/x-tex")},
|
||
data={"target_journal": "default"},
|
||
).json()
|
||
assert rf["ok"] is True
|
||
r = client.post("/api/review/compile", data={"job_id": rf["job_id"]})
|
||
body = r.json()
|
||
assert body["ok"] is True and body["compile_id"] == rf["job_id"]
|
||
pdf = client.get(body["pdf_url"])
|
||
assert pdf.content[:5] == b"%PDF-"
|
||
|
||
|
||
@needs_tectonic
|
||
def test_compile_malicious_write18(client: TestClient, tmp_path):
|
||
canary = tmp_path / "canary_p3"
|
||
evil = (
|
||
"\\documentclass{article}\n"
|
||
f"\\write18{{touch {canary}}}\n"
|
||
"\\begin{document}\nno shell please\n\\end{document}\n"
|
||
)
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("evil.tex", evil.encode(), "application/x-tex")},
|
||
)
|
||
body = r.json()
|
||
# 系统无损即通过（信封 ok 与否均可）：canary 绝不能出现
|
||
assert not canary.exists(), r"\write18 竟然执行了！"
|
||
if body["ok"]:
|
||
pdf = client.get(body["pdf_url"])
|
||
assert pdf.content[:5] == b"%PDF-"
|
||
else:
|
||
assert body["error"]["code"] in ("COMPILE_FAILED", "TIMEOUT")
|
||
|
||
|
||
@needs_tectonic
|
||
def test_compile_failed_returns_log_tail(client: TestClient):
|
||
bad = "\\documentclass{article}\n\\begin{document}\n\\undefinedcommandxyz\n"
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("bad.tex", bad.encode(), "application/x-tex")},
|
||
)
|
||
body = r.json()
|
||
assert body["ok"] is False
|
||
assert body["error"]["code"] == "COMPILE_FAILED"
|
||
assert "log" in body["error"]["message"].lower() or body["error"].get("detail")
|
||
|
||
|
||
@needs_tectonic
|
||
def test_compile_timeout(client: TestClient, monkeypatch):
|
||
monkeypatch.setattr(get_settings(), "compile_timeout_s", 5.0)
|
||
loop_tex = (
|
||
"\\documentclass{article}\n\\begin{document}\n"
|
||
"\\count255=0\n\\loop\\iftrue\\advance\\count255 by 1\\repeat\n"
|
||
"\\end{document}\n"
|
||
)
|
||
started = time.monotonic()
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("loop.tex", loop_tex.encode(), "application/x-tex")},
|
||
)
|
||
elapsed = time.monotonic() - started
|
||
body = r.json()
|
||
assert body["ok"] is False
|
||
assert body["error"]["code"] == "TIMEOUT"
|
||
assert elapsed < 15.0, f"超时回收过慢：{elapsed:.1f}s"
|
||
|
||
|
||
# ================================================================ 队列 / 清扫
|
||
|
||
def test_queue_full(client: TestClient, monkeypatch):
|
||
monkeypatch.setattr(lc, "_pending", 5)
|
||
monkeypatch.setattr(get_settings(), "compile_queue_limit", 5)
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("paper.tex", MINIMAL.encode(), "application/x-tex")},
|
||
)
|
||
assert r.status_code == 429
|
||
body = r.json()
|
||
assert body["ok"] is False and body["error"]["code"] == "QUEUE_FULL"
|
||
|
||
|
||
def test_tool_unavailable(client: TestClient, monkeypatch):
|
||
monkeypatch.setattr(get_settings(), "tectonic_path", "tools/definitely-missing")
|
||
r = client.post(
|
||
"/api/review/compile",
|
||
files={"file": ("paper.tex", MINIMAL.encode(), "application/x-tex")},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["error"]["code"] == "TOOL_UNAVAILABLE"
|
||
assert "install_tectonic" in r.json()["error"]["message"]
|
||
|
||
|
||
def test_sweep_expired(tmp_path, monkeypatch):
|
||
settings = get_settings()
|
||
monkeypatch.setattr(settings, "reformat_dir", str(tmp_path / "jobs"))
|
||
base = tmp_path / "jobs"
|
||
stale = base / "aabbccddeeff"
|
||
stale.mkdir(parents=True)
|
||
(stale / "output.pdf").write_bytes(b"%PDF-fake")
|
||
old = time.time() - 8 * 86400
|
||
os.utime(stale, (old, old))
|
||
|
||
fresh = base / "112233445566"
|
||
fresh.mkdir()
|
||
noise = base / "not-a-job"
|
||
noise.mkdir()
|
||
|
||
removed = lc.sweep_expired()
|
||
assert removed == 1
|
||
assert not stale.exists()
|
||
assert fresh.exists()
|
||
assert noise.exists()
|