scripts/smoke_test.py

#!/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()