91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
import sys
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from app.api.v1.websocket import WSMessageType, handle_client_message
|
|
from app.core.websocket_manager import MessageType
|
|
from app.api.v1 import websocket as websocket_module
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_convert_ifc_to_stp_batch_success(monkeypatch):
|
|
sent_messages = []
|
|
convert_calls = []
|
|
|
|
async def fake_send(message, client_id):
|
|
sent_messages.append((message, client_id))
|
|
|
|
monkeypatch.setattr(websocket_module.websocket_manager, "send_personal_message", fake_send)
|
|
monkeypatch.setattr(websocket_module.websocket_manager, "_get_timestamp", lambda: "2026-03-01T00:00:00")
|
|
|
|
class FakeConverter:
|
|
def convert(self, ifc_path: str, stp_path: str):
|
|
convert_calls.append((ifc_path, stp_path))
|
|
return {
|
|
"ifc_path": ifc_path,
|
|
"stp_path": stp_path,
|
|
"duration_ms": 1,
|
|
"products_total": 10,
|
|
"products_success": 10,
|
|
"products_failed": 0,
|
|
}
|
|
|
|
fake_converter_module = types.SimpleNamespace(Ifc2StpConverter=FakeConverter)
|
|
monkeypatch.setitem(sys.modules, "app.core.ifc2stp_converter", fake_converter_module)
|
|
|
|
await handle_client_message(
|
|
{
|
|
"type": WSMessageType.CONVERT_IFC_TO_STP_BATCH,
|
|
"items": [
|
|
{"ifc_path": "D:/cad/a.ifc", "stp_path": "D:/cad/a.stp"},
|
|
{"ifc_path": "D:/cad/b.ifc", "stp_path": "D:/cad/b.stp"},
|
|
],
|
|
},
|
|
client_id="test-client",
|
|
user_id="test-user",
|
|
)
|
|
|
|
assert convert_calls == [
|
|
("D:/cad/a.ifc", "D:/cad/a.stp"),
|
|
("D:/cad/b.ifc", "D:/cad/b.stp"),
|
|
]
|
|
assert len(sent_messages) == 1
|
|
payload, target_client_id = sent_messages[0]
|
|
assert target_client_id == "test-client"
|
|
assert payload["type"] == MessageType.INFO
|
|
assert payload["data"]["requested_count"] == 2
|
|
assert payload["data"]["success_count"] == 2
|
|
assert payload["data"]["failed_count"] == 0
|
|
assert [item["status"] for item in payload["data"]["results"]] == ["success", "success"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_convert_ifc_to_stp_batch_invalid_items(monkeypatch):
|
|
sent_messages = []
|
|
|
|
async def fake_send(message, client_id):
|
|
sent_messages.append((message, client_id))
|
|
|
|
monkeypatch.setattr(websocket_module.websocket_manager, "send_personal_message", fake_send)
|
|
monkeypatch.setattr(websocket_module.websocket_manager, "_get_timestamp", lambda: "2026-03-01T00:00:00")
|
|
|
|
await handle_client_message(
|
|
{
|
|
"type": WSMessageType.CONVERT_IFC_TO_STP_BATCH,
|
|
"items": [
|
|
{"ifc_path": "D:/cad/a.ifc"},
|
|
"invalid-item",
|
|
],
|
|
},
|
|
client_id="test-client",
|
|
user_id="test-user",
|
|
)
|
|
|
|
assert len(sent_messages) == 1
|
|
payload, target_client_id = sent_messages[0]
|
|
assert target_client_id == "test-client"
|
|
assert payload["type"] == MessageType.ERROR
|
|
assert payload["message"] == "批量参数校验失败"
|
|
assert payload["data"]["invalid_count"] == 2
|