Spaces:
Running
Running
File size: 2,726 Bytes
e317359 | 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 | #!/usr/bin/env python3
"""Offline two-cycle forecast/score test plus a real leaderboard HTTP refresh."""
from __future__ import annotations
import importlib.util
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import time
ROOT = Path(__file__).resolve().parents[1]
def main() -> None:
# The protocol tests include frozen predictions, matching targets and recovery.
subprocess.run(
[
sys.executable,
"-m",
"pytest",
"-q",
"tests/test_prequential_protocol.py",
"tests/test_maintenance.py",
"tests/test_forecast_snapshots.py",
],
cwd=ROOT,
check=True,
)
from gradio_client import Client
import httpx
env = os.environ.copy()
env.update(
GRADIO_ANALYTICS_ENABLED="False",
GRADIO_SERVER_NAME="127.0.0.1",
GRADIO_SERVER_PORT="7863",
)
env["NO_PROXY"] = "localhost,127.0.0.1," + env.get("NO_PROXY", "")
with tempfile.TemporaryFile(mode="w+") as log:
proc = subprocess.Popen(
[sys.executable, str(ROOT / "space/app.py")],
cwd=ROOT,
env=env,
stdout=log,
stderr=log,
)
try:
for _ in range(60):
if proc.poll() is not None:
log.seek(0)
raise RuntimeError(log.read())
try:
response = httpx.get(
"http://127.0.0.1:7863/config", timeout=2, trust_env=False
)
if response.status_code == 200:
break
except httpx.HTTPError:
pass
time.sleep(1)
else:
raise RuntimeError("Leaderboard did not start within 60 seconds")
client = Client(
"http://127.0.0.1:7863",
verbose=False,
httpx_kwargs={"trust_env": False},
)
result = client.predict(api_name="/refresh_leaderboard")
assert result and "summary-grid" in result[0]
print(f"Leaderboard HTTP and refresh passed: {len(result)} outputs.")
from check_forecast_ui import check_forecast_ui
forecasts = check_forecast_ui(client, response.json())
print(f"Forecast HTTP model selection and curves passed: {forecasts['datasets_checked']} datasets.")
finally:
proc.terminate()
try:
proc.wait(timeout=10)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
if __name__ == "__main__":
main()
|