146 lines
6.4 KiB
Python
146 lines
6.4 KiB
Python
# ============================================================
|
||
# PDF 图纸解析黄金测试（R71.1 扩展）
|
||
# 覆盖：解析契约、PNG 预览、上传/清单、HTTP API、批量解析。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.aps_domain.drawing_dxf import build_drawing_master_candidates
|
||
from server.aps_domain.drawing_pdf import inspect_pdf, render_pdf_preview
|
||
from server.gateway.app import create_app
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
|
||
def _make_pdf(path: Path, *, drawing_no: str = "5060102101-001-A",
|
||
extra_lines: tuple[str, ...] = ()) -> Path:
|
||
lines = (drawing_no,) + tuple(extra_lines)
|
||
content = "BT /F1 10 Tf 72 760 Td 14 TL\n" + "".join(f"({line}) Tj T*\n" for line in lines) + "ET"
|
||
stream = content.encode("latin-1")
|
||
objects = [
|
||
b"<< /Type /Catalog /Pages 2 0 R >>",
|
||
b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
|
||
b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 5 0 R >> >> /Contents 4 0 R >>",
|
||
b"<< /Length " + str(len(stream)).encode("ascii") + b" >>\nstream\n" + stream + b"\nendstream",
|
||
b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
|
||
]
|
||
body = bytearray(b"%PDF-1.4\n")
|
||
offsets: list[int] = []
|
||
for idx, obj in enumerate(objects, start=1):
|
||
offsets.append(len(body))
|
||
body += f"{idx} 0 obj\n".encode("ascii") + obj + b"\nendobj\n"
|
||
xref_pos = len(body)
|
||
body += f"xref\n0 {len(objects) + 1}\n".encode("ascii")
|
||
body += b"0000000000 65535 f \n"
|
||
for offset in offsets:
|
||
body += f"{offset:010d} 00000 n \n".encode("ascii")
|
||
body += f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref_pos}\n%%EOF\n".encode("ascii")
|
||
path.write_bytes(body)
|
||
return path
|
||
|
||
|
||
def test_inspect_pdf_contract_and_review_candidates(tmp_path: Path):
|
||
pdf = _make_pdf(
|
||
tmp_path / "5060102101-001-A.pdf",
|
||
extra_lines=("DRAWN BY: ENGINEER", "SCALE 1:1", "MATERIAL: STEEL"),
|
||
)
|
||
parsed = inspect_pdf(pdf)
|
||
|
||
assert parsed["asset"]["kind"] == "PDF"
|
||
assert len(parsed["asset"]["sha256"]) == 64
|
||
assert parsed["drawing"]["drawingNumber"] == "5060102101-001"
|
||
assert parsed["drawing"]["revision"] == "a"
|
||
assert parsed["drawing"]["pageCount"] == 1
|
||
assert any("DRAWN BY" in row["text"] for row in parsed["texts"])
|
||
assert any(
|
||
row["field"] == "drawingNumber" and row["value"] == "5060102101-001" and row["confidence"] >= 0.9
|
||
for row in parsed["fieldCandidates"]
|
||
)
|
||
|
||
candidates = build_drawing_master_candidates(parsed)
|
||
assert candidates["status"] == "PENDING_REVIEW"
|
||
assert candidates["reviewRequired"] is True
|
||
assert candidates["materials"][0]["code"] == "5060102101-001"
|
||
assert candidates["materials"][0]["reviewRequired"] is True
|
||
assert candidates["prohibitedAutoFill"]
|
||
|
||
|
||
def test_render_pdf_preview_returns_png_data_url(tmp_path: Path):
|
||
pdf = _make_pdf(tmp_path / "A-002-a.pdf")
|
||
preview = render_pdf_preview(pdf)
|
||
assert preview["mimeType"] == "image/png"
|
||
assert preview["dataUrl"].startswith("data:image/png;base64,")
|
||
assert len(preview["dataUrl"]) > 100
|
||
assert preview["width"] > 0 and preview["height"] > 0
|
||
|
||
|
||
def test_project_upload_accepts_pdf_and_lists_it(monkeypatch, tmp_path: Path):
|
||
import server.gateway.app as gateway_module
|
||
assert ".pdf" in gateway_module.PROJECT_UPLOAD_EXTS
|
||
|
||
install_test_auth(monkeypatch, "tenant-pdf-upload")
|
||
client = TestClient(create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
created = client.post("/api/projects", json={"name": "PDF 上传验收", "workDir": str(tmp_path)})
|
||
assert created.status_code == 200, created.text
|
||
pid = created.json()["project"]["id"]
|
||
|
||
pdf_bytes = _make_pdf(tmp_path / "B-001-a.pdf").read_bytes()
|
||
resp = client.post(
|
||
f"/api/projects/{pid}/files/upload",
|
||
files=[("files", ("B-001-a.pdf", pdf_bytes, "application/pdf"))],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.json()["saved"] == ["B-001-a.pdf"]
|
||
|
||
listing = client.get(f"/api/projects/{pid}/drawing-files")
|
||
assert listing.status_code == 200, listing.text
|
||
names = [row["name"] for row in listing.json()["files"]]
|
||
assert "B-001-a.pdf" in names
|
||
|
||
|
||
def test_drawing_http_api_inspect_pdf_flow(monkeypatch, tmp_path: Path):
|
||
install_test_auth(monkeypatch, "tenant-pdf-api")
|
||
client = TestClient(create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
created = client.post("/api/projects", json={"name": "PDF 解析验收", "workDir": str(tmp_path)})
|
||
assert created.status_code == 200, created.text
|
||
pid = created.json()["project"]["id"]
|
||
|
||
pdf = _make_pdf(tmp_path / "C-001-a.pdf", extra_lines=("MATERIAL: STEEL", "SCALE 1:1"))
|
||
resp = client.post("/api/drawings/inspect", json={"path": str(pdf), "projectId": pid})
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()
|
||
parsed = data.get("parsed") or data.get("drawing") or data
|
||
assert parsed["asset"]["kind"] == "PDF"
|
||
drawing_id = parsed["asset"]["id"]
|
||
|
||
detail = client.get(f"/api/drawings/{drawing_id}")
|
||
assert detail.status_code == 200, detail.text
|
||
assert detail.json()["format"] == "PDF"
|
||
|
||
preview = client.get(f"/api/drawings/{drawing_id}/preview")
|
||
assert preview.status_code == 200, preview.text
|
||
assert preview.json()["mimeType"] == "image/png"
|
||
|
||
|
||
def test_batch_inspect_accepts_multiple_pdf(monkeypatch, tmp_path: Path):
|
||
install_test_auth(monkeypatch, "tenant-pdf-batch")
|
||
client = TestClient(create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
created = client.post("/api/projects", json={"name": "PDF 批量验收", "workDir": str(tmp_path)})
|
||
assert created.status_code == 200, created.text
|
||
_make_pdf(tmp_path / "D-001-a.pdf")
|
||
_make_pdf(tmp_path / "D-002-b.pdf")
|
||
|
||
resp = client.post("/api/drawings/inspect-batch", json={"directory": str(tmp_path)})
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["count"] == 2
|
||
assert {row["fileName"] for row in body["drawings"]} == {"D-001-a.pdf", "D-002-b.pdf"}
|