Add React Native support to the TDK and environment scripts (2/4)

client/src/cbltest/requests_transport.py

@@ -2,7 +2,7 @@
 from abc import ABC, abstractmethod
 from asyncio import wait_for
 from typing import cast
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse
 from uuid import uuid4
 
 from aiohttp import ClientResponse
@@ -100,6 +100,7 @@ class _RequestWebSocketTransport(RequestTransport):
     def __init__(self, url: str, ws_router: WebSocketRouter) -> None:
         self.__url = url
         self.__ws_router = ws_router
+        self.__is_native_ws = urlparse(url).scheme == "ws"
 
     async def send(self, request: TestServerRequest, message_no: int) -> TestServerResponse:
         if request.payload is not None:
@@ -119,13 +120,19 @@ async def send(self, request: TestServerRequest, message_no: int) -> TestServerR
         if CBLPyTestGlobal.running_test_name is not None:
             data["ts_testName"] = CBLPyTestGlobal.running_test_name
 
+        await self.__ws_router.ensure_connected(self.__url)
+
         future = self.__ws_router.register(message_no)
         ws_conn = self.__ws_router.get_websocket_for_write(self.__url)
         await ws_conn.send_str(json.dumps(data))
         try:
             resp = await wait_for(future, timeout=120)
         except TimeoutError:
             self.__ws_router.unregister(message_no)
+            if self.__is_native_ws:
+                # React Native test servers (ws:// URLs) can be left stuck after a timeout,
+                # so drop the connection and relaunch the app before the next request.
+                self.__ws_router.begin_reconnect(self.__url)
             raise CblTimeoutError(f"No response to {data['ts_command']} from {self.__url} after 120 seconds") from None
 
         resp_version = cast(int, resp.get("ts_apiVersion", 0))

client/src/cbltest/responses.py

@@ -41,7 +41,8 @@ class ServerVariant(Flag):
     IOS = auto()
     JVM = auto()
     JS = auto()
-    ALL = ANDROID | C | DOTNET | IOS | JVM | JS
+    REACTNATIVE = auto()
+    ALL = ANDROID | C | DOTNET | IOS | JVM | JS | REACTNATIVE
 
     def __str__(self) -> str:
         return "|".join([member.name for member in ServerVariant if member in self and member.name is not None])
@@ -133,6 +134,8 @@ def variant(self) -> ServerVariant:
                 return ServerVariant.IOS
             elif self.__cbl == "couchbase-lite-js":
                 return ServerVariant.JS
+            elif self.__cbl == "cbl-reactnative":
+                return ServerVariant.REACTNATIVE
             else:
                 raise ValueError(f"Unknown test server variant: {self.__cbl}")
 

client/src/cbltest/websocket_router.py

@@ -1,15 +1,19 @@
+import asyncio
 import json
+import os
 import socket
+import sys
+import time
 import webbrowser
 from asyncio import Future, Semaphore, wait_for
 from typing import cast
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
 
 import netifaces
 from aiohttp import web
 
 from cbltest.globals import CBLPyTestGlobal
-from cbltest.logging import cbl_error, cbl_info
+from cbltest.logging import cbl_error, cbl_info, cbl_warning
 
 
 class WebSocketRouter:
@@ -22,6 +26,8 @@ def __init__(self, server_urls: list[str]) -> None:
         self.__connections: dict[str, web.WebSocketResponse] = {}
         self.__runner = web.AppRunner(self.__app)
         self.__stopping = False
+        self.__connected_events: dict[str, asyncio.Event] = {url: asyncio.Event() for url in server_urls}
+        self.__reconnect_lock = asyncio.Lock()
 
     async def start(self) -> None:
         if len(self.__server_urls) == 0:
@@ -31,16 +37,46 @@ async def start(self) -> None:
         self.__stopping = False
         await self.__runner.setup()
 
-        # Create a socket so that I can use an OS provided port
+        # Native WebSocket test servers (ws:// URLs, e.g. React Native) connect back to a
+        # fixed port given in their URL. Otherwise let the OS choose one.
+        bind_port = next(
+            (urlparse(u).port or 0 for u in self.__server_urls if urlparse(u).scheme == "ws"),
+            0,
+        )
+
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        sock.bind(("", 0))
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.bind(("", bind_port))
         sock.listen(128)
         site = web.SockSite(self.__runner, sock)
         chosen_port = sock.getsockname()[1]
         await site.start()
+        cbl_info(f"WebSocket router listening on port {chosen_port}")
+
+        relaunch_script = os.environ.get("CBL_NATIVE_WS_RELAUNCH_SCRIPT")
+        has_native_ws = any(urlparse(u).scheme == "ws" for u in self.__server_urls)
+        if has_native_ws and not relaunch_script:
+            cbl_warning(
+                "Native WS test server detected but CBL_NATIVE_WS_RELAUNCH_SCRIPT "
+                "is not set; relying on the app's built-in reconnect logic"
+            )
+        if relaunch_script:
+            cbl_info(f"Relaunching native WS app via: {relaunch_script}")
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable,
+                relaunch_script,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout_bytes, stderr_bytes = await proc.communicate()
+            if stdout_bytes:
+                cbl_info(f"Relaunch script output:\n{stdout_bytes.decode(errors='replace')}")
+            if proc.returncode != 0:
+                err_text = stderr_bytes.decode(errors="replace")
+                raise RuntimeError(f"Native WS relaunch script exited with code {proc.returncode}:\n{err_text}")
+            cbl_info("Native WS app relaunch complete")
 
-        ws_index = 0
-        for url in self.__server_urls:
+        for ws_index, url in enumerate(self.__server_urls):
             local_ip = self._lookup_ip(url)
             cbl_info(f"Connecting to test server at {url}...")
             params = {
@@ -49,12 +85,37 @@ async def start(self) -> None:
                 "device": f"ws{ws_index}",
             }
             query = urlencode(params)
-            timeout = 30
-            if CBLPyTestGlobal.auto_start_tdk_page:
+            parsed_url = urlparse(url)
+            is_native_ws = parsed_url.scheme == "ws"
+            timeout = 90 if is_native_ws else 30
+            if not is_native_ws and CBLPyTestGlobal.auto_start_tdk_page:
                 webbrowser.open_new_tab(f"{url}/tdk.html?{query}")
 
-            await wait_for(self.__conn_sem.acquire(), timeout=timeout)
-            cbl_info(f"Connected to test server at {url}!")
+            t0 = time.monotonic()
+            try:
+                await self._wait_for_connection(timeout)
+            except TimeoutError:
+                cbl_error(f"No connection from {url} after {time.monotonic() - t0:.1f}s (limit {timeout}s)")
+                raise
+            cbl_info(f"Connected to test server at {url} (took {time.monotonic() - t0:.1f}s)")
+
+    async def _wait_for_connection(self, timeout: float) -> None:
+        """Acquire the connection semaphore, logging a heartbeat every 10 s."""
+        deadline = time.monotonic() + timeout
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise TimeoutError
+            chunk = min(10.0, remaining)
+            try:
+                await wait_for(self.__conn_sem.acquire(), timeout=chunk)
+                return
+            except TimeoutError:
+                elapsed = timeout - (deadline - time.monotonic())
+                cbl_info(
+                    f"Still waiting for test server to connect ({elapsed:.0f}s elapsed, "
+                    f"{timeout - elapsed:.0f}s remaining)"
+                )
 
     async def stop(self) -> None:
         self.__stopping = True
@@ -75,6 +136,21 @@ async def stop(self) -> None:
 
         self.__pending.clear()
 
+    async def ensure_connected(self, url: str, timeout: float = 120.0) -> None:
+        """Wait until a live WebSocket connection for *url* is available.
+
+        Returns immediately if already connected; blocks while a reconnect
+        is in progress.
+        """
+        event = self.__connected_events.get(url)
+        if event is None or event.is_set():
+            return
+        cbl_info(f"Waiting up to {timeout:.0f}s for test server to reconnect at {url}")
+        try:
+            await wait_for(event.wait(), timeout=timeout)
+        except TimeoutError:
+            raise ConnectionError(f"Device at {url} did not reconnect within {timeout:.0f} s") from None
+
     def get_websocket_for_write(self, url: str) -> web.WebSocketResponse:
         if not self.is_known_ws_url(url):
             raise IndexError(f"No WebSocket connection for the given URL ({url})")
@@ -88,6 +164,8 @@ async def _websocket_handler(self, request: web.Request) -> web.WebSocketRespons
         ws = web.WebSocketResponse(protocols=request.headers.getall("Sec-WebSocket-Protocol", []))
         await ws.prepare(request)
 
+        current_url: str | None = None
+
         async for msg in ws:
             if self.__stopping:
                 break
@@ -97,29 +175,125 @@ async def _websocket_handler(self, request: web.Request) -> web.WebSocketRespons
                 ts_id = data.get("ts_id")
                 if ts_id is not None and ts_id in self.__pending:
                     future = self.__pending.pop(ts_id)
-                    future.set_result(data)
+                    if not future.done():
+                        future.set_result(data)
                 else:
                     device = data.get("device")
                     if device is not None:
                         try:
                             url_index = int(device[2:])
-                            self.__connections[self.__server_urls[url_index]] = ws
+                            current_url = self.__server_urls[url_index]
+                            self.__connections[current_url] = ws
+                            self.__connected_events[current_url].set()
                             self.__conn_sem.release()
                         except (ValueError, IndexError):
                             cbl_error(f"Unknown or invalid device ID received: {device}")
                             await ws.close(message=b"Unknown or invalid device ID")
             elif msg.type == web.WSMsgType.ERROR:
                 cbl_error(f"WebSocket connection closed with exception {ws.exception()}")
 
+        # ── Connection closed ──────────────────────────────────────────────
+        # Only act if this ws is still THE active connection for the URL.
+        # begin_reconnect() may have already claimed ownership and scheduled
+        # its own _close_and_reconnect, in which case we must not interfere
+        # (especially: we must not clear the connected event after a
+        # successful reconnect has already set it).
+        if current_url is not None and not self.__stopping and self.__connections.get(current_url) is ws:
+            cbl_warning(f"WebSocket connection from {current_url} dropped")
+
+            del self.__connections[current_url]
+            self.__connected_events[current_url].clear()
+
+            self._fail_pending_futures(current_url)
+
+            relaunch_script = os.environ.get("CBL_NATIVE_WS_RELAUNCH_SCRIPT")
+            if relaunch_script:
+                asyncio.ensure_future(self._reconnect(current_url, relaunch_script))
+
         return ws
 
+    # ── Timeout-triggered reconnect (called from send()) ───────────────
+
+    def unregister(self, ts_id: int) -> None:
+        self.__pending.pop(ts_id, None)
+
+    def begin_reconnect(self, url: str) -> None:
+        """Mark *url* as disconnected and schedule WS close + app relaunch.
+
+        This is called **synchronously** from send() after a per-request
+        timeout.  Clearing the connected event HERE (before send() raises
+        ConnectionError) guarantees the next test's ensure_connected() will
+        wait for the reconnect to finish instead of hitting the stale WS.
+        """
+        self.__connected_events[url].clear()
+
+        ws = self.__connections.pop(url, None)
+        cbl_warning(f"Request timeout on {url}; closing the connection and relaunching the test server")
+
+        self._fail_pending_futures(url)
+
+        asyncio.ensure_future(self._close_and_reconnect(url, ws))
+
     def register(self, ts_id: int) -> Future[dict]:
         future: Future[dict] = Future()
         self.__pending[ts_id] = future
         return future
 
-    def unregister(self, ts_id: int) -> None:
-        self.__pending.pop(ts_id, None)
+    # ── Internal helpers ───────────────────────────────────────────────
+
+    def _fail_pending_futures(self, url: str) -> None:
+        pending_ids = list(self.__pending.keys())
+        if pending_ids:
+            cbl_warning(f"Failing {len(pending_ids)} in-flight request(s) for {url}")
+        for ts_id in pending_ids:
+            fut = self.__pending.pop(ts_id)
+            if not fut.done():
+                fut.set_exception(ConnectionError(f"WebSocket connection to {url} lost while waiting for response"))
+
+    async def _close_and_reconnect(self, url: str, ws: web.WebSocketResponse | None) -> None:
+        """Close *ws* (if still open) and relaunch the native app."""
+        if ws is not None and not ws.closed:
+            try:
+                await ws.close(
+                    code=1011,
+                    message=b"Request timed out forcing reconnect",
+                )
+            except Exception:
+                pass
+
+        relaunch_script = os.environ.get("CBL_NATIVE_WS_RELAUNCH_SCRIPT")
+        if relaunch_script:
+            await self._reconnect(url, relaunch_script)
+
+    async def _reconnect(self, url: str, relaunch_script: str) -> None:
+        """Re-launch the native app and wait for it to reconnect."""
+        async with self.__reconnect_lock:
+            if self.__connected_events[url].is_set():
+                return
+
+            cbl_info(f"Reconnecting {url}: running relaunch script")
+            try:
+                proc = await asyncio.create_subprocess_exec(
+                    sys.executable,
+                    relaunch_script,
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.PIPE,
+                )
+                stdout_bytes, stderr_bytes = await proc.communicate()
+                if stdout_bytes:
+                    cbl_info(f"Relaunch script output:\n{stdout_bytes.decode(errors='replace')}")
+                if proc.returncode != 0:
+                    err_text = stderr_bytes.decode(errors="replace")
+                    cbl_error(f"Relaunch script failed (exit {proc.returncode}):\n{err_text}", include_stack=False)
+                    return
+
+                cbl_info(f"Test server relaunched; waiting up to 90s for it to reconnect at {url}")
+                await wait_for(self.__connected_events[url].wait(), timeout=90)
+                cbl_info(f"Test server reconnected at {url}")
+            except TimeoutError:
+                cbl_error(f"Test server at {url} did not reconnect within 90s after relaunch")
+            except Exception as exc:
+                cbl_error(f"Error during reconnect for {url}: {exc}")
 
     def _lookup_ip(self, remote_url: str) -> str:
         if "localhost" in remote_url:

client/tests/test_requests_transport.py

@@ -90,6 +90,7 @@ async def test_bad_response_reads_ts_error(self) -> None:
         reply: asyncio.Future[dict] = asyncio.get_running_loop().create_future()
         reply.set_result({"ts_apiVersion": 1, "ts_serverID": "server-uuid", "ts_error": NOT_FOUND_BODY})
         ws_router = MagicMock()
+        ws_router.ensure_connected = AsyncMock()
         ws_router.register.return_value = reply
         ws_router.get_websocket_for_write.return_value.send_str = AsyncMock()
         transport = _RequestWebSocketTransport("ws://localhost:8080", ws_router)

environment/aws/start_backend.py

@@ -37,8 +37,11 @@
 from environment.aws.logslurp_setup.setup_logslurp import main as logslurp_main
 from environment.aws.server_setup.setup_server import main as server_main
 from environment.aws.sgw_setup.setup_sgw import main as sgw_main
-from environment.aws.topology_setup.setup_topology import TopologyConfig
+from environment.aws.topology_setup.setup_topology import TopologyConfig, websocket_device_ids
 from environment.aws.topology_setup.setup_topology import main as topology_main
+from environment.aws.topology_setup.test_server_platforms.reactnative_register import (
+    WS_PORT as REACT_NATIVE_WS_PORT,
+)
 
 
 class TopologyParamType(click.ParamType):
@@ -169,12 +172,22 @@ def write_config(in_config_file: str, topology: TopologyConfig, output: IO[str])
 
         if len(topology.test_servers) > 0:
             test_servers = []
-            for ts in topology.test_servers:
+            device_ids = websocket_device_ids([ts.platform for ts in topology.test_servers])
+            for ts, device_id in zip(topology.test_servers, device_ids, strict=True):
                 port = 5555 if ts.platform.startswith("dotnet") else 5173 if ts.platform == "js" else 8080
                 ts_definition = {
                     "url": f"http://{ts.ip_address}:{port}",
                 }
 
+                if ts.platform.startswith("reactnative"):
+                    # React Native test servers connect back to the pytest WebSocket router on
+                    # a fixed port. The URL only identifies the server, so it carries the
+                    # device ID that the app announces when it connects.
+                    ts_definition = {
+                        "url": f"ws://localhost:{REACT_NATIVE_WS_PORT}/{device_id}",
+                        "transport": "ws",
+                    }
+
                 if ts.dataset_version:
                     ts_definition["dataset_version"] = ts.dataset_version