jester1177/cloud-native-debug-env
0
1"""Comprehensive smoke test for the CI/CD Debug FastAPI server.2 3Usage:4 .\\.venv\\Scripts\\python.exe smoke_test.py5 .\\.venv\\Scripts\\python.exe smoke_test.py --mode live --base-url http://127.0.0.1:80006 7Modes:8 - inprocess (default): uses FastAPI TestClient, no running server needed.9 - live: uses requests against a running server.10"""11 12 13import argparse14import json15import sys16from dataclasses import dataclass17from typing import Any, Dict, Optional, Tuple18 19 20@dataclass21class TestResult:22 name: str23 ok: bool24 details: str = ""25 26 27class EndpointClient:28 def get(self, path: str) -> Tuple[int, Dict[str, Any]]:29 raise NotImplementedError30 31 def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:32 raise NotImplementedError33 34 35class InProcessClient(EndpointClient):36 def __init__(self):37 from fastapi.testclient import TestClient38 from server.app import app39 40 self._client = TestClient(app)41 42 def get(self, path: str) -> Tuple[int, Dict[str, Any]]:43 response = self._client.get(path)44 try:45 data = response.json()46 except Exception:47 data = {}48 return response.status_code, data49 50 def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:51 response = self._client.post(path, json=body or {})52 data = response.json() if response.content else {}53 return response.status_code, data54 55 56class LiveClient(EndpointClient):57 def __init__(self, base_url: str):58 import requests59 60 self._requests = requests61 self._base_url = base_url.rstrip("/")62 63 def get(self, path: str) -> Tuple[int, Dict[str, Any]]:64 response = self._requests.get(f"{self._base_url}{path}", timeout=20)65 try:66 data = response.json()67 except Exception:68 data = {}69 return response.status_code, data70 71 def post(self, path: str, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:72 response = self._requests.post(f"{self._base_url}{path}", json=body or {}, timeout=20)73 data = response.json() if response.content else {}74 return response.status_code, data75 76 77def assert_true(name: str, cond: bool, details: str = "") -> TestResult:78 return TestResult(name=name, ok=bool(cond), details=details if not cond else "")79 80 81def run_smoke(client: EndpointClient) -> int:82 results = []83 84 # root serves the landing page now (HTML), just check it's 20085 status, _ = client.get("/")86 results.append(assert_true("GET / landing page", status == 200))87 88 status, health = client.get("/health")89 results.append(assert_true("GET /health", status == 200 and health.get("status") == "healthy", str(health)))90 91 status, info = client.get("/info")92 results.append(assert_true("GET /info", status == 200 and isinstance(info.get("tasks"), list), str(info)))93 94 status, tasks_payload = client.get("/tasks")95 tasks = tasks_payload.get("tasks", []) if isinstance(tasks_payload, dict) else []96 results.append(assert_true("GET /tasks", status == 200 and len(tasks) >= 6, str(tasks_payload)))97 98 status, reset_data = client.post("/reset", {"seed": 123})99 obs = reset_data.get("observation", {})100 results.append(101 assert_true(102 "POST /reset random",103 status == 200 and isinstance(obs.get("task_id"), str) and isinstance(obs.get("files"), list),104 str(reset_data),105 )106 )107 108 status_int, reset_int = client.post("/reset", {"task_id": 1, "seed": 1})109 status_str, reset_str = client.post("/reset", {"task_id": "1", "seed": 1})110 int_task = reset_int.get("observation", {}).get("task_id")111 str_task = reset_str.get("observation", {}).get("task_id")112 results.append(113 assert_true(114 "POST /reset accepts int/string index",115 status_int == 200 and status_str == 200 and int_task == str_task,116 f"int={status_int}:{int_task}, str={status_str}:{str_task}",117 )118 )119 120 status_a, reset_a = client.post("/reset", {"seed": 999})121 status_b, reset_b = client.post("/reset", {"seed": 999})122 a_obs = reset_a.get("observation", {})123 b_obs = reset_b.get("observation", {})124 results.append(125 assert_true(126 "Deterministic reset with seed",127 status_a == 200128 and status_b == 200129 and a_obs.get("task_id") == b_obs.get("task_id")130 and a_obs.get("error", {}).get("error_message") == b_obs.get("error", {}).get("error_message"),131 f"A={a_obs.get('task_id')} B={b_obs.get('task_id')}",132 )133 )134 135 status, _ = client.post("/reset", {"task_id": "dockerfile_syntax", "scenario_id": "typo_filename", "seed": 7})136 results.append(assert_true("POST /reset specific scenario", status == 200))137 138 status, step_hint = client.post(139 "/step",140 {"action": {"action_type": "request_hint", "reasoning": "Need help"}},141 )142 results.append(143 assert_true(144 "POST /step request_hint",145 status == 200 and "observation" in step_hint and "reward" in step_hint,146 str(step_hint),147 )148 )149 150 status, step_fix = client.post(151 "/step",152 {153 "action": {154 "action_type": "replace_line",155 "edits": [{"file_path": "Dockerfile", "line_number": 3, "new_content": "COPY requirements.txt ."}],156 "reasoning": "Fix typo",157 }158 },159 )160 fix_info = step_fix.get("info", {})161 results.append(162 assert_true(163 "POST /step replace_line",164 status == 200 and fix_info.get("issues_fixed", 0) >= 1,165 str(step_fix),166 )167 )168 169 status, state = client.get("/state")170 results.append(assert_true("GET /state", status == 200 and "observation" in state, str(state)))171 172 status, submit = client.post("/step", {"action": {"action_type": "submit", "reasoning": "Done"}})173 results.append(assert_true("POST /step submit", status == 200 and submit.get("done") is True, str(submit)))174 175 trajectory = [176 {177 "step": 1,178 "action": {"action_type": "replace_line", "edits": [{"file_path": "Dockerfile", "line_number": 3}]},179 "reward": 0.3,180 "done": False,181 "info": {"issues_fixed": 1, "issues_total": 1},182 },183 {184 "step": 2,185 "action": {"action_type": "submit"},186 "reward": 0.7,187 "done": True,188 "info": {"issues_fixed": 1, "issues_total": 1},189 },190 ]191 status, grader = client.post("/grader", {"task_id": "dockerfile_syntax", "trajectory": trajectory})192 score = grader.get("result", {}).get("score")193 results.append(194 assert_true(195 "POST /grader",196 status == 200 and isinstance(score, (int, float)) and 0.0 <= float(score) <= 1.0,197 str(grader),198 )199 )200 201 status, baseline = client.post("/baseline", {"task_id": "dockerfile_syntax", "num_episodes": 1})202 results.append(203 assert_true(204 "POST /baseline",205 status == 200 and isinstance(baseline.get("results"), list),206 str(baseline),207 )208 )209 210 passed = sum(1 for r in results if r.ok)211 total = len(results)212 213 print("\n=== Smoke Test Results ===")214 for r in results:215 marker = "PASS" if r.ok else "FAIL"216 print(f"[{marker}] {r.name}")217 if not r.ok and r.details:218 detail = r.details219 if len(detail) > 300:220 detail = detail[:300] + "..."221 print(f" {detail}")222 223 print(f"\nSummary: {passed}/{total} passed")224 return 0 if passed == total else 1225 226 227def main() -> int:228 parser = argparse.ArgumentParser(description="Smoke test CI/CD Debug FastAPI server")229 parser.add_argument("--mode", choices=["inprocess", "live"], default="inprocess")230 parser.add_argument("--base-url", default="http://127.0.0.1:8000")231 args = parser.parse_args()232 233 if args.mode == "inprocess":234 client = InProcessClient()235 else:236 client = LiveClient(args.base_url)237 238 return run_smoke(client)239 240 241if __name__ == "__main__":242 raise SystemExit(main())243