238 lines
9.6 KiB
Python
238 lines
9.6 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.aps_domain.drawing_dxf import inspect_dxf, render_dxf_preview
|
||
from server.gateway.app import create_app
|
||
from tests.auth_provider import install_test_auth
|
||
from tests.external_data import external_dir
|
||
|
||
_DEFAULT_SAMPLE = external_dir("RUIYANG_DEMO_DIR", "ruiyang") / "5060102101-001-e(1).dxf"
|
||
_FORBIDDEN_SVG = re.compile(r"<(?:script|foreignObject)\b|(?:href|xlink:href)\s*=|\bon\w+\s*=", re.IGNORECASE)
|
||
|
||
|
||
def _sample_dxf() -> Path:
|
||
try:
|
||
import ezdxf # noqa: F401
|
||
except ImportError as exc:
|
||
pytest.skip(f"DXF 运行依赖未完整安装：{exc}; 需要 ezdxf 及其 fontTools 依赖")
|
||
path = Path(os.environ.get("APS_DXF_SAMPLE", str(_DEFAULT_SAMPLE)))
|
||
if not path.is_file():
|
||
pytest.skip(f"真实 DXF 样例不存在：{path}；可通过 APS_DXF_SAMPLE 指定")
|
||
return path
|
||
|
||
|
||
def test_real_dxf_metadata_entities_texts_and_dimensions_contract():
|
||
parsed = inspect_dxf(_sample_dxf())
|
||
|
||
drawing = parsed["drawing"]
|
||
assert drawing["metadata"]["acadVersion"].startswith("AC")
|
||
assert drawing["modelspaceEntityCount"] >= 9_000
|
||
assert drawing["layerCount"] > 0
|
||
assert drawing["bbox"]["width"] > 0
|
||
assert drawing["bbox"]["height"] > 0
|
||
assert sum(parsed["entityStatistics"].values()) == drawing["modelspaceEntityCount"]
|
||
assert parsed["entityStatistics"].get("LINE", 0) > 0
|
||
assert parsed["entityStatistics"].get("ARC", 0) > 0
|
||
assert len(parsed["texts"]) >= 10
|
||
assert all(row["text"] and row["evidence"].get("entityHandle") for row in parsed["texts"][:10])
|
||
assert len(parsed["dimensions"]) >= 50
|
||
assert any(row.get("measurement") is not None for row in parsed["dimensions"])
|
||
assert any(row["field"] == "drawingNumber" and row["value"] == "5060102101-001"
|
||
for row in parsed["fieldCandidates"])
|
||
|
||
|
||
def test_real_dxf_svg_preview_is_standalone_and_safe():
|
||
svg = render_dxf_preview(_sample_dxf())
|
||
|
||
assert svg.startswith("<svg ")
|
||
assert svg.endswith("</svg>")
|
||
assert "viewBox=" in svg
|
||
assert "sourceSha256=" in svg
|
||
assert "data-handle=" in svg
|
||
assert len(svg.encode("utf-8")) <= 2_000_000
|
||
assert not _FORBIDDEN_SVG.search(svg), "SVG 不得包含脚本、事件处理器或外部引用"
|
||
|
||
|
||
def test_drawing_http_api_contract_when_gateway_routes_are_merged(monkeypatch):
|
||
sample = _sample_dxf()
|
||
monkeypatch.setenv("APS_AUTH_ENABLED", "0")
|
||
monkeypatch.setenv("APS_DRAWING_ALLOWED_ROOTS", str(sample.parent))
|
||
app = create_app()
|
||
paths = app.openapi().get("paths", {})
|
||
required = {"/api/drawings/inspect", "/api/drawings/{drawing_id}", "/api/drawings/{drawing_id}/preview"}
|
||
missing = sorted(required - set(paths))
|
||
if missing:
|
||
pytest.skip(f"依赖未合并：Drawing HTTP API 缺少 {missing}")
|
||
|
||
client = TestClient(app)
|
||
response = client.post("/api/drawings/inspect", json={"path": str(sample)})
|
||
assert response.status_code == 200, response.text
|
||
data = response.json()
|
||
parsed = data.get("parsed") or data.get("drawing") or data
|
||
asset = parsed["asset"]
|
||
assert asset["kind"] == "DXF"
|
||
assert parsed["drawing"]["drawingNumber"] == "5060102101-001"
|
||
|
||
drawing_id = asset["id"]
|
||
detail = client.get(f"/api/drawings/{drawing_id}")
|
||
assert detail.status_code == 200, detail.text
|
||
preview = client.get(f"/api/drawings/{drawing_id}/preview")
|
||
assert preview.status_code == 200, preview.text
|
||
assert "<svg" in preview.text
|
||
assert not _FORBIDDEN_SVG.search(preview.text)
|
||
|
||
|
||
def _auth_client(monkeypatch: pytest.MonkeyPatch, tenant: str) -> TestClient:
|
||
install_test_auth(monkeypatch, tenant)
|
||
client = TestClient(create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
return client
|
||
|
||
|
||
def _create_project(client: TestClient, name: str, work_dir: Path) -> str:
|
||
created = client.post("/api/projects", json={"name": name, "workDir": str(work_dir)})
|
||
assert created.status_code == 200, created.text
|
||
return created.json()["project"]["id"]
|
||
|
||
|
||
def test_project_upload_accepts_dxf_and_rejects_unknown_type(monkeypatch, tmp_path: Path):
|
||
client = _auth_client(monkeypatch, "tenant-dxf-upload")
|
||
pid = _create_project(client, "DXF 上传验收", tmp_path)
|
||
|
||
ok = client.post(
|
||
f"/api/projects/{pid}/files/upload",
|
||
files=[("files", ("E2E-001-a.dxf", b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n", "application/octet-stream"))],
|
||
)
|
||
assert ok.status_code == 200, ok.text
|
||
assert ok.json()["saved"] == ["E2E-001-a.dxf"]
|
||
assert (tmp_path / "E2E-001-a.dxf").is_file()
|
||
|
||
bad = client.post(
|
||
f"/api/projects/{pid}/files/upload",
|
||
files=[("files", ("说明.docx", b"not a drawing", "application/octet-stream"))],
|
||
)
|
||
assert bad.status_code == 400
|
||
detail = bad.json()["detail"]
|
||
assert "仅支持" in detail
|
||
assert ".dxf" in detail
|
||
|
||
|
||
def test_upload_size_limit_returns_readable_error(monkeypatch, tmp_path: Path):
|
||
import server.gateway.app as gateway_module
|
||
|
||
client = _auth_client(monkeypatch, "tenant-dxf-upload-limit")
|
||
monkeypatch.setattr(gateway_module, "PROJECT_UPLOAD_MAX_BYTES", 32)
|
||
pid = _create_project(client, "超限验收", tmp_path)
|
||
original = b"existing-valid-drawing"
|
||
(tmp_path / "big.dxf").write_bytes(original)
|
||
resp = client.post(
|
||
f"/api/projects/{pid}/files/upload",
|
||
files=[("files", ("big.dxf", b"x" * 64, "application/octet-stream"))],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
errors = resp.json().get("errors") or []
|
||
assert errors
|
||
assert "超过" in errors[0]["error"]
|
||
assert (tmp_path / "big.dxf").read_bytes() == original
|
||
assert not list(tmp_path.glob(".*.upload"))
|
||
|
||
|
||
def test_drawing_inspect_errors_are_readable_chinese(monkeypatch, tmp_path: Path):
|
||
client = _auth_client(monkeypatch, "tenant-dxf-errors")
|
||
allowed = tmp_path / "allowed"
|
||
allowed.mkdir()
|
||
monkeypatch.setenv("APS_DRAWING_ALLOWED_ROOTS", str(allowed))
|
||
|
||
outside = tmp_path / "outside.dxf"
|
||
outside.write_bytes(b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n")
|
||
denied = client.post("/api/drawings/inspect", json={"path": str(outside)})
|
||
assert denied.status_code == 403
|
||
assert "不在项目工程目录" in denied.json()["detail"]
|
||
|
||
wrong = tmp_path / "drawing.txt"
|
||
wrong.write_text("not a drawing", encoding="utf-8")
|
||
bad = client.post("/api/drawings/inspect", json={"path": str(wrong)})
|
||
assert bad.status_code == 400
|
||
assert "仅允许读取存在的 DXF" in bad.json()["detail"]
|
||
|
||
|
||
def test_project_drawing_files_listing_returns_dxf_paths(monkeypatch, tmp_path: Path):
|
||
client = _auth_client(monkeypatch, "tenant-dxf-files")
|
||
pid = _create_project(client, "图纸清单验收", tmp_path)
|
||
(tmp_path / "A-001-a.dxf").write_bytes(b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n")
|
||
(tmp_path / "订单.csv").write_text("订单号,数量\nSO-1,1\n", encoding="utf-8")
|
||
|
||
resp = client.get(f"/api/projects/{pid}/drawing-files")
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["count"] == 1
|
||
assert body["files"][0]["name"] == "A-001-a.dxf"
|
||
assert body["files"][0]["path"] == "A-001-a.dxf"
|
||
assert body["workDir"] is None
|
||
|
||
|
||
def test_drawing_inspect_is_strictly_scoped_to_requested_project(monkeypatch, tmp_path: Path):
|
||
client = _auth_client(monkeypatch, "tenant-dxf-project-scope")
|
||
first_root = tmp_path / "first"
|
||
second_root = tmp_path / "second"
|
||
first_root.mkdir()
|
||
second_root.mkdir()
|
||
first_pid = _create_project(client, "图纸项目 A", first_root)
|
||
second_pid = _create_project(client, "图纸项目 B", second_root)
|
||
drawing = first_root / "A-001-a.dxf"
|
||
drawing.write_bytes(b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n")
|
||
|
||
denied = client.post(
|
||
"/api/drawings/inspect",
|
||
json={"projectId": second_pid, "path": str(drawing)},
|
||
)
|
||
assert denied.status_code == 403
|
||
|
||
allowed = client.post(
|
||
"/api/drawings/inspect",
|
||
json={"projectId": first_pid, "path": drawing.name},
|
||
)
|
||
assert allowed.status_code == 200, allowed.text
|
||
response_body = allowed.json()
|
||
parsed = response_body["drawing"]
|
||
assert "sourcePath" not in response_body["parsed"]["asset"]
|
||
assert parsed["filePath"] is None
|
||
assert all(row["status"] == "PENDING_REVIEW" for row in parsed["candidates"])
|
||
assert all(row["status"] == "PENDING_REVIEW" for row in parsed["fields"])
|
||
assert all(isinstance(row["reviewRequired"], bool) for row in parsed["fields"])
|
||
listed = client.get(f"/api/projects/{first_pid}/drawings")
|
||
assert listed.status_code == 200
|
||
assert listed.json()["drawings"][0]["projectId"] == first_pid
|
||
|
||
|
||
def test_project_drawing_scope_cannot_be_widened_by_global_roots(monkeypatch, tmp_path: Path):
|
||
client = _auth_client(monkeypatch, "tenant-dxf-global-root-scope")
|
||
first_root = tmp_path / "first"
|
||
second_root = tmp_path / "second"
|
||
first_root.mkdir()
|
||
second_root.mkdir()
|
||
first_pid = _create_project(client, "全局根项目 A", first_root)
|
||
second_pid = _create_project(client, "全局根项目 B", second_root)
|
||
drawing = first_root / "A-002-a.dxf"
|
||
drawing.write_bytes(b"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nEOF\n")
|
||
monkeypatch.setenv("APS_DRAWING_ALLOWED_ROOTS", str(first_root))
|
||
|
||
denied = client.post(
|
||
"/api/drawings/inspect",
|
||
json={"projectId": second_pid, "path": str(drawing)},
|
||
)
|
||
assert denied.status_code == 403
|
||
|
||
allowed = client.post(
|
||
"/api/drawings/inspect",
|
||
json={"projectId": first_pid, "path": drawing.name},
|
||
)
|
||
assert allowed.status_code == 200, allowed.text
|