Wait for server to become available with retry logic.
(config: TestConfig)
| 44 | |
| 45 | |
| 46 | def wait_for_server(config: TestConfig) -> bool: |
| 47 | """Wait for server to become available with retry logic.""" |
| 48 | url = f"http://{config.host}:{config.port}/ping" |
| 49 | |
| 50 | console.print(f"Waiting for server at {url}...") |
| 51 | |
| 52 | for attempt in range(1, config.max_retries + 1): |
| 53 | try: |
| 54 | with httpx.Client(timeout=config.timeout) as client: |
| 55 | response = client.get(url) |
| 56 | if response.status_code == 200: |
| 57 | console.print("✓ Server is ready!", style="green") |
| 58 | return True |
| 59 | except (httpx.ConnectError, httpx.TimeoutException): |
| 60 | delay = config.retry_delay * (2 ** (attempt - 1)) |
| 61 | console.print( |
| 62 | f" Attempt {attempt}/{config.max_retries}: Not ready (retry in {delay:.1f}s)" |
| 63 | ) |
| 64 | if attempt < config.max_retries: |
| 65 | time.sleep(delay) |
| 66 | except KeyboardInterrupt: |
| 67 | console.print("\nInterrupted by user", style="yellow") |
| 68 | _thread.interrupt_main() |
| 69 | raise |
| 70 | |
| 71 | console.print("✗ Server not reachable", style="red") |
| 72 | return False |
| 73 | |
| 74 | |
| 75 | def send_request(client: httpx.Client, url: str, config: TestConfig) -> RequestResult: |
no test coverage detected