diff --git a/docs/AGENTS.md b/docs/AGENTS.md index 9f7f806fa1..769b9578dd 100644 --- a/docs/AGENTS.md +++ b/docs/AGENTS.md @@ -9,6 +9,9 @@ Rules that the docs tree cannot state for itself. Anything that describes how a currently put together belongs next to it instead — in the `.rst`, in the YAML header, in the CSS — where it moves when the code moves. +Keep small API-option updates to purpose, prerequisites, and usage; avoid +example wrappers and lifecycle walkthroughs unless needed to use the option. + ## Build the way CI does `make current-docs` wipes `build/current` and runs `sphinx-build -W --keep-going`, which diff --git a/docs/source/references/cloudxr.rst b/docs/source/references/cloudxr.rst index 658b5314b8..e71b79d900 100644 --- a/docs/source/references/cloudxr.rst +++ b/docs/source/references/cloudxr.rst @@ -276,6 +276,19 @@ container that way. Setting ``CI`` skips the interactive pause described under `Configuration`_, alongside the check for a non-interactive stdin, so automated runs never wait. +.. _cloudxr-deferred-oob-connect: + +Connect the OOB headset after application startup +------------------------------------------------- + +Set ``defer_oob_connect=True`` on ``CloudXRService`` or ``CloudXRLauncher`` +to delay OOB automation and WebSocket signaling. Call ``connect_headset()`` +after the OpenXR session and renderer are ready. + +Requires ``setup_oob=True`` and a fixed ``NV_DEVICE_PROFILE`` (not ``auto*``). +For ``CloudXRLauncher``, also set ``run_embedded=True``. The default ``False`` +keeps immediate connection. + Configuration ------------- diff --git a/src/python/isaaccapture/cloudxr/launcher.py b/src/python/isaaccapture/cloudxr/launcher.py index ed141b1a3e..b7f408cb5c 100644 --- a/src/python/isaaccapture/cloudxr/launcher.py +++ b/src/python/isaaccapture/cloudxr/launcher.py @@ -164,6 +164,7 @@ def __init__( host_client: bool = True, run_embedded: bool = False, start_wss_proxy: bool | None = None, + defer_oob_connect: bool = False, ) -> None: """Attach to the running runtime, or own one when *run_embedded*. @@ -175,6 +176,9 @@ def __init__( to one this process cannot configure. start_wss_proxy: Deprecated no-op; the proxy always starts with the runtime. + defer_oob_connect: Hold OOB automation and backend WebSocket signaling + until :meth:`connect_headset`. Defaults to ``False`` (connect + immediately). Requires ``run_embedded=True`` and ``setup_oob=True``. Every other argument is forwarded to :class:`CloudXRService` and only applies when this process owns it. When attaching they describe a @@ -184,10 +188,16 @@ def __init__( when that is available. Raises: + ValueError: If *defer_oob_connect* is set without *run_embedded* + and *setup_oob*. RuntimeError: If the runtime fails to start or come up, if *run_embedded* is set while a runtime is already serving, or if the requested OOB/USB mode cannot be verified or matched. """ + if defer_oob_connect and not (run_embedded and setup_oob): + raise ValueError( + "defer_oob_connect requires run_embedded=True and setup_oob=True" + ) if start_wss_proxy is not None: self._warn_start_wss_proxy_deprecated() @@ -206,6 +216,7 @@ def __init__( setup_oob=setup_oob, usb_local=usb_local, host_client=host_client, + defer_oob_connect=defer_oob_connect, ) return @@ -846,6 +857,16 @@ def stop(self) -> None: if self._service is not None: self._service.stop() + def connect_headset(self) -> None: + """Release deferred OOB automation once the application's XR session is ready. + + Requires ``defer_oob_connect=True``. Non-blocking and idempotent; + connection progress uses the WSS log as usual. + """ + if self._service is None: + raise RuntimeError("connect_headset requires an embedded CloudXR service") + self._service.connect_headset() + def health_check(self) -> None: """Raise :class:`RuntimeError` if the runtime is no longer available. diff --git a/src/python/isaaccapture/cloudxr/service/_service.py b/src/python/isaaccapture/cloudxr/service/_service.py index 4cf34b58ab..3faa3df3a0 100644 --- a/src/python/isaaccapture/cloudxr/service/_service.py +++ b/src/python/isaaccapture/cloudxr/service/_service.py @@ -119,6 +119,7 @@ def __init__( usb_local: bool = False, host_client: bool = False, recovery_config=None, + defer_oob_connect: bool = False, ) -> None: """Start the CloudXR runtime and the WSS proxy. @@ -148,12 +149,20 @@ def __init__( host_client: Serve the web client at ``/client/`` on the WSS proxy port. Assets are fetched once from GitHub Pages into ``TELEOP_WEB_CLIENT_STATIC_DIR`` or ``~/.cloudxr/static-client``. + defer_oob_connect: Hold OOB automation and backend WebSocket signaling + until :meth:`connect_headset`. Defaults to ``False`` (connect + immediately). Requires *setup_oob*; the runtime and WSS listener + start immediately. Raises: + ValueError: If *defer_oob_connect* is set without *setup_oob*. RuntimeError: If the EULA is not accepted, another runtime is already serving *install_dir*, or the runtime or WSS proxy fails to start within its timeout. """ + if defer_oob_connect and not setup_oob: + raise ValueError("defer_oob_connect requires setup_oob=True") + self._connect_requested = asyncio.Event() if defer_oob_connect else None self._install_dir = install_dir self._env_config = str(env_config) if env_config is not None else None self._device_profile = device_profile @@ -749,6 +758,15 @@ def _terminate_runtime_windows(_proc: subprocess.Popen) -> None: # WSS proxy (background thread with its own event loop) # ------------------------------------------------------------------ + def connect_headset(self) -> None: + """Start deferred headset automation without blocking the caller's frame loop.""" + if self._connect_requested is None: + raise RuntimeError("connect_headset requires defer_oob_connect=True") + self.health_check() + if self._wss_loop is None or self._wss_loop.is_closed(): + raise RuntimeError("CloudXR WSS proxy is not running") + self._wss_loop.call_soon_threadsafe(self._connect_requested.set) + def _start_wss_proxy_thread( self, log_path: Path, timeout_sec: float = WSS_STARTUP_TIMEOUT_SEC ) -> None: @@ -801,6 +819,7 @@ def _on_listening() -> None: recovery_config=self._recovery_config, on_oob_status=self._publish_oob_status, on_oob_fatal=self._on_oob_fatal, + connect_requested=self._connect_requested, ) ) if ( diff --git a/src/python/isaaccapture/cloudxr/wss.py b/src/python/isaaccapture/cloudxr/wss.py index 686e1b6222..c76b271c6e 100755 --- a/src/python/isaaccapture/cloudxr/wss.py +++ b/src/python/isaaccapture/cloudxr/wss.py @@ -566,6 +566,22 @@ def default_cert_paths() -> CertPaths: return cert_paths_from_dir(Path(get_env_config().openxr_run_dir()).parent / "certs") +async def _wait_for_connect( + requested: asyncio.Event | None, stop_future: asyncio.Future +) -> bool: + """Gate headset automation without holding shutdown behind application startup.""" + if requested is not None and not stop_future.done(): + waiter = asyncio.create_task(requested.wait()) + try: + await asyncio.wait( + (waiter, stop_future), return_when=asyncio.FIRST_COMPLETED + ) + finally: + waiter.cancel() + await asyncio.gather(waiter, return_exceptions=True) + return not stop_future.done() + + async def run( log_file_path: str | Path | None, stop_future: asyncio.Future, @@ -579,6 +595,7 @@ async def run( recovery_config=None, on_oob_status: Callable[[dict], object] | None = None, on_oob_fatal: Callable[[Exception], object] | None = None, + connect_requested: asyncio.Event | None = None, ) -> None: """Start the WSS proxy server and run until *stop_future* is resolved. @@ -662,13 +679,27 @@ async def report_fatal(error: Exception, circumstance: str) -> None: initial, ) - def handler(ws): - """Route an incoming WebSocket to the OOB hub or the backend proxy.""" + async def handler(ws) -> None: + """Route to the hub or proxy; WebSocket handlers do not return a value.""" if hub is not None: path = _normalize_request_path(ws.request.path or "/") if path == OOB_WS_PATH: - return hub.handle_connection(ws) - return proxy_handler(ws, backend_host, backend_port) + await hub.handle_connection(ws) + return + # An already-open headset page may reconnect before ADB automation. + ready = asyncio.create_task( + _wait_for_connect(connect_requested, stop_future) + ) + closed = asyncio.create_task(ws.wait_closed()) + try: + await asyncio.wait((ready, closed), return_when=asyncio.FIRST_COMPLETED) + # Disconnected clients must not forward queued offers when the gate opens. + if not closed.done() and ready.result(): + await proxy_handler(ws, backend_host, backend_port) + finally: + ready.cancel() + closed.cancel() + await asyncio.gather(ready, closed, return_exceptions=True) # /client/ on this WSS port for --host-client and --usb-local (same # files; USB-local reaches them via adb reverse of PROXY_PORT). @@ -709,7 +740,11 @@ def handler(ws): if on_listening is not None: on_listening() was_listening = True - if setup_oob and not os.getenv("TELEOP_OOB_HUB_ONLY"): + if ( + setup_oob + and not os.getenv("TELEOP_OOB_HUB_ONLY") + and await _wait_for_connect(connect_requested, stop_future) + ): from .oob_teleop_env import resolve_oob_recovery_config lifecycle = OobLifecycle( diff --git a/src/python/isaaccapture/teleop_session_manager/config.py b/src/python/isaaccapture/teleop_session_manager/config.py index c51d587b4f..8307a43af9 100644 --- a/src/python/isaaccapture/teleop_session_manager/config.py +++ b/src/python/isaaccapture/teleop_session_manager/config.py @@ -10,7 +10,7 @@ from __future__ import annotations import math -from collections.abc import Sequence +from collections.abc import Callable, Sequence from dataclasses import dataclass, field from enum import Enum from pathlib import Path @@ -382,6 +382,11 @@ class TeleopSessionConfig: trackers: Optional list of manual trackers (usually not needed - auto-discovered!) plugins: List of plugin configurations verbose: Whether to print detailed progress information during setup + oxr_session_factory: Optional callback taking the complete required-extension + list and returning external OpenXR handles. Called before DeviceIO starts. + The caller owns the session and must stop rendering before context exit, + then destroy it after exit. Requires LIVE mode; mutually exclusive with + oxr_handles and joint_publisher. oxr_handles: Optional pre-existing OpenXRSessionHandles from an external runtime (e.g. Kit's XR system). When provided, TeleopSession will use these handles instead of creating its own OpenXR session via OpenXRSession.create(). @@ -472,9 +477,20 @@ class TeleopSessionConfig: ) joint_publisher: Optional["RobotTwinPublisher"] = None twin_render: TwinRenderConfig = field(default_factory=TwinRenderConfig) + # Called with all tracker extensions before DeviceIO starts. The caller owns + # the returned session and must stop rendering before TeleopSession exits. + oxr_session_factory: Optional[Callable[[List[str]], OpenXRSessionHandles]] = None def __post_init__(self) -> None: """Validate configuration consistency.""" + if self.oxr_session_factory is not None and ( + self.mode != SessionMode.LIVE + or self.oxr_handles is not None + or self.joint_publisher is not None + ): + raise ValueError( + "oxr_session_factory requires LIVE mode without oxr_handles or joint_publisher" + ) if self.mode == SessionMode.REPLAY and self.mcap_config is None: raise ValueError("mcap_config is required when mode is SessionMode.REPLAY") diff --git a/src/python/isaaccapture/teleop_session_manager/teleop_session.py b/src/python/isaaccapture/teleop_session_manager/teleop_session.py index f22cc7faae..9645931141 100644 --- a/src/python/isaaccapture/teleop_session_manager/teleop_session.py +++ b/src/python/isaaccapture/teleop_session_manager/teleop_session.py @@ -1087,10 +1087,14 @@ def _enter_resources(self, stack: ExitStack) -> None: self._resolved_plugins = [] live_mode = self.config.mode == SessionMode.LIVE + external_openxr = ( + self.config.oxr_handles is not None + or self.config.oxr_session_factory is not None + ) self._status_monitor.rebuild( (), include_openxr=live_mode, - external_openxr=live_mode and self.config.oxr_handles is not None, + external_openxr=live_mode and external_openxr, ) if live_mode: self._resolve_monitored_plugins() @@ -1101,7 +1105,7 @@ def _enter_resources(self, stack: ExitStack) -> None: if spec is not None ), include_openxr=True, - external_openxr=self.config.oxr_handles is not None, + external_openxr=external_openxr, ) # Auto-populate mcap_config from pipeline sources if recording or replaying. @@ -1165,6 +1169,8 @@ def _add_tracker(tracker: Any) -> None: # Resolve OpenXR handles if self.config.oxr_handles is not None: handles = self.config.oxr_handles + elif self.config.oxr_session_factory is not None: + handles = self.config.oxr_session_factory(required_extensions) elif self.config.joint_publisher is not None: handles = oxr.OpenXRSessionHandles( *self._start_twin(stack, required_extensions) diff --git a/tests/python/core/cloudxr/conftest.py b/tests/python/core/cloudxr/conftest.py index 8fb3dd1dac..0a8592a9e3 100644 --- a/tests/python/core/cloudxr/conftest.py +++ b/tests/python/core/cloudxr/conftest.py @@ -184,6 +184,11 @@ def mock_service_deps(tmp_path, ready=True, wss=True): patch( "isaaccapture.cloudxr.service._service.check_eula", ) as m_eula, + # The worker is mocked, so lifecycle tests need no installed native SDK. + patch( + "isaaccapture.cloudxr.service._service.get_sdk_path", + return_value=str(tmp_path / "sdk"), + ), patch( "isaaccapture.cloudxr.service._service.wait_for_runtime_ready_sync", return_value=ready, diff --git a/tests/python/core/cloudxr/test_launcher.py b/tests/python/core/cloudxr/test_launcher.py index 3a58ff15bf..840b1ce552 100644 --- a/tests/python/core/cloudxr/test_launcher.py +++ b/tests/python/core/cloudxr/test_launcher.py @@ -141,6 +141,32 @@ def test_run_embedded_is_refused(self, tmp_path): mocks["from_args"].assert_not_called() mocks["popen"].assert_not_called() + @pytest.mark.parametrize( + "options", [{}, {"run_embedded": True}, {"setup_oob": True}] + ) + def test_deferred_connect_requires_an_owned_oob_service(self, options): + with pytest.raises(ValueError, match="defer_oob_connect requires"): + CloudXRLauncher(defer_oob_connect=True, **options) + + def test_deferred_connect_delegates_to_embedded_service(self): + with ( + _live(False), + patch("isaaccapture.cloudxr.launcher.CloudXRService") as service, + ): + launcher = CloudXRLauncher( + run_embedded=True, setup_oob=True, defer_oob_connect=True + ) + assert service.call_args.kwargs["defer_oob_connect"] is True + launcher.connect_headset() + service.return_value.connect_headset.assert_called_once_with() + + def test_connect_cannot_control_a_borrowed_service(self, tmp_path): + install = _env_file(tmp_path, XR_RUNTIME_JSON="/x/openxr.json") + with _live(): + launcher = CloudXRLauncher(install_dir=install) + with pytest.raises(RuntimeError, match="embedded"): + launcher.connect_headset() + def test_stop_does_not_touch_a_borrowed_runtime(self, tmp_path): install = _env_file(tmp_path, XR_RUNTIME_JSON="/x/openxr.json") with _live(): diff --git a/tests/python/core/cloudxr/test_service.py b/tests/python/core/cloudxr/test_service.py index 9a32069390..d32eea5385 100644 --- a/tests/python/core/cloudxr/test_service.py +++ b/tests/python/core/cloudxr/test_service.py @@ -11,6 +11,7 @@ import socket import subprocess import sys +import threading import types from pathlib import Path from unittest.mock import MagicMock, patch @@ -467,6 +468,60 @@ def test_a_proxy_that_never_listens_times_out(self, tmp_path): # The thread outlives the timeout; stop() is what reaps it. service._stop_wss_proxy() + @pytest.mark.parametrize("defer_oob_connect", [False, True]) + def test_headset_connection_is_explicit_only_when_deferred( + self, tmp_path, defer_oob_connect + ): + from cloudxr_py_test_ns.wss import _wait_for_connect + + connected = threading.Event() + + async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): + on_listening() + if await _wait_for_connect(connect_requested, stop_future): + connected.set() + await stop_future + + with ( + mock_service_deps(tmp_path, wss=False), + _stub_wss(proxy), + patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), + ): + service = CloudXRService( + setup_oob=True, defer_oob_connect=defer_oob_connect + ) + try: + if defer_oob_connect: + assert not connected.wait(0.05) + service.connect_headset() + service.connect_headset() # Releasing the same gate is harmless. + assert connected.wait(1) + finally: + service._stop_wss_proxy() + + def test_stopping_before_connect_does_not_launch_headset(self, tmp_path): + from cloudxr_py_test_ns.wss import _wait_for_connect + + connected = [] + + async def proxy(*, stop_future, on_listening, connect_requested, **kwargs): + on_listening() + if await _wait_for_connect(connect_requested, stop_future): + connected.append(True) + + with ( + mock_service_deps(tmp_path, wss=False), + _stub_wss(proxy), + patch("isaaccapture.cloudxr.oob_teleop_adb.require_adb_on_path"), + ): + service = CloudXRService(setup_oob=True, defer_oob_connect=True) + thread = service._wss_thread + service._stop_wss_proxy() + assert not thread.is_alive() + assert connected == [] + with pytest.raises(RuntimeError, match="not running"): + service.connect_headset() + # ============================================================================ # TestSignalHandlers @@ -578,3 +633,159 @@ def test_sigint_handler_with_sig_ign_prev_is_noop(self, tmp_path): signal.signal(signal.SIGINT, orig_sigint) assert not stop_called, "signal handler must not call stop() directly" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "outcome", ["connect", "shutdown", "disconnect", "disconnect_and_connect", "cancel"] +) +async def test_early_websocket_waits_for_app_or_disconnect(monkeypatch, outcome): + from unittest.mock import AsyncMock + from cloudxr_py_test_ns import wss + + requested = asyncio.Event() + listening = asyncio.Event() + closed = asyncio.Event() + watching_close = asyncio.Event() + close_wait_finished = asyncio.Event() + stop = asyncio.get_running_loop().create_future() + handlers = [] + + async def wait_closed(): + watching_close.set() + try: + await closed.wait() + finally: + close_wait_finished.set() + + @contextlib.asynccontextmanager + async def serve(handler, **kwargs): + handlers.append(handler) + yield + + proxy = AsyncMock() + # Keep this signaling test independent of an installed CloudXR environment. + monkeypatch.setattr(wss, "default_cert_paths", lambda: None) + monkeypatch.setattr(wss, "ensure_certificate", lambda *args: None) + monkeypatch.setattr(wss, "build_ssl_context", lambda *args: None) + monkeypatch.setattr(wss, "ws_serve", serve) + monkeypatch.setattr(wss, "proxy_handler", proxy) + task = asyncio.create_task( + wss.run( + None, + stop, + on_listening=listening.set, + connect_requested=requested, + ) + ) + client = None + try: + await asyncio.wait_for(listening.wait(), timeout=1) + client = asyncio.create_task( + handlers[0](types.SimpleNamespace(wait_closed=wait_closed)) + ) + await asyncio.wait_for(watching_close.wait(), timeout=1) + proxy.assert_not_called() + if outcome in ("disconnect", "disconnect_and_connect"): + closed.set() + if outcome in ("connect", "disconnect_and_connect"): + requested.set() + if outcome == "shutdown": + stop.set_result(None) + if outcome == "cancel": + client.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(client, timeout=1) + else: + await asyncio.wait_for(client, timeout=1) + assert proxy.call_count == int(outcome == "connect") + assert close_wait_finished.is_set() + assert stop.done() == (outcome == "shutdown") + finally: + if client is not None: + client.cancel() + await asyncio.gather(client, return_exceptions=True) + if not stop.done(): + stop.set_result(None) + await asyncio.wait_for(task, timeout=1) + + +@pytest.mark.asyncio +async def test_disconnected_websockets_do_not_forward_stale_offers(monkeypatch): + from websockets.asyncio.client import connect + from websockets.asyncio.server import serve + from cloudxr_py_test_ns import wss + + requested = asyncio.Event() + listening = asyncio.Event() + stop = asyncio.get_running_loop().create_future() + finished = asyncio.Queue() + connections = [] + offers = [] + proxy_port = None + + async def backend_handler(ws): + connections.append(ws.request.path) + async for offer in ws: + offers.append(offer) + await ws.send(offer) + + @contextlib.asynccontextmanager + async def serve_proxy(handler, **kwargs): + nonlocal proxy_port + + async def tracked_handler(ws): + try: + await handler(ws) + finally: + finished.put_nowait(ws.request.path) + + kwargs.update(host="127.0.0.1", port=0) + async with serve(tracked_handler, **kwargs) as server: + proxy_port = server.sockets[0].getsockname()[1] + yield + + monkeypatch.setattr(wss, "default_cert_paths", lambda: None) + monkeypatch.setattr(wss, "ensure_certificate", lambda *args: None) + monkeypatch.setattr(wss, "build_ssl_context", lambda *args: None) + monkeypatch.setattr(wss, "ws_serve", serve_proxy) + + async with serve(backend_handler, "127.0.0.1", 0) as backend: + task = asyncio.create_task( + wss.run( + None, + stop, + backend_host="127.0.0.1", + backend_port=backend.sockets[0].getsockname()[1], + proxy_port=0, + on_listening=listening.set, + connect_requested=requested, + ) + ) + try: + await asyncio.wait_for(listening.wait(), timeout=2) + uri = f"ws://127.0.0.1:{proxy_port}" + for index in range(3): + async with connect(f"{uri}/stale-{index}") as client: + await client.send(f"stale-offer-{index}") + + # Disconnected handlers must exit while the startup gate is still held. + exited = { + await asyncio.wait_for(finished.get(), timeout=2) for _ in range(3) + } + assert exited == {f"/stale-{index}" for index in range(3)} + assert not connections + assert not offers + + async with connect(f"{uri}/live") as client: + await client.send("fresh-offer") + requested.set() + assert await asyncio.wait_for(client.recv(), timeout=2) == "fresh-offer" + assert await asyncio.wait_for(finished.get(), timeout=2) == "/live" + finally: + if not stop.done(): + stop.set_result(None) + await asyncio.wait_for(task, timeout=2) + + assert connections == ["/live"] + assert offers == ["fresh-offer"] diff --git a/tests/python/core/cloudxr/test_service_oob_status.py b/tests/python/core/cloudxr/test_service_oob_status.py index 3a2cc2ef18..24aa0bbe5b 100644 --- a/tests/python/core/cloudxr/test_service_oob_status.py +++ b/tests/python/core/cloudxr/test_service_oob_status.py @@ -34,6 +34,7 @@ def _service_for_status(tmp_path): service._runtime_proc.pid = os.getpid() service._runtime_proc.poll.return_value = None service._wss_thread = None + service._connect_requested = None return service diff --git a/tests/python/core/cloudxr/test_wss_oob_lifecycle.py b/tests/python/core/cloudxr/test_wss_oob_lifecycle.py index 943081186d..294967fee5 100644 --- a/tests/python/core/cloudxr/test_wss_oob_lifecycle.py +++ b/tests/python/core/cloudxr/test_wss_oob_lifecycle.py @@ -18,13 +18,22 @@ @pytest.mark.asyncio -async def test_wss_lifecycle_waits_without_headset_and_cleans_up_in_order( - monkeypatch, tmp_path +@pytest.mark.parametrize("connection_mode", ["immediate", "deferred", "cancelled"]) +async def test_wss_lifecycle_respects_connection_gate_and_cleanup_order( + monkeypatch, tmp_path, connection_mode ): + monkeypatch.delenv("TELEOP_OOB_HUB_ONLY", raising=False) events = [] pending = asyncio.Event() + listening = asyncio.Event() + started = asyncio.Event() + requested = asyncio.Event() if connection_mode != "immediate" else None stop = asyncio.get_running_loop().create_future() + def on_listening(): + events.append("callback") + listening.set() + @asynccontextmanager async def serving(*args, **kwargs): events.append("wss listening") @@ -33,6 +42,7 @@ async def serving(*args, **kwargs): async def lifecycle_run(): events.append("lifecycle started") + started.set() # Match the package loggers that WSS forwards into its session file. logging.getLogger("isaaccapture.cloudxr.oob_teleop_lifecycle").info( "lifecycle test transition" @@ -73,18 +83,28 @@ async def lifecycle_run(): setup_oob=True, usb_local=True, recovery_config=RecoveryConfig(), - on_listening=lambda: events.append("callback"), + on_listening=on_listening, + connect_requested=requested, ) ) - for _ in range(10): - await asyncio.sleep(0) - if "lifecycle started" in events: - break - assert not task.done() - assert events[:2] == ["wss listening", "callback"] - factory.assert_called_once() - stop.set_result(None) - await task + try: + await asyncio.wait_for(listening.wait(), timeout=1) + assert not task.done() + assert events[:2] == ["wss listening", "callback"] + if requested is not None: + factory.assert_not_called() + if connection_mode == "deferred": + requested.set() + if connection_mode != "cancelled": + await asyncio.wait_for(started.wait(), timeout=1) + factory.assert_called_once() + finally: + stop.set_result(None) + await asyncio.wait_for(task, timeout=1) + if connection_mode == "cancelled": + factory.assert_not_called() + assert events == ["wss listening", "callback", "wss closed"] + return assert events.index("lifecycle stopped") < events.index("wss closed") assert "lifecycle test transition" in (tmp_path / "wss.log").read_text() assert "hub test registration" in (tmp_path / "wss.log").read_text() diff --git a/tests/python/core/teleop_session_manager/test_external_oxr_handles.py b/tests/python/core/teleop_session_manager/test_external_oxr_handles.py index 1c7d4942a5..a1537976a7 100644 --- a/tests/python/core/teleop_session_manager/test_external_oxr_handles.py +++ b/tests/python/core/teleop_session_manager/test_external_oxr_handles.py @@ -20,6 +20,7 @@ from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +import pytest from isaaccapture.teleop_session_manager import TeleopSession, TeleopSessionConfig @@ -124,6 +125,54 @@ def test_oxr_handles_accepts_stub(self): class TestTeleopSessionExternalHandles: """Tests for TeleopSession when external OpenXR handles are provided.""" + def test_factory_receives_aggregated_extensions_before_deviceio(self): + handles = _make_stub_handles() + tracker = MagicMock() + factory = MagicMock(return_value=handles) + config = TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + trackers=[tracker], + oxr_session_factory=factory, + ) + with ( + _mock_deviceio_and_oxr() as mocks, + patch( + "isaaccapture.deviceio.DeviceIOSession.get_required_extensions", + return_value=["XR_EXT_hand_tracking", "XR_NVX1_push_tensor"], + ) as extensions, + ): + factory.side_effect = lambda required: ( + mocks.deviceio_run.assert_not_called() or handles + ) + with TeleopSession(config): + assert tracker in extensions.call_args.args[0] + factory.assert_called_once_with(extensions.return_value) + assert mocks.deviceio_run.call_args.args[1] is handles + mocks.oxr_cls.assert_not_called() + + def test_factory_failure_does_not_start_deviceio(self): + config = TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + oxr_session_factory=MagicMock(side_effect=RuntimeError("viz failed")), + ) + with _mock_deviceio_and_oxr() as mocks: + with pytest.raises(RuntimeError, match="viz failed"): + with TeleopSession(config): + pass + mocks.deviceio_run.assert_not_called() + + @pytest.mark.parametrize("field", ["oxr_handles", "joint_publisher"]) + def test_factory_rejects_competing_session_owners(self, field): + with pytest.raises(ValueError, match="oxr_session_factory requires LIVE"): + TeleopSessionConfig( + app_name="test", + pipeline=_make_empty_pipeline(), + oxr_session_factory=MagicMock(), + **{field: MagicMock()}, + ) + def test_skips_oxr_session_creation(self): """When oxr_handles is set, OpenXRSession.create() must not be called.""" external_handles = _make_stub_handles()