Run StreamableHTTP transport tests in process instead of over sockets (#2767)

This commit is contained in:
Max 2026-06-03 12:45:00 +01:00 • committed by GitHub
commit 19fe9faec8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 969 additions and 1177 deletions

File diff suppressed because it is too large Load diff

View file

@ -2,7 +2,6 @@
import socket
import threading
import time
from collections.abc import Generator
from contextlib import contextmanager
from typing import Any
@ -56,30 +55,3 @@ def run_uvicorn_in_thread(app: Any, **config_kwargs: Any) -> Generator[str, None
finally:
server.should_exit = True
thread.join(timeout=_SERVER_SHUTDOWN_TIMEOUT_S)
def wait_for_server(port: int, timeout: float = 20.0) -> None:
"""Wait for server to be ready to accept connections.
Polls the server port until it accepts connections or timeout is reached.
This eliminates race conditions without arbitrary sleeps.
Args:
port: The port number to check
timeout: Maximum time to wait in seconds (default 5.0)
Raises:
TimeoutError: If server doesn't start within the timeout period
"""
start_time = time.time()
while time.time() - start_time < timeout:
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(0.1)
s.connect(("127.0.0.1", port))
# Server is ready
return
except (ConnectionRefusedError, OSError):
# Server not ready yet, retry quickly
time.sleep(0.01)
raise TimeoutError(f"Server on port {port} did not start within {timeout} seconds") # pragma: no cover