728 lines
24 KiB
Python
728 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from threading import Event
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from k1link.simulation.gaussian_pipeline_gateway import (
|
|
GaussianPipelineUnavailableError,
|
|
GaussianSourceBundleUpload,
|
|
GaussianSourceMemberUpload,
|
|
)
|
|
from k1link.simulation.projects import (
|
|
SimulationProjectConflictError,
|
|
SimulationProjectNotFoundError,
|
|
SimulationProjectService,
|
|
SimulationProjectStore,
|
|
_world_manifest,
|
|
)
|
|
from k1link.web.simulation_projects_api import build_simulation_projects_router
|
|
|
|
|
|
def _folder_files() -> list[dict[str, object]]:
|
|
return [
|
|
{"logical_path": "export/scene.lcc", "byte_length": 23},
|
|
{"logical_path": "export/index.bin", "byte_length": 5},
|
|
{"logical_path": "export/data.bin", "byte_length": 4},
|
|
]
|
|
|
|
|
|
def _default_ugv_settings() -> dict[str, object]:
|
|
return {
|
|
"preset_name": "UGV 100 кг",
|
|
"mass_kg": 100.0,
|
|
"dimensions_m": {
|
|
"length": 1.0,
|
|
"width": 0.8,
|
|
"height": 0.4,
|
|
"ground_clearance": 0.15,
|
|
},
|
|
"max_speed_mps": 1.2,
|
|
"max_turn_rate_degrees": 45.0,
|
|
"invert_steering": False,
|
|
}
|
|
|
|
|
|
def _upload_all(store: SimulationProjectStore, project: dict[str, Any]) -> None:
|
|
payloads = {
|
|
"export/scene.lcc": b'{"fileType":"Portable"}',
|
|
"export/index.bin": b"index",
|
|
"export/data.bin": b"data",
|
|
}
|
|
for source_file in project["source"]["files"]:
|
|
store.append_upload(
|
|
project["project_id"],
|
|
source_file["file_id"],
|
|
offset=0,
|
|
payload=payloads[source_file["logical_path"]],
|
|
)
|
|
|
|
|
|
def test_store_accepts_four_gib_archive_without_allocating_payload(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Four GiB archive",
|
|
scene_type="outdoor",
|
|
source_kind="archive",
|
|
files=[{"logical_path": "four-gib.7z", "byte_length": 4 * 1024**3}],
|
|
)
|
|
|
|
assert project["status"] == "uploading"
|
|
assert project["source"]["total_byte_length"] == 4 * 1024**3
|
|
|
|
|
|
def test_store_persists_resumable_folder_upload_and_state_transitions(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name=" Smolensky Boulevard ",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
assert project["name"] == "Smolensky Boulevard"
|
|
first = project["source"]["files"][0]
|
|
store.append_upload(
|
|
project["project_id"],
|
|
first["file_id"],
|
|
offset=0,
|
|
payload=b'{"fileType":',
|
|
)
|
|
with pytest.raises(SimulationProjectConflictError, match="offset"):
|
|
store.append_upload(
|
|
project["project_id"],
|
|
first["file_id"],
|
|
offset=0,
|
|
payload=b'"Portable"}',
|
|
)
|
|
store.append_upload(
|
|
project["project_id"],
|
|
first["file_id"],
|
|
offset=len(b'{"fileType":'),
|
|
payload=b'"Portable"}',
|
|
)
|
|
for source_file, payload in zip(
|
|
project["source"]["files"][1:],
|
|
(b"index", b"data"),
|
|
strict=True,
|
|
):
|
|
store.append_upload(
|
|
project["project_id"],
|
|
source_file["file_id"],
|
|
offset=0,
|
|
payload=payload,
|
|
)
|
|
queued = store.begin_build(project["project_id"])
|
|
assert queued["status"] == "queued"
|
|
assert queued["provider"]["job_created_at_utc"] is None
|
|
assert queued["provider"]["state_started_at_utc"] is None
|
|
with pytest.raises(SimulationProjectConflictError, match="active"):
|
|
store.delete(project["project_id"])
|
|
|
|
|
|
def test_store_preserves_provider_job_and_stage_timing(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Timed scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
|
|
first = store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_job_id="gsp-20260826000000-deadbeef",
|
|
provider_job_created_at_utc="2026-08-26T09:45:14.000Z",
|
|
provider_state="building_streamed_sog",
|
|
provider_state_started_at_utc="2026-08-26T09:48:20.000Z",
|
|
)
|
|
repeated = store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_state="building_streamed_sog",
|
|
provider_state_started_at_utc="2026-08-26T09:55:00.000Z",
|
|
)
|
|
|
|
assert first["provider"]["job_created_at_utc"] == "2026-08-26T09:45:14.000Z"
|
|
assert first["provider"]["state_started_at_utc"] == "2026-08-26T09:48:20.000Z"
|
|
assert repeated["provider"]["state_started_at_utc"] == "2026-08-26T09:48:20.000Z"
|
|
|
|
|
|
class _ReadyProvider:
|
|
def __init__(self) -> None:
|
|
self.deleted: list[str] = []
|
|
self.upload_calls = 0
|
|
self.submit_calls = 0
|
|
self.submitted_document: dict[str, object] | None = None
|
|
|
|
def capabilities(self) -> dict[str, object]:
|
|
return {"outputs": ["preview.sog", "streamed-sog"]}
|
|
|
|
def upload_source_bundle(
|
|
self,
|
|
_root: Path,
|
|
*,
|
|
entrypoint: str,
|
|
source_format: str,
|
|
) -> GaussianSourceBundleUpload:
|
|
self.upload_calls += 1
|
|
members = (
|
|
GaussianSourceMemberUpload("upload-1", entrypoint, "a" * 64, 23),
|
|
GaussianSourceMemberUpload("upload-2", "export/data.bin", "b" * 64, 4),
|
|
GaussianSourceMemberUpload("upload-3", "export/index.bin", "c" * 64, 5),
|
|
)
|
|
return GaussianSourceBundleUpload(
|
|
format=source_format,
|
|
entrypoint=entrypoint,
|
|
bundle_sha256="d" * 64,
|
|
total_byte_length=32,
|
|
members=members,
|
|
)
|
|
|
|
def submit_build(self, document: dict[str, object]) -> dict[str, object]:
|
|
self.submit_calls += 1
|
|
self.submitted_document = document
|
|
return {
|
|
"schema_version": "gaussian-pipeline.job/v1",
|
|
"job_id": "gsp-20260826000000-deadbeef",
|
|
"state": "queued",
|
|
"created_at_utc": "2026-08-26T09:45:14.000Z",
|
|
"updated_at_utc": "2026-08-26T09:45:14.000Z",
|
|
}
|
|
|
|
def get_job(self, _job_id: str) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "gaussian-pipeline.job/v1",
|
|
"job_id": "gsp-20260826000000-deadbeef",
|
|
"state": "ready",
|
|
"created_at_utc": "2026-08-26T09:45:14.000Z",
|
|
"updated_at_utc": "2026-08-26T10:26:57.000Z",
|
|
"progress": {"completed_steps": 4, "total_steps": 4},
|
|
}
|
|
|
|
def get_result(self, _job_id: str) -> dict[str, object]:
|
|
return {
|
|
"schema_version": "gaussian-pipeline.build-result/v1",
|
|
"job_id": "gsp-20260826000000-deadbeef",
|
|
"runtime": {
|
|
"source_revision": "e" * 40,
|
|
"image_digest": f"sha256:{'f' * 64}",
|
|
},
|
|
"artifacts": [
|
|
{
|
|
"role": "preview",
|
|
"logical_path": "preview.sog",
|
|
"media_type": "application/octet-stream",
|
|
"sha256": "1" * 64,
|
|
"byte_length": 7,
|
|
},
|
|
{
|
|
"role": "stream-manifest",
|
|
"logical_path": "streamed/lod-meta.json",
|
|
"media_type": "application/json",
|
|
"sha256": "2" * 64,
|
|
"byte_length": 2,
|
|
},
|
|
],
|
|
}
|
|
|
|
def download_artifact(
|
|
self,
|
|
_job_id: str,
|
|
descriptor: dict[str, object],
|
|
destination: Path,
|
|
) -> Path:
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_bytes(b"preview" if descriptor["role"] == "preview" else b"{}")
|
|
return destination
|
|
|
|
def delete_job(self, job_id: str) -> None:
|
|
self.deleted.append(job_id)
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
|
|
def test_service_builds_visual_world_without_automatic_collision(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="MOJJOKER",
|
|
scene_type="interior",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
provider = _ReadyProvider()
|
|
service = SimulationProjectService(store, provider_factory=lambda: provider) # type: ignore[arg-type]
|
|
service.process(project["project_id"])
|
|
|
|
ready = store.get(project["project_id"])
|
|
assert ready["status"] == "ready"
|
|
assert ready["provider"]["job_created_at_utc"] == "2026-08-26T09:45:14.000Z"
|
|
assert ready["world_manifest"]["visual"]["preview_sog_url"].endswith("/preview.sog")
|
|
assert ready["world_manifest"]["collision"]["available"] is False
|
|
assert ready["world_manifest"]["transforms"]["world_from_visual"] == [
|
|
1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
-1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
-1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
]
|
|
assert ready["world_manifest"]["transforms"]["world_from_collision"] == [
|
|
-1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
-1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
]
|
|
assert provider.upload_calls == 1
|
|
assert provider.submit_calls == 1
|
|
assert provider.submitted_document is not None
|
|
assert provider.submitted_document["outputs"] == {
|
|
"preview_sog": True,
|
|
"streamed_sog": True,
|
|
"collision": False,
|
|
}
|
|
assert provider.submitted_document["collision_profile"] is None
|
|
service.delete(project["project_id"])
|
|
assert provider.deleted == ["gsp-20260826000000-deadbeef"]
|
|
|
|
|
|
def test_service_queue_processes_projects_strictly_one_at_a_time(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
projects: list[dict[str, Any]] = []
|
|
for name in ("First scene", "Second scene"):
|
|
project = store.create(
|
|
name=name,
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
projects.append(store.begin_build(project["project_id"]))
|
|
|
|
service = SimulationProjectService(store, provider_factory=lambda: None)
|
|
first_started = Event()
|
|
release_first = Event()
|
|
second_started = Event()
|
|
completed = Event()
|
|
order: list[str] = []
|
|
|
|
def process(project_id: str) -> None:
|
|
order.append(project_id)
|
|
if project_id == projects[0]["project_id"]:
|
|
first_started.set()
|
|
assert release_first.wait(1.0)
|
|
else:
|
|
second_started.set()
|
|
completed.set()
|
|
|
|
service.process = process # type: ignore[method-assign]
|
|
service.enqueue(projects[0]["project_id"])
|
|
assert first_started.wait(1.0)
|
|
service.enqueue(projects[1]["project_id"])
|
|
assert not second_started.wait(0.05)
|
|
release_first.set()
|
|
assert completed.wait(1.0)
|
|
assert order == [projects[0]["project_id"], projects[1]["project_id"]]
|
|
|
|
|
|
def test_service_deletes_a_queued_project_before_worker_submission(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Queued scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
service = SimulationProjectService(store, provider_factory=lambda: None)
|
|
|
|
service.delete(project["project_id"])
|
|
|
|
with pytest.raises(SimulationProjectNotFoundError, match="unavailable"):
|
|
store.get(project["project_id"])
|
|
|
|
|
|
def test_world_manifest_prefers_combined_rover_collision_world() -> None:
|
|
manifest = _world_manifest(
|
|
"sim-" + "a" * 32,
|
|
[
|
|
{
|
|
"role": "collision-mesh",
|
|
"logical_path": "collision/scene.collision.glb",
|
|
},
|
|
{
|
|
"role": "rover-terrain-mesh",
|
|
"logical_path": "rover/terrain.collision.glb",
|
|
},
|
|
{
|
|
"role": "rover-collision-mesh",
|
|
"logical_path": "rover/scene.collision.glb",
|
|
},
|
|
],
|
|
)
|
|
|
|
assert manifest["collision"]["available"] is True
|
|
assert manifest["collision"]["mesh_url"].endswith("/artifacts/rover/scene.collision.glb")
|
|
assert manifest["transforms"]["world_from_collision"] == [
|
|
1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
1,
|
|
]
|
|
|
|
|
|
def test_service_resumes_a_persisted_provider_job_without_reupload(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Restart-safe scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_job_id="gsp-20260826000000-deadbeef",
|
|
provider_state="building_streamed_sog",
|
|
bundle_sha256="d" * 64,
|
|
)
|
|
provider = _ReadyProvider()
|
|
service = SimulationProjectService(store, provider_factory=lambda: provider) # type: ignore[arg-type]
|
|
|
|
service.process(project["project_id"])
|
|
|
|
assert store.get(project["project_id"])["status"] == "ready"
|
|
assert provider.upload_calls == 0
|
|
assert provider.submit_calls == 0
|
|
|
|
|
|
def test_failed_project_retries_from_retained_source_and_releases_old_job(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Retry scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_job_id="gsp-20260826000000-deadbeef",
|
|
provider_state="failed",
|
|
progress={"completed_steps": 1, "total_steps": 4},
|
|
bundle_sha256="d" * 64,
|
|
)
|
|
store.fail(project["project_id"], "provider failure")
|
|
provider = _ReadyProvider()
|
|
service = SimulationProjectService(store, provider_factory=lambda: provider) # type: ignore[arg-type]
|
|
|
|
queued = service.begin_build(project["project_id"])
|
|
|
|
assert provider.deleted == ["gsp-20260826000000-deadbeef"]
|
|
assert queued["status"] == "queued"
|
|
assert queued["error"] is None
|
|
assert queued["provider"]["job_id"] is None
|
|
assert queued["source"]["uploaded_byte_length"] == queued["source"]["total_byte_length"]
|
|
|
|
|
|
def test_failed_local_project_reattaches_to_live_provider_job_without_rebuild(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Reconnect scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_job_id="gsp-20260826000000-deadbeef",
|
|
provider_state="building_streamed_sog",
|
|
progress={"completed_steps": 3, "total_steps": 5},
|
|
bundle_sha256="d" * 64,
|
|
)
|
|
store.fail(project["project_id"], "temporary provider transport failure")
|
|
provider = _ReadyProvider()
|
|
service = SimulationProjectService(
|
|
store,
|
|
provider_factory=lambda: provider,
|
|
) # type: ignore[arg-type]
|
|
|
|
resumed = service.begin_build(project["project_id"])
|
|
|
|
assert resumed["status"] == "processing"
|
|
assert resumed["error"] is None
|
|
assert resumed["provider"]["job_id"] == "gsp-20260826000000-deadbeef"
|
|
assert resumed["provider"]["state"] == "ready"
|
|
assert provider.deleted == []
|
|
assert provider.upload_calls == 0
|
|
assert provider.submit_calls == 0
|
|
|
|
|
|
def test_live_provider_reattach_retries_temporary_unavailability(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Flaky tunnel scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
store.update_processing(
|
|
project["project_id"],
|
|
status="processing",
|
|
provider_job_id="gsp-20260826000000-deadbeef",
|
|
provider_state="building_collision",
|
|
bundle_sha256="d" * 64,
|
|
)
|
|
store.fail(project["project_id"], "temporary provider transport failure")
|
|
|
|
class _FlakyProvider(_ReadyProvider):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.get_job_calls = 0
|
|
|
|
def get_job(self, job_id: str) -> dict[str, object]:
|
|
self.get_job_calls += 1
|
|
if self.get_job_calls < 3:
|
|
raise GaussianPipelineUnavailableError("temporary tunnel failure")
|
|
return super().get_job(job_id)
|
|
|
|
provider = _FlakyProvider()
|
|
monkeypatch.setattr("k1link.simulation.projects.time.sleep", lambda _delay: None)
|
|
service = SimulationProjectService(
|
|
store,
|
|
provider_factory=lambda: provider,
|
|
) # type: ignore[arg-type]
|
|
|
|
resumed = service.begin_build(project["project_id"])
|
|
|
|
assert resumed["status"] == "processing"
|
|
assert provider.get_job_calls == 3
|
|
assert provider.deleted == []
|
|
|
|
|
|
def test_ready_project_rebuilds_from_retained_source_and_preserves_viewer_settings(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Rebuild scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
_upload_all(store, project)
|
|
store.begin_build(project["project_id"])
|
|
provider = _ReadyProvider()
|
|
service = SimulationProjectService(
|
|
store,
|
|
provider_factory=lambda: provider,
|
|
) # type: ignore[arg-type]
|
|
service.process(project["project_id"])
|
|
settings = {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v3",
|
|
"quality": "medium",
|
|
"visual": {"rotation_degrees": {"x": 0, "y": 0, "z": 90}},
|
|
"collision": {"rotation_degrees": {"x": 0, "y": -90, "z": 0}},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
"ugv": _default_ugv_settings(),
|
|
}
|
|
store.update_viewer_settings(project["project_id"], settings)
|
|
|
|
queued = service.begin_build(project["project_id"])
|
|
|
|
assert provider.deleted == ["gsp-20260826000000-deadbeef"]
|
|
assert queued["status"] == "queued"
|
|
assert queued["viewer_settings"] == settings
|
|
assert queued["source"]["uploaded_byte_length"] == queued["source"]["total_byte_length"]
|
|
|
|
|
|
def test_api_exposes_same_origin_resumable_upload_contract(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
service = SimulationProjectService(store, provider_factory=lambda: None)
|
|
app = FastAPI()
|
|
app.include_router(build_simulation_projects_router(store=store, service=service))
|
|
client = TestClient(app)
|
|
created = client.post(
|
|
"/api/v1/simulation-worlds/projects",
|
|
json={
|
|
"schema_version": "missioncore.simulation-project-create/v1",
|
|
"name": "Archive scene",
|
|
"scene_type": "outdoor",
|
|
"source_kind": "archive",
|
|
"files": [{"logical_path": "scan.rar", "byte_length": 6}],
|
|
},
|
|
)
|
|
assert created.status_code == 201
|
|
project = created.json()
|
|
assert project["viewer_settings"] == {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v3",
|
|
"quality": "high",
|
|
"visual": {"rotation_degrees": {"x": 180.0, "y": 0.0, "z": 0.0}},
|
|
"collision": {"rotation_degrees": {"x": 0.0, "y": 0.0, "z": 0.0}},
|
|
"camera": {"invert_horizontal": True, "invert_vertical": False},
|
|
"ugv": _default_ugv_settings(),
|
|
}
|
|
source_file = project["source"]["files"][0]
|
|
upload_url = (
|
|
f"/api/v1/simulation-worlds/projects/{project['project_id']}"
|
|
f"/source/{source_file['file_id']}"
|
|
)
|
|
assert client.head(upload_url).headers["upload-offset"] == "0"
|
|
first = client.patch(upload_url, headers={"Upload-Offset": "0"}, content=b"abc")
|
|
assert first.status_code == 204
|
|
assert first.headers["upload-offset"] == "3"
|
|
mismatch = client.patch(upload_url, headers={"Upload-Offset": "0"}, content=b"def")
|
|
assert mismatch.status_code == 409
|
|
second = client.patch(upload_url, headers={"Upload-Offset": "3"}, content=b"def")
|
|
assert second.status_code == 204
|
|
catalog = client.get("/api/v1/simulation-worlds/projects").json()
|
|
assert catalog["projects"][0]["project_id"] == project["project_id"]
|
|
|
|
settings = {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v3",
|
|
"quality": "medium",
|
|
"visual": {"rotation_degrees": {"x": 0, "y": 0, "z": 90}},
|
|
"collision": {"rotation_degrees": {"x": -90, "y": 0, "z": 0}},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
"ugv": {
|
|
**_default_ugv_settings(),
|
|
"mass_kg": 1_000_000_000,
|
|
"max_speed_mps": 250.0,
|
|
"max_turn_rate_degrees": 720.0,
|
|
"invert_steering": True,
|
|
},
|
|
}
|
|
saved = client.put(
|
|
f"/api/v1/simulation-worlds/projects/{project['project_id']}/viewer-settings",
|
|
json=settings,
|
|
)
|
|
assert saved.status_code == 200
|
|
assert saved.json()["viewer_settings"] == settings
|
|
persisted = client.get(f"/api/v1/simulation-worlds/projects/{project['project_id']}").json()
|
|
assert persisted["viewer_settings"] == settings
|
|
|
|
|
|
def test_catalog_migrates_legacy_layer_inversions_to_euler_rotations(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Legacy scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
project_path = (
|
|
tmp_path / "simulation-worlds" / "projects" / project["project_id"] / "project.json"
|
|
)
|
|
document = json.loads(project_path.read_text(encoding="utf-8"))
|
|
document["viewer_settings"] = {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v1",
|
|
"quality": "ultra",
|
|
"visual": {"inverted": True, "axis": "z"},
|
|
"collision": {"inverted": False, "axis": "y"},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
}
|
|
project_path.write_text(json.dumps(document), encoding="utf-8")
|
|
|
|
migrated = store.get(project["project_id"])["viewer_settings"]
|
|
|
|
assert migrated == {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v3",
|
|
"quality": "ultra",
|
|
"visual": {"rotation_degrees": {"x": 0.0, "y": 0.0, "z": 180.0}},
|
|
"collision": {"rotation_degrees": {"x": 0.0, "y": 0.0, "z": 0.0}},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
"ugv": _default_ugv_settings(),
|
|
}
|
|
|
|
|
|
def test_catalog_adds_default_ugv_preset_to_v2_viewer_settings(tmp_path: Path) -> None:
|
|
store = SimulationProjectStore(tmp_path)
|
|
project = store.create(
|
|
name="Current scene",
|
|
scene_type="outdoor",
|
|
source_kind="folder",
|
|
files=_folder_files(),
|
|
)
|
|
project_path = (
|
|
tmp_path / "simulation-worlds" / "projects" / project["project_id"] / "project.json"
|
|
)
|
|
document = json.loads(project_path.read_text(encoding="utf-8"))
|
|
document["viewer_settings"] = {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v2",
|
|
"quality": "medium",
|
|
"visual": {"rotation_degrees": {"x": -180.0, "y": 0.0, "z": 0.0}},
|
|
"collision": {"rotation_degrees": {"x": -90.0, "y": 0.0, "z": 0.0}},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
}
|
|
project_path.write_text(json.dumps(document), encoding="utf-8")
|
|
|
|
migrated = store.get(project["project_id"])["viewer_settings"]
|
|
|
|
assert migrated == {
|
|
"schema_version": "missioncore.simulation-viewer-settings/v3",
|
|
"quality": "medium",
|
|
"visual": {"rotation_degrees": {"x": -180.0, "y": 0.0, "z": 0.0}},
|
|
"collision": {"rotation_degrees": {"x": -90.0, "y": 0.0, "z": 0.0}},
|
|
"camera": {"invert_horizontal": False, "invert_vertical": True},
|
|
"ugv": _default_ugv_settings(),
|
|
}
|