369 lines
14 KiB
Python
369 lines
14 KiB
Python
# ============================================================
|
||
# 工程图纸 DXF 自动识别黄金测试（domain-dxf-drawing）
|
||
# 覆盖：样例图纸解析（标题栏/位置号/技术说明）、文件名交叉校验、
|
||
# 物料/BOM/工艺路线候选安全边界（missing 不编造）、SVG 预览、
|
||
# 工程目录发现、多模态提取器接线与入库 fail closed、
|
||
# /api/dxf/* 端点、P2 确认卡 staging 与确认后写主数据。
|
||
# 样例：demand/dxf-samples/5060102101-001-e.dxf
|
||
# （MTU 12V 183 TB32 发动机安装图，AC1015，德英双语标题栏）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.agent_core import harness
|
||
from server.aps_domain.dxf_drawing import (
|
||
discover_dxf_files,
|
||
drawing_to_master_candidates,
|
||
dxf_svg_cached,
|
||
dxf_to_svg,
|
||
get_cached_drawing,
|
||
parse_dxf,
|
||
parse_dxf_bytes,
|
||
)
|
||
from server.aps_domain.masterdata import confirmation_for_master_action
|
||
from server.aps_domain.multimodal import candidates_to_batches, default_registry
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
||
from server.state.seed import seed_world
|
||
|
||
_SAMPLE = os.path.join(os.path.dirname(__file__), "..", "..",
|
||
"demand", "dxf-samples", "5060102101-001-e.dxf")
|
||
_SAMPLE = os.path.abspath(_SAMPLE)
|
||
_PLANNER = IdentityContext(1001, "planner", "计划员", "tenant-dxf", roles=("planner",))
|
||
|
||
|
||
def _next_id(world):
|
||
counters: dict[str, int] = {}
|
||
|
||
def next_id(kind: str) -> int:
|
||
table = "auditEvents" if kind == "audit" else kind + "s"
|
||
if kind not in counters:
|
||
rows = world.get(table, [])
|
||
counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
||
counters[kind] += 1
|
||
return counters[kind]
|
||
|
||
return next_id
|
||
|
||
|
||
class _WorldStore:
|
||
def __init__(self, world, next_id_fn):
|
||
self.data = world
|
||
self._next_id_fn = next_id_fn
|
||
self.tenant_uuid = "tenant-dxf"
|
||
self.world_key = "personal-1001"
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
return self._next_id_fn(kind)
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def parsed():
|
||
assert os.path.isfile(_SAMPLE), f"样例图纸缺失：{_SAMPLE}"
|
||
return parse_dxf(_SAMPLE)
|
||
|
||
|
||
# ---------------- 解析：标题栏 / 位置号 / 技术说明 ----------------
|
||
|
||
def test_parse_sample_title_block(parsed):
|
||
assert parsed["dxfVersion"] == "AC1015"
|
||
tb = parsed["titleBlock"]
|
||
assert tb["found"] is True
|
||
assert tb["source"] == "block-attribs"
|
||
assert tb["blockName"] == "SF_STD_A"
|
||
fields = tb["fields"]
|
||
dn = fields["drawingNumber"]
|
||
assert dn["value"] == "506 010 21 01"
|
||
assert dn["normalized"] == "5060102101"
|
||
assert dn["confidence"] == pytest.approx(0.95) # 属性 + 文件名一致互证
|
||
assert "12V 183 TB32" in fields["title"]["value"]
|
||
assert fields["scale"]["value"] == "1:5"
|
||
assert fields["sheet"]["value"] == "1"
|
||
assert fields["drawingType"]["value"] == "EB/Installation"
|
||
assert fields["applicableTo"]["value"] == "OM 444 LA"
|
||
assert fields["revision"]["value"] == "e" # 文件名后缀 + B_FREI 互证
|
||
assert fields["revision"]["confidence"] == pytest.approx(0.9)
|
||
# 未填写字段保持空 + missing（不编造材质）
|
||
assert fields["materialSpec"]["value"] == ""
|
||
assert fields["materialSpec"]["missing"] is True
|
||
|
||
|
||
def test_parse_positions_notes_and_stats(parsed):
|
||
positions = {p["position"] for p in parsed["positionNumbers"]}
|
||
assert {"14.001", "06.020", "15.108", "09.152"} <= positions
|
||
assert len(parsed["positionNumbers"]) >= 30
|
||
assert len(parsed["technicalNotes"]) >= 5
|
||
assert parsed["entityStats"]["LINE"] > 0
|
||
assert parsed["entityStats"]["ARC"] > 0
|
||
assert any(l["name"] == "RAHM2" for l in parsed["layers"])
|
||
assert parsed["extents"]["maxX"] > parsed["extents"]["minX"]
|
||
assert parsed["warnings"] == []
|
||
|
||
|
||
def test_parse_fail_closed():
|
||
with pytest.raises(ValueError, match="不存在"):
|
||
parse_dxf(r"D:\no-such-dir\no-such-file.dxf")
|
||
bad = os.path.join(os.path.dirname(_SAMPLE), "not-a-dxf.dxf")
|
||
with open(bad, "w", encoding="utf-8") as fh:
|
||
fh.write("这不是 DXF 内容")
|
||
try:
|
||
with pytest.raises(ValueError, match="解析失败"):
|
||
parse_dxf(bad)
|
||
finally:
|
||
os.unlink(bad)
|
||
|
||
|
||
def test_parse_bytes_uses_display_name():
|
||
with open(_SAMPLE, "rb") as fh:
|
||
data = fh.read()
|
||
parsed = parse_dxf_bytes("5060102101-001-e.dxf", data)
|
||
assert parsed["filename"] == "5060102101-001-e.dxf"
|
||
assert parsed["fileMeta"]["revision"] == "e"
|
||
assert parsed["titleBlock"]["fields"]["revision"]["value"] == "e"
|
||
entry = get_cached_drawing(parsed["drawingId"])
|
||
assert entry is not None and entry.get("tempPath")
|
||
|
||
|
||
# ---------------- 候选：安全边界（missing 不编造） ----------------
|
||
|
||
def test_master_candidates_boundary(parsed):
|
||
cands = drawing_to_master_candidates(parsed)
|
||
material = cands["material"]
|
||
assert material["code"] == "5060102101-001" # 张次沿用文件名 001 写法
|
||
assert "12V 183 TB32" in material["name"]
|
||
assert material["type"] == "FINISHED_PRODUCT" # 安装图推断（置信封顶 0.6）
|
||
assert material["requiresConfirm"] is True
|
||
assert material["confidence"] <= 0.95
|
||
assert material["materialSpec"] == "" # 图纸未承载：不编造材质
|
||
assert any("材质" in x for x in material["notFromDrawing"])
|
||
|
||
bom = cands["bom"]
|
||
assert bom["complete"] is False
|
||
assert bom["requiresConfirm"] is True
|
||
assert len(bom["items"]) >= 30
|
||
item = bom["items"][0]
|
||
assert item["missing"] == ["materialCode", "name", "quantity"]
|
||
assert item["quantity"] is None # 数量绝不臆造
|
||
|
||
routing = cands["routing"]
|
||
assert routing["complete"] is False
|
||
assert routing["steps"] == [] # 工序/工时留空待人工编制
|
||
assert "人工确认" in routing["note"]
|
||
|
||
|
||
# ---------------- SVG 预览与目录发现 ----------------
|
||
|
||
def test_svg_preview(parsed):
|
||
svg = dxf_svg_cached(parsed["drawingId"])
|
||
assert svg.startswith("<?xml") and "<svg" in svg
|
||
assert len(svg) > 100_000
|
||
svg2 = dxf_to_svg(_SAMPLE)
|
||
assert "<svg" in svg2
|
||
with pytest.raises(ValueError, match="未解析"):
|
||
dxf_svg_cached("0" * 16)
|
||
|
||
|
||
def test_discover_dxf_files():
|
||
result = discover_dxf_files(os.path.dirname(_SAMPLE))
|
||
assert result["count"] >= 1
|
||
hit = next(f for f in result["files"]
|
||
if f["filename"] == "5060102101-001-e.dxf")
|
||
assert hit["fileMeta"]["drawingNumberNormalized"] == "5060102101"
|
||
assert hit["fileMeta"]["sheet"] == "001"
|
||
assert hit["fileMeta"]["revision"] == "e"
|
||
with pytest.raises(ValueError, match="不存在"):
|
||
discover_dxf_files(r"D:\no-such-dir\x")
|
||
|
||
|
||
# ---------------- 多模态接线：提取可用 / 入库 fail closed ----------------
|
||
|
||
def test_multimodal_dxf_extractor():
|
||
registry = default_registry(world=seed_world())
|
||
assert "dxf_drawing" in registry.list_kinds()
|
||
cands = registry.extract("dxf_drawing", {"path": _SAMPLE})
|
||
assert len(cands) == 1
|
||
c = cands[0]
|
||
assert c.kind == "dxf_drawing" and c.source == "DXF_PARSE"
|
||
assert c.complete is True
|
||
assert c.confidence >= 0.9
|
||
assert c.value["material"]["code"] == "5060102101-001"
|
||
assert c.value["positionCount"] >= 30
|
||
# DXF 候选不经 importers 批次：显式报错并指引 /api/dxf/stage
|
||
with pytest.raises(ValueError, match="/api/dxf/stage"):
|
||
candidates_to_batches([c])
|
||
|
||
|
||
# ---------------- P2 确认卡：确认前不写世界，确认后写主数据 ----------------
|
||
|
||
def test_dxf_stage_then_confirm_writes_material(parsed):
|
||
"""候选 → master.material.upsert P2 卡 → 批准写物料 / 驳回不写。"""
|
||
world = seed_world()
|
||
next_id = _next_id(world)
|
||
material = drawing_to_master_candidates(parsed)["material"]
|
||
payload = {
|
||
"code": material["code"], "name": material["name"],
|
||
"type": material["type"], "unit": material["unit"],
|
||
"spec": material["spec"],
|
||
}
|
||
title, lines = confirmation_for_master_action(
|
||
world, "master.material.upsert", payload)
|
||
assert material["code"] in title or material["name"] in title
|
||
assert lines
|
||
token = bind_identity(_PLANNER)
|
||
try:
|
||
block = harness.stage_confirmation(
|
||
"dxf-test", "master.material.upsert", payload,
|
||
title=title, summary_lines=lines)
|
||
cid = str(block.props["confirmId"])
|
||
assert not any(m["code"] == material["code"] for m in world["materials"])
|
||
store = _WorldStore(world, next_id)
|
||
message = execute_confirmed(store, cid, True, actor="planner")
|
||
assert "物料" in message or "完成" in message
|
||
written = next(m for m in world["materials"] if m["code"] == material["code"])
|
||
assert written["name"] == material["name"]
|
||
assert any(e["action"] == "master.material.upsert"
|
||
for e in world["auditEvents"])
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_dxf_stage_validation_fail_closed(parsed):
|
||
"""缺名称/编码的候选不出卡：confirmation_for_master_action 直接报错。"""
|
||
world = seed_world()
|
||
with pytest.raises(ValueError):
|
||
confirmation_for_master_action(
|
||
world, "master.material.upsert", {"code": "", "name": ""})
|
||
|
||
|
||
# ---------------- 网关接线：/api/dxf/* ----------------
|
||
|
||
class _AuditStore:
|
||
def __init__(self, tenant_uuid: str = "tenant-dxf"):
|
||
self.tenant_uuid = tenant_uuid
|
||
self.world_key = "personal-1001"
|
||
self.data: dict = seed_world()
|
||
self._counter = 0
|
||
|
||
def next_id(self, _kind: str) -> int:
|
||
self._counter += 1
|
||
return self._counter
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
|
||
class _CheckpointStore:
|
||
def create(self, *_args, **_kwargs) -> dict:
|
||
return {"pairId": "dxf-pair"}
|
||
|
||
def get(self, pair_id: str) -> dict | None:
|
||
return {"pairId": pair_id}
|
||
|
||
|
||
class _ProjectStore:
|
||
def active_world_key(self) -> str:
|
||
return "default"
|
||
|
||
def require_active_write(self) -> None:
|
||
pass
|
||
|
||
|
||
@pytest.fixture
|
||
def client_factory(monkeypatch):
|
||
import server.gateway.app as gateway_module
|
||
import server.state.projects as projects_module
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
provider = install_test_auth(monkeypatch, "tenant-dxf")
|
||
|
||
def factory():
|
||
store = _AuditStore()
|
||
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
||
monkeypatch.setattr(gateway_module, "get_checkpoints", lambda: _CheckpointStore())
|
||
monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
|
||
client = TestClient(gateway_module.create_app())
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
return client, store, provider
|
||
|
||
return factory
|
||
|
||
|
||
def test_api_parse_path_svg_and_discover(client_factory):
|
||
client, store, _ = client_factory()
|
||
resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()
|
||
drawing = data["drawing"]
|
||
assert drawing["titleBlock"]["fields"]["drawingNumber"]["normalized"] == "5060102101"
|
||
assert data["candidates"]["material"]["code"] == "5060102101-001"
|
||
assert any(e["action"] == "dxf.parse" for e in store.data["auditEvents"])
|
||
|
||
svg_resp = client.get(f"/api/dxf/{drawing['drawingId']}/svg")
|
||
assert svg_resp.status_code == 200
|
||
assert svg_resp.headers["content-type"].startswith("image/svg+xml")
|
||
assert "<svg" in svg_resp.text
|
||
|
||
disc = client.get("/api/dxf/discover", params={"dir": os.path.dirname(_SAMPLE)})
|
||
assert disc.status_code == 200
|
||
assert disc.json()["count"] >= 1
|
||
|
||
bad = client.post("/api/dxf/parse-path", json={"path": r"D:\no-such\x.dxf"})
|
||
assert bad.status_code == 200 and "error" in bad.json()
|
||
missing = client.get("/api/dxf/ffffffffffffffff/svg")
|
||
assert missing.status_code == 404
|
||
|
||
|
||
def test_api_dxf_stage_material_card(client_factory):
|
||
client, store, _ = client_factory()
|
||
resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
|
||
drawing = resp.json()["drawing"]
|
||
material = resp.json()["candidates"]["material"]
|
||
before = len(store.data["materials"])
|
||
stage = client.post("/api/dxf/stage", json={
|
||
"drawingId": drawing["drawingId"],
|
||
"target": "material",
|
||
"payload": {"code": material["code"], "name": material["name"],
|
||
"type": material["type"], "unit": material["unit"],
|
||
"spec": material["spec"]},
|
||
})
|
||
assert stage.status_code == 200, stage.text
|
||
body = stage.json()
|
||
assert "P2 确认队列" in body["message"]
|
||
assert body["block"]["props"]["confirmId"]
|
||
assert len(store.data["materials"]) == before # 确认前绝不写世界
|
||
assert any(e["action"] == "dxf.stage.material" for e in store.data["auditEvents"])
|
||
|
||
bad = client.post("/api/dxf/stage", json={
|
||
"target": "material", "payload": {"code": "", "name": ""}})
|
||
assert "error" in bad.json()
|
||
unknown = client.post("/api/dxf/stage", json={"target": "magic", "payload": {}})
|
||
assert "error" in unknown.json()
|
||
|
||
|
||
def test_svg_layer_filter_variant(parsed):
|
||
"""图层开关：过滤渲染显著小于全量渲染，且变体缓存命中。"""
|
||
all_svg = dxf_svg_cached(parsed["drawingId"])
|
||
only6 = dxf_svg_cached(parsed["drawingId"], include_layers=["6"])
|
||
assert "<svg" in only6
|
||
assert len(only6) < len(all_svg) // 2 # 单图层远小于全图
|
||
again = dxf_svg_cached(parsed["drawingId"], include_layers=["6"])
|
||
assert again is only6 # 变体缓存命中（同一对象）
|
||
|
||
|
||
def test_api_svg_layer_param(client_factory):
|
||
client, _, _ = client_factory()
|
||
resp = client.post("/api/dxf/parse-path", json={"path": _SAMPLE})
|
||
drawing = resp.json()["drawing"]
|
||
full = client.get(f"/api/dxf/{drawing['drawingId']}/svg")
|
||
filtered = client.get(f"/api/dxf/{drawing['drawingId']}/svg", params={"layers": "6"})
|
||
assert full.status_code == 200 and filtered.status_code == 200
|
||
assert len(filtered.text) < len(full.text) // 2
|