wehub-resource-sync
e4dcfc49aa
Tests / Lint and Format (push) Waiting to run
Tests / Web Node Tests (push) Waiting to run
Tests / Import Check (Python 3.11) (push) Waiting to run
Tests / Import Check (Python 3.12) (push) Waiting to run
Tests / Import Check (Python 3.13) (push) Waiting to run
Tests / Import Check (Python 3.14) (push) Waiting to run
Tests / Python Tests (Python 3.11) (push) Blocked by required conditions
Tests / Python Tests (Python 3.12) (push) Blocked by required conditions
Tests / Python Tests (Python 3.13) (push) Blocked by required conditions
Tests / Python Tests (Python 3.14) (push) Blocked by required conditions
Tests / Test Summary (push) Blocked by required conditions
441 lines
14 KiB
Python
441 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import zipfile
|
|
|
|
import httpx as real_httpx
|
|
import pytest
|
|
|
|
from deeptutor.services.parsing.engines.mineru import backend as mineru_backend
|
|
from deeptutor.services.parsing.engines.mineru import cloud as mineru_cloud
|
|
from deeptutor.services.parsing.engines.mineru import config as mineru_config
|
|
from deeptutor.services.parsing.engines.mineru.config import MinerUConfig, MinerUError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resolve_mineru_config_reads_settings(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
mineru_config,
|
|
"load_mineru_settings",
|
|
lambda: {
|
|
"mode": "cloud",
|
|
"api_token": "tok",
|
|
"model_version": "vlm",
|
|
"language": "ch",
|
|
"api_base_url": "https://x",
|
|
"enable_formula": False,
|
|
"enable_table": True,
|
|
"is_ocr": True,
|
|
},
|
|
)
|
|
cfg = mineru_config.resolve_mineru_config()
|
|
assert cfg.is_cloud and not cfg.is_local
|
|
assert cfg.api_token == "tok"
|
|
assert cfg.model_version == "vlm"
|
|
assert cfg.api_language == "ch"
|
|
|
|
monkeypatch.setattr(
|
|
mineru_config, "load_mineru_settings", lambda: {"mode": "local", "language": "auto"}
|
|
)
|
|
auto_cfg = mineru_config.resolve_mineru_config()
|
|
assert auto_cfg.is_local
|
|
assert auto_cfg.api_language is None # "auto" → omit the language hint
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backend dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_pdf_to_workdir_dispatches_local(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import local as pdf_parser
|
|
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
out = tmp_path / "out"
|
|
|
|
def fake_local(p: str, base: str, **kwargs) -> bool: # noqa: ANN003
|
|
(Path(base) / Path(p).stem).mkdir(parents=True, exist_ok=True)
|
|
return True
|
|
|
|
monkeypatch.setattr(pdf_parser, "parse_pdf_with_mineru", fake_local)
|
|
|
|
workdir = mineru_backend.parse_pdf_to_workdir(pdf, out, config=MinerUConfig(mode="local"))
|
|
assert workdir == out / "exam"
|
|
|
|
|
|
def test_parse_pdf_to_workdir_local_failure_raises(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import local as pdf_parser
|
|
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
monkeypatch.setattr(pdf_parser, "parse_pdf_with_mineru", lambda *a, **k: False)
|
|
|
|
with pytest.raises(MinerUError):
|
|
mineru_backend.parse_pdf_to_workdir(
|
|
pdf, tmp_path / "out", config=MinerUConfig(mode="local")
|
|
)
|
|
|
|
|
|
def test_parse_pdf_to_workdir_dispatches_cloud(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
called: dict[str, bool] = {}
|
|
|
|
def fake_cloud(p, base, cfg, **kwargs): # noqa: ANN001, ANN003
|
|
called["yes"] = True
|
|
target = Path(base) / "clouddir"
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
monkeypatch.setattr(mineru_cloud, "parse_cloud", fake_cloud)
|
|
|
|
workdir = mineru_backend.parse_pdf_to_workdir(
|
|
pdf, tmp_path / "out", config=MinerUConfig(mode="cloud", api_token="t")
|
|
)
|
|
assert called.get("yes") is True
|
|
assert workdir.name == "clouddir"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cloud client internals
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_verify_credentials_requires_token() -> None:
|
|
with pytest.raises(MinerUError):
|
|
mineru_cloud.verify_credentials(MinerUConfig(mode="cloud", api_token=""))
|
|
|
|
|
|
def test_local_cli_probe_reports_path(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
mineru_backend.shutil,
|
|
"which",
|
|
lambda cmd: "/opt/env/bin/mineru" if cmd == "mineru" else None,
|
|
)
|
|
probe = mineru_backend.local_cli_probe()
|
|
assert probe == {
|
|
"found": True,
|
|
"command": "mineru",
|
|
"path": "/opt/env/bin/mineru",
|
|
"source": "path",
|
|
}
|
|
|
|
monkeypatch.setattr(mineru_backend.shutil, "which", lambda cmd: None)
|
|
assert mineru_backend.local_cli_probe()["found"] is False
|
|
|
|
|
|
def test_local_cli_probe_configured_path_takes_precedence(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# Even with a PATH hit available, an explicit configured path wins.
|
|
monkeypatch.setattr(mineru_backend.shutil, "which", lambda cmd: "/usr/bin/magic-pdf")
|
|
|
|
exe = tmp_path / "mineru"
|
|
exe.write_text("#!/bin/sh\n", encoding="utf-8")
|
|
exe.chmod(0o755)
|
|
probe = mineru_backend.local_cli_probe(str(exe))
|
|
assert probe == {
|
|
"found": True,
|
|
"command": "mineru",
|
|
"path": str(exe),
|
|
"source": "configured",
|
|
}
|
|
|
|
# A configured path that isn't an executable file reports not-found
|
|
# (no silent fallback to PATH — the admin asked for this exact binary).
|
|
probe = mineru_backend.local_cli_probe(str(tmp_path / "missing"))
|
|
assert probe["found"] is False
|
|
assert probe["source"] == "configured"
|
|
|
|
|
|
def test_parse_local_rejects_bad_configured_path(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
with pytest.raises(MinerUError) as exc:
|
|
mineru_backend.parse_pdf_to_workdir(
|
|
pdf,
|
|
tmp_path / "out",
|
|
config=MinerUConfig(mode="local", local_cli_path=str(tmp_path / "nope")),
|
|
)
|
|
assert "not an executable" in str(exc.value)
|
|
|
|
|
|
def test_local_cli_version_rejects_unknown_command() -> None:
|
|
# Whitelist guard: bare names outside the whitelist (and non-executable
|
|
# paths) never reach subprocess.
|
|
assert mineru_backend.local_cli_version("definitely-not-a-cli") == ""
|
|
|
|
|
|
def test_pdf_parser_streams_output_lines(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from deeptutor.services.parsing.engines.mineru import local as pdf_parser
|
|
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
out = tmp_path / "out"
|
|
|
|
class FakeProcess:
|
|
def __init__(self, cmd, **kwargs): # noqa: ANN002, ANN003
|
|
# Simulate the CLI writing its artifacts under the -o directory.
|
|
target = Path(cmd[cmd.index("-o") + 1]) / "exam"
|
|
(target / "auto").mkdir(parents=True, exist_ok=True)
|
|
(target / "auto" / "exam.md").write_text("# parsed", encoding="utf-8")
|
|
self.stdout = iter(["downloading model...\n", "\n", "parsing page 1/3\n"])
|
|
|
|
def wait(self):
|
|
return 0
|
|
|
|
monkeypatch.setattr(pdf_parser, "check_mineru_installed", lambda: "mineru")
|
|
monkeypatch.setattr(pdf_parser.subprocess, "Popen", FakeProcess)
|
|
|
|
seen: list[str] = []
|
|
ok = pdf_parser.parse_pdf_with_mineru(str(pdf), str(out), on_output=seen.append)
|
|
|
|
assert ok is True
|
|
# Throttled, but the first line always gets through; blanks are dropped.
|
|
assert seen and seen[0] == "downloading model..."
|
|
assert (out / "exam" / "auto" / "exam.md").exists()
|
|
|
|
|
|
def test_parse_cloud_reports_progress(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
|
|
poll_count = {"n": 0}
|
|
|
|
def poll(path): # noqa: ANN001
|
|
poll_count["n"] += 1
|
|
if poll_count["n"] == 1:
|
|
return _Resp(
|
|
{
|
|
"code": 0,
|
|
"data": {
|
|
"extract_result": [
|
|
{
|
|
"file_name": "exam.pdf",
|
|
"state": "running",
|
|
"extract_progress": {"extracted_pages": 1, "total_pages": 3},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
return _Resp(
|
|
{
|
|
"code": 0,
|
|
"data": {
|
|
"extract_result": [
|
|
{"file_name": "exam.pdf", "state": "done", "full_zip_url": "https://zip"}
|
|
]
|
|
},
|
|
}
|
|
)
|
|
|
|
_install_fake_httpx(
|
|
monkeypatch,
|
|
submit=lambda path, body: _Resp(
|
|
{"code": 0, "data": {"batch_id": "B", "file_urls": ["https://up"]}}
|
|
),
|
|
poll=poll,
|
|
download=lambda url: _Resp(content=_zip_bytes(), status=200),
|
|
)
|
|
|
|
reports: list[str] = []
|
|
mineru_cloud.parse_cloud(
|
|
pdf,
|
|
tmp_path / "out",
|
|
MinerUConfig(mode="cloud", api_token="tok"),
|
|
poll_interval=0,
|
|
timeout=10,
|
|
on_progress=reports.append,
|
|
)
|
|
assert any("1/3 pages" in r for r in reports)
|
|
assert any("done" in r for r in reports)
|
|
|
|
|
|
def test_match_entry_prefers_filename_then_first() -> None:
|
|
rows = [
|
|
{"file_name": "a.pdf", "state": "running"},
|
|
{"file_name": "b.pdf", "state": "done"},
|
|
]
|
|
assert mineru_cloud._match_entry(rows, "b.pdf")["state"] == "done"
|
|
assert mineru_cloud._match_entry(rows, "zzz.pdf")["file_name"] == "a.pdf"
|
|
assert mineru_cloud._match_entry([], "x.pdf") is None
|
|
|
|
|
|
def test_extract_archive_rejects_zip_slip(tmp_path: Path) -> None:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as archive:
|
|
archive.writestr("../evil.txt", "pwned")
|
|
archive.writestr("safe.md", "ok")
|
|
archive.writestr("images/fig.png", b"img")
|
|
workdir = tmp_path / "wd"
|
|
|
|
mineru_cloud._extract_archive(buf.getvalue(), workdir)
|
|
|
|
assert (workdir / "safe.md").exists()
|
|
assert (workdir / "images" / "fig.png").exists()
|
|
# The traversal member must not escape the working dir.
|
|
assert not (tmp_path / "evil.txt").exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cloud client end-to-end (mocked transport)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _zip_bytes() -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as archive:
|
|
archive.writestr("full.md", "# Exam\n1. Question one")
|
|
archive.writestr("exam_content_list.json", json.dumps([{"type": "text", "text": "Q1"}]))
|
|
archive.writestr("images/fig.png", b"img")
|
|
return buf.getvalue()
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, json_data=None, content=b"", status=200): # noqa: ANN001
|
|
self._json = json_data
|
|
self.content = content
|
|
self.status_code = status
|
|
|
|
def json(self):
|
|
return self._json
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
request = real_httpx.Request("GET", "http://x")
|
|
response = real_httpx.Response(self.status_code, request=request)
|
|
raise real_httpx.HTTPStatusError("err", request=request, response=response)
|
|
|
|
|
|
def _install_fake_httpx(monkeypatch: pytest.MonkeyPatch, *, submit, poll, download) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
class FakeClient:
|
|
def __init__(self, *a, **k): # noqa: ANN002, ANN003
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a): # noqa: ANN002
|
|
return False
|
|
|
|
def post(self, path, json=None, timeout=None): # noqa: A002, ANN001
|
|
return submit(path, json)
|
|
|
|
def get(self, path, timeout=None): # noqa: ANN001
|
|
return poll(path)
|
|
|
|
fake = SimpleNamespace(
|
|
Client=FakeClient,
|
|
put=lambda url, content=None, timeout=None: _Resp(status=200),
|
|
get=lambda url, timeout=None, follow_redirects=False: download(url),
|
|
HTTPError=real_httpx.HTTPError,
|
|
HTTPStatusError=real_httpx.HTTPStatusError,
|
|
)
|
|
monkeypatch.setattr(mineru_cloud, "httpx", fake)
|
|
|
|
|
|
def test_parse_cloud_happy_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4 test")
|
|
|
|
_install_fake_httpx(
|
|
monkeypatch,
|
|
submit=lambda path, body: _Resp(
|
|
{"code": 0, "data": {"batch_id": "B", "file_urls": ["https://up"]}}
|
|
),
|
|
poll=lambda path: _Resp(
|
|
{
|
|
"code": 0,
|
|
"data": {
|
|
"extract_result": [
|
|
{"file_name": "exam.pdf", "state": "done", "full_zip_url": "https://zip"}
|
|
]
|
|
},
|
|
}
|
|
),
|
|
download=lambda url: _Resp(content=_zip_bytes(), status=200),
|
|
)
|
|
|
|
workdir = mineru_cloud.parse_cloud(
|
|
pdf,
|
|
tmp_path / "out",
|
|
MinerUConfig(mode="cloud", api_token="tok"),
|
|
poll_interval=0,
|
|
timeout=10,
|
|
)
|
|
assert (workdir / "full.md").exists()
|
|
assert (workdir / "exam_content_list.json").exists()
|
|
assert (workdir / "images" / "fig.png").exists()
|
|
|
|
|
|
def test_parse_cloud_surfaces_api_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
|
|
_install_fake_httpx(
|
|
monkeypatch,
|
|
submit=lambda path, body: _Resp({"code": 401, "msg": "bad token"}),
|
|
poll=lambda path: _Resp({"code": 0, "data": {"extract_result": []}}),
|
|
download=lambda url: _Resp(content=b"", status=200),
|
|
)
|
|
|
|
with pytest.raises(MinerUError) as exc:
|
|
mineru_cloud.parse_cloud(
|
|
pdf,
|
|
tmp_path / "out",
|
|
MinerUConfig(mode="cloud", api_token="bad"),
|
|
poll_interval=0,
|
|
timeout=10,
|
|
)
|
|
assert "401" in str(exc.value)
|
|
|
|
|
|
def test_parse_cloud_failed_state_raises(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF-1.4")
|
|
|
|
_install_fake_httpx(
|
|
monkeypatch,
|
|
submit=lambda path, body: _Resp(
|
|
{"code": 0, "data": {"batch_id": "B", "file_urls": ["https://up"]}}
|
|
),
|
|
poll=lambda path: _Resp(
|
|
{
|
|
"code": 0,
|
|
"data": {
|
|
"extract_result": [
|
|
{"file_name": "exam.pdf", "state": "failed", "err_msg": "corrupt pdf"}
|
|
]
|
|
},
|
|
}
|
|
),
|
|
download=lambda url: _Resp(content=b"", status=200),
|
|
)
|
|
|
|
with pytest.raises(MinerUError) as exc:
|
|
mineru_cloud.parse_cloud(
|
|
pdf,
|
|
tmp_path / "out",
|
|
MinerUConfig(mode="cloud", api_token="tok"),
|
|
poll_interval=0,
|
|
timeout=10,
|
|
)
|
|
assert "corrupt pdf" in str(exc.value)
|