Start optillm server for testing Returns tuple of (process_handle, actual_port_used)
(model: str = TEST_MODEL, port: Optional[int] = None)
| 42 | return OpenAI(api_key="optillm", base_url=base_url) |
| 43 | |
| 44 | def start_test_server(model: str = TEST_MODEL, port: Optional[int] = None) -> Tuple[subprocess.Popen, int]: |
| 45 | """ |
| 46 | Start optillm server for testing |
| 47 | Returns tuple of (process_handle, actual_port_used) |
| 48 | """ |
| 49 | if port is None: |
| 50 | port = find_free_port() |
| 51 | |
| 52 | # Set environment for local inference |
| 53 | env = os.environ.copy() |
| 54 | env["OPTILLM_API_KEY"] = "optillm" |
| 55 | |
| 56 | # Pass HF_TOKEN if available (needed for model downloads in CI) |
| 57 | if "HF_TOKEN" in os.environ: |
| 58 | env["HF_TOKEN"] = os.environ["HF_TOKEN"] |
| 59 | |
| 60 | print(f"Starting optillm server on port {port}...") |
| 61 | |
| 62 | # Start server (don't capture output to avoid pipe buffer deadlock) |
| 63 | proc = subprocess.Popen([ |
| 64 | "optillm", |
| 65 | "--model", model, |
| 66 | "--port", str(port) |
| 67 | ], env=env) |
| 68 | |
| 69 | # Wait for server to start |
| 70 | for i in range(30): |
| 71 | try: |
| 72 | response = requests.get(f"http://localhost:{port}/health", timeout=2) |
| 73 | if response.status_code == 200: |
| 74 | print(f"✅ optillm server started successfully on port {port}") |
| 75 | return proc, port |
| 76 | except Exception as e: |
| 77 | if i < 5: # Only print for first few attempts to avoid spam |
| 78 | print(f"Attempt {i+1}: Waiting for server... ({e})") |
| 79 | pass |
| 80 | time.sleep(1) |
| 81 | |
| 82 | # Server didn't start in time - clean up |
| 83 | error_msg = f"optillm server failed to start on port {port}" |
| 84 | print(f"❌ {error_msg} - check that optillm is installed and model is available") |
| 85 | |
| 86 | # Clean up |
| 87 | try: |
| 88 | proc.terminate() |
| 89 | proc.wait(timeout=5) |
| 90 | except subprocess.TimeoutExpired: |
| 91 | proc.kill() |
| 92 | proc.wait() |
| 93 | |
| 94 | raise RuntimeError(error_msg) |
| 95 | |
| 96 | def stop_test_server(proc: subprocess.Popen): |
| 97 | """Stop the test server""" |
no test coverage detected