aps-agent/tests/golden/test_dxf_master_flow.py

91 lines
3.6 KiB
Python
Raw Normal View History

from __future__ import annotations
import os
from copy import deepcopy
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from server.aps_domain.drawing_dxf import build_drawing_master_candidates, inspect_dxf
from server.gateway.app import create_app
from server.state.seed import seed_world
_DEFAULT_SAMPLE = Path(r"D:\ItemSpace\14.工业智核\锐扬\5060102101-001-e(1).dxf")
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 _master_snapshot(world: dict) -> dict:
return {
key: deepcopy(world.get(key, []))
for key in ("materials", "boms", "bom", "routing", "routings", "operations")
}
def test_dxf_candidates_are_review_only_and_do_not_mutate_master_data():
world = seed_world()
before = _master_snapshot(world)
candidates = build_drawing_master_candidates(inspect_dxf(_sample_dxf()))
assert candidates["contractVersion"] == "drawing-master-candidates.v1"
assert candidates["status"] == "PENDING_REVIEW"
assert candidates["reviewRequired"] is True
assert candidates["materials"]
assert candidates["materials"][0]["code"] == "5060102101-001"
assert candidates["materials"][0]["reviewRequired"] is True
assert candidates["bomReferences"]
assert all(row["resolutionStatus"] == "UNRESOLVED" for row in candidates["bomReferences"])
assert all(row["quantity"] is None for row in candidates["bomReferences"])
assert candidates["routingOperations"]
assert all(row["standardTime"] is None and row["resourceCode"] is None
for row in candidates["routingOperations"])
assert _master_snapshot(world) == before, "解析和候选生成阶段不得直接写主数据"
def test_dxf_p2_confirmation_is_required_before_master_commit_contract():
app = create_app()
paths = app.openapi().get("paths", {})
stage_path = "/api/drawings/{drawing_id}/stage"
confirm_path = "/api/actions/confirm"
if stage_path not in paths or confirm_path not in paths:
pytest.skip(
"依赖未合并:需要 Drawing stage 接口接入现有 P2 /api/actions/confirm 通道,"
"并在确认前保持主数据不变、确认后原子提交变更集。"
)
client = TestClient(app)
inspect_resp = client.post("/api/drawings/inspect", json={"path": str(_sample_dxf())})
if inspect_resp.status_code in (401, 403):
pytest.skip("Drawing API 已存在但测试认证装配尚未接入;需复用 tests.auth_provider")
assert inspect_resp.status_code == 200, inspect_resp.text
parsed = inspect_resp.json().get("parsed") or inspect_resp.json().get("drawing") or inspect_resp.json()
drawing_id = parsed["asset"]["id"]
stage = client.post(f"/api/drawings/{drawing_id}/stage", json={})
assert stage.status_code == 200, stage.text
staged = stage.json()
assert staged["status"] in {"staged", "PENDING_REVIEW"}
assert staged.get("confirmId"), "候选必须生成 P2 确认卡"
assert staged.get("masterCommitted", False) is False
confirm = client.post("/api/actions/confirm", json={
"confirmId": staged["confirmId"],
"approve": True,
})
assert confirm.status_code == 200, confirm.text
result = confirm.json()
assert result.get("approved", True) is True
assert result.get("masterCommitted", True) is True