212 lines
8.1 KiB
Python
212 lines
8.1 KiB
Python
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
|
|
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"
|
|
|
|
|
|
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")
|
|
}
|
|
|
|
|
|
class _WorldStore:
|
|
def __init__(self, data: dict, tenant_uuid: str, world_key: str = "default") -> None:
|
|
self.data = data
|
|
self.world_key = world_key
|
|
self.tenant_uuid = tenant_uuid
|
|
|
|
def next_id(self, _kind: str) -> int:
|
|
return len(self.data.get("auditEvents") or []) + 1
|
|
|
|
def save(self) -> None:
|
|
pass
|
|
|
|
|
|
def _auth_client(monkeypatch: pytest.MonkeyPatch, tenant: str) -> TestClient:
|
|
install_test_auth(monkeypatch, tenant)
|
|
monkeypatch.setenv("APS_DRAWING_ALLOWED_ROOTS", str(_sample_dxf().parent))
|
|
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 _synthetic_dxf(path: Path, text: str) -> Path:
|
|
payload = (
|
|
"0\nSECTION\n2\nHEADER\n0\nENDSEC\n0\nSECTION\n2\nENTITIES\n"
|
|
"0\nTEXT\n8\n0\n10\n0.0\n20\n0.0\n40\n2.5\n1\n"
|
|
f"{text}\n0\nENDSEC\n0\nEOF\n"
|
|
)
|
|
path.write_text(payload, encoding="ascii")
|
|
return path
|
|
|
|
|
|
def _bind_project(monkeypatch: pytest.MonkeyPatch, work_dir: Path) -> None:
|
|
class FakeProjectStore:
|
|
def snapshot(self, include_messages: bool = False):
|
|
return {
|
|
"projects": [{"id": "p-dxf-batch", "name": "批量验收项目", "workDir": str(work_dir)}],
|
|
"sessions": [{"id": "s-dxf-batch", "projectId": "p-dxf-batch"}],
|
|
"files": [],
|
|
}
|
|
|
|
def active_world_key(self) -> str:
|
|
return "default"
|
|
|
|
def ensure_session(self, session_id: str, project_id: str | None = None, title: str = ""):
|
|
return {"id": session_id, "projectId": project_id or "p-dxf-batch", "title": title}
|
|
|
|
def require_active_write(self) -> None:
|
|
pass
|
|
|
|
monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakeProjectStore())
|
|
|
|
|
|
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(monkeypatch):
|
|
import server.gateway.app as gateway_module
|
|
|
|
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 通道，"
|
|
"并在确认前保持主数据不变、确认后原子提交变更集。"
|
|
)
|
|
|
|
store = _WorldStore(seed_world(), "tenant-dxf-master-flow", "personal-1001")
|
|
monkeypatch.setattr(
|
|
gateway_module,
|
|
"_drawing_project_store",
|
|
lambda project_id, write=False: store,
|
|
)
|
|
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
|
_bind_project(monkeypatch, _sample_dxf().parent)
|
|
client = _auth_client(monkeypatch, "tenant-dxf-master-flow")
|
|
inspect_resp = client.post("/api/drawings/inspect", json={"path": str(_sample_dxf())})
|
|
assert inspect_resp.status_code == 200, inspect_resp.text
|
|
parsed = inspect_resp.json()["parsed"]
|
|
drawing_id = parsed["asset"]["id"]
|
|
|
|
before = _master_snapshot(store.data)
|
|
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
|
|
assert _master_snapshot(store.data) == before, "P2 暂存阶段不得直接写主数据"
|
|
|
|
confirm = client.post("/api/actions/confirm", json={
|
|
"confirmId": staged["confirmId"],
|
|
"approve": True,
|
|
})
|
|
assert confirm.status_code == 200, confirm.text
|
|
result = confirm.json()
|
|
assert "Drawing review applied" in result["message"]
|
|
|
|
world = store.data
|
|
assert any(row.get("code") == "5060102101-001" for row in world.get("materials") or [])
|
|
assert any(link.get("drawingId") == drawing_id for link in world.get("drawingLinks") or [])
|
|
asset = next(row for row in world.get("drawingAssets") or [] if row.get("id") == drawing_id)
|
|
assert asset.get("status") == "COMMITTED"
|
|
|
|
|
|
def test_batch_inspect_keeps_independent_candidate_change_sets(monkeypatch, tmp_path):
|
|
import server.gateway.app as gateway_module
|
|
|
|
p1 = _synthetic_dxf(tmp_path / "E2E-001-a.dxf", "E2E-DXF-ONE")
|
|
p2 = _synthetic_dxf(tmp_path / "E2E-002-a.dxf", "E2E-DXF-TWO")
|
|
_bind_project(monkeypatch, tmp_path)
|
|
install_test_auth(monkeypatch, "tenant-dxf-batch")
|
|
store = _WorldStore(seed_world(), "tenant-dxf-batch")
|
|
monkeypatch.setattr(
|
|
gateway_module,
|
|
"_drawing_project_store",
|
|
lambda project_id, write=False: store,
|
|
)
|
|
client = TestClient(create_app())
|
|
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
|
assert login.status_code == 200, login.text
|
|
|
|
resp = client.post("/api/drawings/inspect-batch", json={
|
|
"projectId": "p-dxf-batch",
|
|
"paths": [str(p1), str(p2)],
|
|
})
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["count"] == 2
|
|
assert body["errors"] == []
|
|
assert len({row["id"] for row in body["drawings"]}) == 2
|
|
|
|
world = store.data
|
|
changes = world.get("drawingCandidates") or []
|
|
assets = world.get("drawingAssets") or []
|
|
assert len(changes) == 2
|
|
assert len(assets) == 2
|
|
assert len({row.get("drawingAssetId") for row in changes}) == 2
|
|
assert all(row.get("projectId") == "p-dxf-batch" for row in changes)
|
|
|
|
again = client.post("/api/drawings/inspect-batch", json={
|
|
"projectId": "p-dxf-batch",
|
|
"paths": [str(p1), str(p2)],
|
|
})
|
|
assert again.status_code == 200, again.text
|
|
world = store.data
|
|
changes = world.get("drawingCandidates") or []
|
|
assets = world.get("drawingAssets") or []
|
|
assert len(changes) == 2
|
|
assert len(assets) == 2
|
|
codes = [
|
|
{row.get("code") for row in (change.get("materials") or [])}
|
|
for change in changes
|
|
]
|
|
assert codes[0] != codes[1]
|