from __future__ import annotations import json from pathlib import Path from threading import Event from time import monotonic, sleep 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, *, source_mesh: bool = False) -> None: self.deleted: list[str] = [] self.upload_calls = 0 self.submit_calls = 0 self.submitted_document: dict[str, object] | None = None self.source_mesh = source_mesh 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), ] if self.source_mesh: members.append( GaussianSourceMemberUpload( "upload-4", "export/Mesh_Files/scene.ply", "e" * 64, 3, ) ) return GaussianSourceBundleUpload( format=source_format, entrypoint=entrypoint, bundle_sha256="d" * 64, total_byte_length=sum(member.byte_length for member in members), members=tuple(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]: 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, }, ] if self.source_mesh: artifacts.extend([ { "role": "collision-mesh", "logical_path": "collision/scene.collision.glb", "media_type": "model/gltf-binary", "sha256": "3" * 64, "byte_length": 3, }, { "role": "collision-repair-report", "logical_path": "collision/scene.repair.json", "media_type": "application/json", "sha256": "4" * 64, "byte_length": 2, }, ]) 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": artifacts, } def download_artifact( self, _job_id: str, descriptor: dict[str, object], destination: Path, ) -> Path: destination.parent.mkdir(parents=True, exist_ok=True) payload = { "preview": b"preview", "stream-manifest": b"{}", "collision-mesh": b"glb", "collision-repair-report": b"{}", }[str(descriptor["role"])] destination.write_bytes(payload) 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_automatically_builds_repaired_source_mesh_collision(tmp_path: Path) -> None: store = SimulationProjectStore(tmp_path) files = [ *_folder_files(), {"logical_path": "export/Mesh_Files/scene.ply", "byte_length": 3}, ] project = store.create( name="Source mesh scene", scene_type="outdoor", source_kind="folder", files=files, ) payloads = { "export/scene.lcc": b'{"fileType":"Portable"}', "export/index.bin": b"index", "export/data.bin": b"data", "export/Mesh_Files/scene.ply": b"ply", } 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"]], ) store.begin_build(project["project_id"]) provider = _ReadyProvider(source_mesh=True) 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["world_manifest"]["collision"]["available"] is True assert ready["world_manifest"]["collision"]["mesh_url"].endswith( "/collision/scene.collision.glb" ) assert provider.submitted_document is not None assert provider.submitted_document["outputs"] == { "preview_sog": True, "streamed_sog": True, "collision": True, } assert provider.submitted_document["collision_profile"] == { "scene_type": "outdoor", "seed_position": [0.0, 0.0, 0.0], "capsule_height": 0.4, "capsule_radius": 0.4, "voxel_size": 0.05, "mesh_shape": "source", } 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_keeps_retained_source_queued_across_temporary_worker_outage( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: store = SimulationProjectStore(tmp_path) project = store.create( name="Reconnect without browser reupload", scene_type="outdoor", source_kind="folder", files=_folder_files(), ) _upload_all(store, project) store.begin_build(project["project_id"]) class _ReconnectProvider(_ReadyProvider): def __init__(self) -> None: super().__init__() self.capability_calls = 0 def capabilities(self) -> dict[str, object]: self.capability_calls += 1 if self.capability_calls == 1: raise GaussianPipelineUnavailableError("temporary tunnel failure") return super().capabilities() provider = _ReconnectProvider() monkeypatch.setattr( "k1link.simulation.projects.PROVIDER_UNAVAILABLE_RETRY_LIMIT", 1, ) service = SimulationProjectService( store, provider_factory=lambda: provider, ) # type: ignore[arg-type] service.enqueue(project["project_id"]) deadline = monotonic() + 1.0 while store.get(project["project_id"])["status"] != "ready" and monotonic() < deadline: sleep(0.01) recovered = store.get(project["project_id"]) assert recovered["status"] == "ready" assert recovered["error"] is None assert recovered["source"]["uploaded_byte_length"] == recovered["source"]["total_byte_length"] assert provider.capability_calls == 2 assert provider.upload_calls == 1 assert provider.submit_calls == 1 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_import_evicts_only_worker_backed_archive_staging_when_disk_is_low( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: store = SimulationProjectStore(tmp_path) project = store.create( name="Worker-backed archive", scene_type="outdoor", source_kind="archive", files=[{"logical_path": "scene.rar", "byte_length": 6}], ) source_file = project["source"]["files"][0] store.append_upload( project["project_id"], source_file["file_id"], offset=0, payload=b"source", ) store.begin_build(project["project_id"]) store.update_processing( project["project_id"], status="processing", provider_job_id="gsp-20260826000000-deadbeef", provider_state="ready", ) service = SimulationProjectService(store, provider_factory=lambda: None) source_path = store.source_root(project["project_id"]) / "scene.rar" class _DiskUsage: def __init__(self, free: int) -> None: self.free = free monkeypatch.setattr( "k1link.simulation.projects.shutil.disk_usage", lambda _path: _DiskUsage(0 if source_path.exists() else 10 * 1024**3), ) artifacts = [ { "role": "preview", "logical_path": "preview.sog", "media_type": "application/octet-stream", "sha256": "a" * 64, "byte_length": 1024, } ] service._ensure_import_capacity( project["project_id"], artifacts, store.artifacts_root(project["project_id"]), ) assert not source_path.exists() retained_metadata = store.get(project["project_id"])["source"] assert retained_metadata["uploaded_byte_length"] == retained_metadata["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(), }