78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
import asyncio
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from app.config import software_config
|
|
from app.core.cad_batch_manager import CadBatchManager
|
|
from app.core.cad_task_router import CadTaskRouter
|
|
from app.core.plugin_callback_registry import PluginCallbackRegistry
|
|
from app.models.cad_batch import BatchStatus, BatchSubmitItem, BatchSubmitRequest
|
|
|
|
|
|
class SyncClosePluginClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def submit_task(self, software_id: str, task_type: str, payload: dict) -> dict:
|
|
self.calls.append((software_id, task_type, payload))
|
|
if task_type == "close_model":
|
|
return {
|
|
"success": True,
|
|
"data": {
|
|
"model_name": "abc.prt",
|
|
"was_modified": False,
|
|
"close_time": "2026-03-01T12:34:56Z",
|
|
},
|
|
"error": None,
|
|
}
|
|
return {"success": False, "error": "unsupported task in this test"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creo_close_model_sync_completion():
|
|
backup = deepcopy(software_config.load_config())
|
|
try:
|
|
software_config._config["plugins"]["creo"]["tasks"]["close_model"]["completion_mode"] = "sync"
|
|
|
|
manager = CadBatchManager(
|
|
task_router=CadTaskRouter({".prt": "creo"}),
|
|
plugin_client=SyncClosePluginClient(),
|
|
callback_registry=PluginCallbackRegistry(),
|
|
)
|
|
await manager.start()
|
|
|
|
req = BatchSubmitRequest(
|
|
items=[
|
|
BatchSubmitItem(
|
|
model_path="x.prt",
|
|
task_type="close_model",
|
|
task_params={"force_close": True},
|
|
)
|
|
]
|
|
)
|
|
|
|
try:
|
|
batch = await manager.create_batch(req, submitter_id="tester")
|
|
|
|
start = asyncio.get_event_loop().time()
|
|
final_batch = None
|
|
while asyncio.get_event_loop().time() - start < 2:
|
|
final_batch = await manager.get_batch(batch.id)
|
|
if final_batch and final_batch.status in {BatchStatus.COMPLETED, BatchStatus.COMPLETED_WITH_ERRORS, BatchStatus.FAILED}:
|
|
break
|
|
await asyncio.sleep(0.02)
|
|
|
|
assert final_batch is not None
|
|
assert final_batch.status == BatchStatus.COMPLETED
|
|
|
|
items = await manager.get_batch_items(batch.id)
|
|
assert len(items) == 1
|
|
assert items[0].status.value == "succeeded"
|
|
assert items[0].result.get("success") is True
|
|
assert items[0].result.get("data", {}).get("model_name") == "abc.prt"
|
|
finally:
|
|
await manager.stop()
|
|
finally:
|
|
software_config._config = backup
|