File size: 2,853 Bytes
21b071a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """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),
# submit.py's boot-time stuck-pending sweep hits the Hub on
# import. Off in tests so a Hub blip doesn't slow the fixture
# or pollute the log; the sweep is exercised separately in
# submit-specific tests.
"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()
|