test(mosaic): wait for broker socket readiness

packages/mosaic/src/lease-broker/recovery_runtime_unittest.py

@@ -31,17 +31,26 @@ PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts"
 
 
 def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
-    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+    deadline = time.monotonic() + 5.0
-        connection.settimeout(3.0)
+    while True:
-        connection.connect(str(socket_path))
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
-        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+            connection.settimeout(3.0)
-        connection.shutdown(socket.SHUT_WR)
+            try:
-        response = bytearray()
+                connection.connect(str(socket_path))
-        while True:
+            except ConnectionRefusedError:
-            chunk = connection.recv(4096)
+                if time.monotonic() >= deadline:
-            if not chunk:
+                    raise
-                break
+                time.sleep(0.02)
-            response.extend(chunk)
+                continue
+            connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+            connection.shutdown(socket.SHUT_WR)
+            response = bytearray()
+            while True:
+                chunk = connection.recv(4096)
+                if not chunk:
+                    break
+                response.extend(chunk)
+        break
     if not response.endswith(b"\n") or response.count(b"\n") != 1:
         raise AssertionError(f"unframed broker response: {bytes(response)!r}")
     reply = json.loads(response[:-1])