| """Shared pytest fixtures for the Space test suite. |
| |
| Two responsibilities: |
| |
| 1. Put the Space root on ``sys.path`` so unit tests in later commits |
| can ``import leaderboard`` / ``import submit`` directly without a |
| package layout shim. |
| 2. Expose an ``app_url`` session fixture that boots ``app.py`` in a |
| subprocess on a free port, polls until the HTTP server answers, |
| yields the URL, and terminates the process on teardown. The |
| subprocess's stdout + stderr are captured to a tmp log file so a |
| readiness-timeout or early exit surfaces the actual Gradio log. |
| """ |
| from __future__ import annotations |
|
|
| import os |
| import socket |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
|
|
| import pytest |
| import requests |
|
|
| SPACE_ROOT = Path(__file__).resolve().parent.parent |
| sys.path.insert(0, str(SPACE_ROOT)) |
|
|
| APP_BOOT_TIMEOUT_SECONDS = 90 |
|
|
|
|
| def _pick_free_port() -> int: |
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: |
| s.bind(("127.0.0.1", 0)) |
| return s.getsockname()[1] |
|
|
|
|
| def _wait_for_ready(proc: subprocess.Popen, url: str, log_path: Path) -> None: |
| deadline = time.time() + APP_BOOT_TIMEOUT_SECONDS |
| while time.time() < deadline: |
| if proc.poll() is not None: |
| raise RuntimeError( |
| f"app.py exited with code {proc.returncode} before HTTP " |
| f"was ready. Log:\n{log_path.read_text()}" |
| ) |
| try: |
| if requests.get(url, timeout=2).status_code == 200: |
| return |
| except requests.RequestException: |
| pass |
| time.sleep(0.5) |
| raise RuntimeError( |
| f"app.py did not respond on {url} within " |
| f"{APP_BOOT_TIMEOUT_SECONDS}s. Log:\n{log_path.read_text()}" |
| ) |
|
|
|
|
| @pytest.fixture(scope="session") |
| def app_url(tmp_path_factory): |
| port = _pick_free_port() |
| log_path = tmp_path_factory.mktemp("space-smoke") / "app.log" |
| env = { |
| **os.environ, |
| "GRADIO_SERVER_NAME": "127.0.0.1", |
| "GRADIO_SERVER_PORT": str(port), |
| |
| |
| |
| |
| "CADGENBENCH_DISABLE_BOOT_SWEEP": "1", |
| } |
| log_file = log_path.open("w", buffering=1) |
| proc = subprocess.Popen( |
| [sys.executable, "app.py"], |
| cwd=str(SPACE_ROOT), |
| env=env, |
| stdout=log_file, |
| stderr=subprocess.STDOUT, |
| ) |
| try: |
| _wait_for_ready(proc, f"http://127.0.0.1:{port}", log_path) |
| yield f"http://127.0.0.1:{port}" |
| finally: |
| proc.terminate() |
| try: |
| proc.wait(timeout=10) |
| except subprocess.TimeoutExpired: |
| proc.kill() |
| proc.wait(timeout=5) |
| log_file.close() |
|
|