added something necessary

pre_submission_check.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import site
+import socket
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any, Dict, List
+
+
+REPO_ROOT = Path(__file__).resolve().parent
+REQUIRED_FILES = [
+    "inference.py",
+    "openenv.yaml",
+    "Dockerfile",
+    "uv.lock",
+]
+
+
+def build_pythonpath() -> str:
+    parts: List[str] = []
+    candidates = [
+        REPO_ROOT,
+        REPO_ROOT / "src",
+        REPO_ROOT / ".deps",
+    ]
+    for candidate in candidates:
+        if candidate.exists():
+            parts.append(str(candidate))
+
+    try:
+        user_site = Path(site.getusersitepackages())
+        if user_site.exists():
+            parts.append(str(user_site))
+    except Exception:
+        pass
+
+    existing = os.environ.get("PYTHONPATH", "").strip()
+    if existing:
+        parts.append(existing)
+    return os.pathsep.join(parts)
+
+
+def command_env() -> Dict[str, str]:
+    env = os.environ.copy()
+    env["PYTHONPATH"] = build_pythonpath()
+    return env
+
+
+def run_command(args: List[str], timeout: int = 120, check: bool = False) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        args,
+        cwd=REPO_ROOT,
+        env=command_env(),
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=check,
+    )
+
+
+def file_checks() -> Dict[str, Any]:
+    missing = [path for path in REQUIRED_FILES if not (REPO_ROOT / path).exists()]
+    return {
+        "name": "required_artifacts",
+        "passed": not missing,
+        "missing": missing,
+    }
+
+
+def local_openenv_validation() -> Dict[str, Any]:
+    result = run_command([sys.executable, "-m", "openenv.cli", "validate", "--json"])
+    payload: Dict[str, Any]
+    try:
+        payload = json.loads(result.stdout.strip() or "{}")
+    except json.JSONDecodeError:
+        payload = {"raw_stdout": result.stdout}
+
+    return {
+        "name": "local_openenv_validate",
+        "passed": result.returncode == 0 and bool(payload.get("passed", False)),
+        "returncode": result.returncode,
+        "report": payload,
+        "stderr": result.stderr.strip(),
+    }
+
+
+def pick_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def wait_for_health(url: str, timeout_s: float = 20.0) -> bool:
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        try:
+            with urllib.request.urlopen(f"{url}/health", timeout=2) as response:
+                payload = json.loads(response.read().decode("utf-8"))
+                if payload.get("status") in {"healthy", "ok"}:
+                    return True
+        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, ValueError):
+            time.sleep(0.5)
+    return False
+
+
+def runtime_openenv_validation() -> Dict[str, Any]:
+    port = pick_port()
+    url = f"http://127.0.0.1:{port}"
+    env = command_env()
+    server = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "uvicorn",
+            "server.app:app",
+            "--host",
+            "127.0.0.1",
+            "--port",
+            str(port),
+        ],
+        cwd=REPO_ROOT,
+        env=env,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        text=True,
+    )
+
+    try:
+        if not wait_for_health(url):
+            return {
+                "name": "runtime_openenv_validate",
+                "passed": False,
+                "url": url,
+                "error": "Server did not become healthy in time.",
+            }
+
+        result = run_command(
+            [sys.executable, "-m", "openenv.cli", "validate", "--url", url],
+            timeout=120,
+        )
+        payload: Dict[str, Any]
+        try:
+            payload = json.loads(result.stdout.strip() or "{}")
+        except json.JSONDecodeError:
+            payload = {"raw_stdout": result.stdout}
+
+        return {
+            "name": "runtime_openenv_validate",
+            "passed": result.returncode == 0 and bool(payload.get("passed", False)),
+            "url": url,
+            "returncode": result.returncode,
+            "report": payload,
+            "stderr": result.stderr.strip(),
+        }
+    finally:
+        if server.poll() is None:
+            server.terminate()
+            try:
+                server.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                server.kill()
+
+
+def summarize(results: List[Dict[str, Any]]) -> Dict[str, Any]:
+    failed = [result["name"] for result in results if not result.get("passed", False)]
+    return {
+        "passed": not failed,
+        "failed_checks": failed,
+        "checks": results,
+    }
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run local pre-submission checks for the IDPI benchmark.")
+    parser.add_argument("--skip-runtime", action="store_true", help="Skip live runtime validation against a local server.")
+    parser.add_argument("--json", action="store_true", help="Emit a JSON report instead of a human summary.")
+    return parser.parse_args()
+
+
+def print_human_report(report: Dict[str, Any]) -> None:
+    status = "PASS" if report["passed"] else "FAIL"
+    print(f"[{status}] Pre-submission check")
+    for check in report["checks"]:
+        check_status = "PASS" if check.get("passed", False) else "FAIL"
+        print(f"- {check_status}: {check['name']}")
+        if check["name"] == "required_artifacts" and check.get("missing"):
+            print(f"  missing: {', '.join(check['missing'])}")
+        if check["name"].endswith("openenv_validate") and not check.get("passed", False):
+            if check.get("stderr"):
+                print(f"  stderr: {check['stderr']}")
+            if check.get("error"):
+                print(f"  error: {check['error']}")
+
+
+def main() -> None:
+    args = parse_args()
+    results = [file_checks(), local_openenv_validation()]
+    if not args.skip_runtime:
+        results.append(runtime_openenv_validation())
+
+    report = summarize(results)
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print_human_report(report)
+
+    raise SystemExit(0 if report["passed"] else 1)
+
+
+if __name__ == "__main__":
+    main()