"""Comprehensive smoke test for the Cloud-Native DevOps Debug FastAPI server. Usage: .\\.venv\\Scripts\\python.exe smoke_test.py .\\.venv\\Scripts\\python.exe smoke_test.py --mode live --base-url http://127.0.0.1:7860 Modes: - inprocess (default): uses FastAPI TestClient, no running server needed. - live: uses requests against a running server. """ import argparse import json import sys from dataclasses import dataclass from typing import Any, Dict, Optional, Tuple @dataclass class TestResult: name: str ok: bool details: str = "" class EndpointClient: def get(self, path: str) -> Tuple[int, Dict[str, Any]]: raise NotImplementedError def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]: raise NotImplementedError class InProcessClient(EndpointClient): def __init__(self): from fastapi.testclient import TestClient from server.app import app self._client = TestClient(app) def get(self, path: str) -> Tuple[int, Dict[str, Any]]: response = self._client.get(path) try: data = response.json() except Exception: data = {} return response.status_code, data def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]: response = self._client.post(path, json=body or {}) data = response.json() if response.content else {} return response.status_code, data class LiveClient(EndpointClient): def __init__(self, base_url: str): import requests self._requests = requests self._base_url = base_url.rstrip("/") def get(self, path: str) -> Tuple[int, Dict[str, Any]]: response = self._requests.get(f"{self._base_url}{path}", timeout=20) try: data = response.json() except Exception: data = {} return response.status_code, data def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]: response = self._requests.post(f"{self._base_url}{path}", json=body or {}, timeout=20) data = response.json() if response.content else {} return response.status_code, data def assert_true(name: str, cond: bool, details: str = "") -> TestResult: return TestResult(name=name, ok=bool(cond), details=details if not cond else "") def run_smoke(client: EndpointClient) -> int: results = [] # root serves the landing page now (HTML), just check it's 200 status, _ = client.get("/") results.append(assert_true("GET / landing page", status == 200)) status, health = client.get("/health") results.append(assert_true("GET /health", status == 200 and health.get("status") == "healthy", str(health))) status, info = client.get("/info") results.append(assert_true("GET /info", status == 200 and isinstance(info.get("tasks"), list), str(info))) status, tasks_payload = client.get("/tasks") tasks = tasks_payload.get("tasks", []) if isinstance(tasks_payload, dict) else [] results.append(assert_true("GET /tasks", status == 200 and len(tasks) >= 6, str(tasks_payload))) status, reset_data = client.post("/reset", {"seed": 123}) obs = reset_data.get("observation", {}) results.append( assert_true( "POST /reset random", status == 200 and isinstance(obs.get("task_id"), str) and isinstance(obs.get("files"), list), str(reset_data), ) ) status_int, reset_int = client.post("/reset", {"task_id": 1, "seed": 1}) status_str, reset_str = client.post("/reset", {"task_id": "1", "seed": 1}) int_task = reset_int.get("observation", {}).get("task_id") str_task = reset_str.get("observation", {}).get("task_id") results.append( assert_true( "POST /reset accepts int/string index", status_int == 200 and status_str == 200 and int_task == str_task, f"int={status_int}:{int_task}, str={status_str}:{str_task}", ) ) status_a, reset_a = client.post("/reset", {"seed": 999}) status_b, reset_b = client.post("/reset", {"seed": 999}) a_obs = reset_a.get("observation", {}) b_obs = reset_b.get("observation", {}) results.append( assert_true( "Deterministic reset with seed", status_a == 200 and status_b == 200 and a_obs.get("task_id") == b_obs.get("task_id") and a_obs.get("error", {}).get("error_message") == b_obs.get("error", {}).get("error_message"), f"A={a_obs.get('task_id')} B={b_obs.get('task_id')}", ) ) status, _ = client.post("/reset", {"task_id": "dockerfile_syntax", "scenario_id": "typo_filename", "seed": 7}) results.append(assert_true("POST /reset specific scenario", status == 200)) status, step_hint = client.post( "/step", {"action": {"action_type": "request_hint", "reasoning": "Need help"}}, ) results.append( assert_true( "POST /step request_hint", status == 200 and "observation" in step_hint and "reward" in step_hint, str(step_hint), ) ) status, step_fix = client.post( "/step", { "action": { "action_type": "replace_line", "edits": [{"file_path": "Dockerfile", "line_number": 3, "new_content": "COPY requirements.txt ."}], "reasoning": "Fix typo", } }, ) fix_info = step_fix.get("info", {}) results.append( assert_true( "POST /step replace_line", status == 200 and fix_info.get("issues_fixed", 0) >= 1, str(step_fix), ) ) status, state = client.get("/state") results.append(assert_true("GET /state", status == 200 and "observation" in state, str(state))) status, submit = client.post("/step", {"action": {"action_type": "submit", "reasoning": "Done"}}) results.append(assert_true("POST /step submit", status == 200 and submit.get("done") is True, str(submit))) trajectory = [ { "step": 1, "action": {"action_type": "replace_line", "edits": [{"file_path": "Dockerfile", "line_number": 3}]}, "reward": 0.3, "done": False, "info": {"issues_fixed": 1, "issues_total": 1}, }, { "step": 2, "action": {"action_type": "submit"}, "reward": 0.7, "done": True, "info": {"issues_fixed": 1, "issues_total": 1}, }, ] status, grader = client.post("/grader", {"task_id": "dockerfile_syntax", "trajectory": trajectory}) score = grader.get("result", {}).get("score") results.append( assert_true( "POST /grader", status == 200 and isinstance(score, (int, float)) and 0.0 <= float(score) <= 1.0, str(grader), ) ) status, baseline = client.post("/baseline", {"task_id": "dockerfile_syntax", "num_episodes": 1}) results.append( assert_true( "POST /baseline", status == 200 and isinstance(baseline.get("results"), list), str(baseline), ) ) passed = sum(1 for r in results if r.ok) total = len(results) print("\n=== Smoke Test Results ===") for r in results: marker = "PASS" if r.ok else "FAIL" print(f"[{marker}] {r.name}") if not r.ok and r.details: detail = r.details if len(detail) > 300: detail = detail[:300] + "..." print(f" {detail}") print(f"\nSummary: {passed}/{total} passed") return 0 if passed == total else 1 def main() -> int: parser = argparse.ArgumentParser(description="Smoke test Cloud-Native DevOps Debug FastAPI server") parser.add_argument("--mode", choices=["inprocess", "live"], default="inprocess") parser.add_argument("--base-url", default="http://127.0.0.1:7860") args = parser.parse_args() if args.mode == "inprocess": client = InProcessClient() else: client = LiveClient(args.base_url) return run_smoke(client) if __name__ == "__main__": raise SystemExit(main())