refactor(e2e_test): fix smg ci e2e test code quality (#16664)

sgl-model-gateway/e2e_test/benchmarks/conftest.py

@@ -152,6 +152,7 @@ def genai_bench_runner():
         )
         timeout = timeout_sec or int(os.environ.get("GENAI_BENCH_TEST_TIMEOUT", "120"))
 
+        try:
             proc = subprocess.Popen(
                 cmd,
                 env=os.environ.copy(),
@@ -159,6 +160,12 @@ def genai_bench_runner():
                 stderr=subprocess.PIPE,
                 text=True,
             )
+        except FileNotFoundError:
+            pytest.fail(f"genai-bench executable not found at {cli}")
+        except PermissionError:
+            pytest.fail(f"Permission denied executing {cli}")
+        except OSError as e:
+            pytest.fail(f"Failed to start genai-bench: {e}")
 
         # Start GPU monitor if needed
         gpu_monitor: GPUMonitor | None = None
@@ -172,6 +179,16 @@ def genai_bench_runner():
         except subprocess.TimeoutExpired:
             proc.kill()
             stdout, stderr = proc.communicate()
+            logger.error("genai-bench timed out after %ds", timeout)
+
+        # Log output if process failed or for debugging
+        if proc.returncode != 0:
+            logger.error(
+                "genai-bench exited with code %d\nstdout:\n%s\nstderr:\n%s",
+                proc.returncode,
+                stdout or "(empty)",
+                stderr or "(empty)",
+            )
 
         try:
             # Parse and validate results
@@ -187,6 +204,16 @@ def genai_bench_runner():
                 gpu_monitor.log_summary()
                 gpu_monitor.assert_thresholds(thresholds)
 
+        except AssertionError:
+            # Log genai-bench output when results not found
+            logger.error(
+                "genai-bench output (returncode=%d):\nstdout:\n%s\nstderr:\n%s",
+                proc.returncode,
+                stdout or "(empty)",
+                stderr or "(empty)",
+            )
+            raise
+
         finally:
             _cleanup_procs(kill_procs, drain_delay_sec)
             if gpu_monitor:

sgl-model-gateway/e2e_test/conftest.py

@@ -140,9 +140,9 @@ def pytest_runtest_logstart(nodeid: str, location: tuple) -> None:
     """Print clear test header at start of each test."""
     # Extract test name from nodeid (e.g., "test_mmlu.py::TestMMLU::test_mmlu_basic[grpc]")
     test_name = nodeid.split("::")[-1] if "::" in nodeid else nodeid
-    print(f"\n{'='*60}")
+    print(f"\n{'=' * LOG_SEPARATOR_WIDTH}")
     print(f"TEST: {test_name}")
-    print(f"{'='*60}")
+    print(f"{'=' * LOG_SEPARATOR_WIDTH}")
 
 
 # Path setup for imports
@@ -172,6 +172,7 @@ from infra import (
     ENV_SKIP_MODEL_POOL,
     ENV_STARTUP_TIMEOUT,
     LOCAL_MODES,
+    LOG_SEPARATOR_WIDTH,
     PARAM_MODEL,
     PARAM_SETUP_BACKEND,
     ConnectionMode,
@@ -434,17 +435,18 @@ def pytest_collection_finish(session: pytest.Session) -> None:
     max_required, available_gpus = validate_gpu_requirements()
 
     if max_required > available_gpus:
+        sep = "=" * LOG_SEPARATOR_WIDTH
         raise pytest.UsageError(
-            f"\n{'='*60}\n"
+            f"\n{sep}\n"
             f"GPU REQUIREMENTS EXCEEDED\n"
-            f"{'='*60}\n"
+            f"{sep}\n"
             f"Test '{_max_test_name}' requires {max_required} GPUs\n"
             f"Available: {available_gpus} GPUs\n"
             f"\nOptions:\n"
             f"  1. Run tests that fit: pytest -k 'not {_max_test_name.split('::')[0]}'\n"
             f"  2. Reduce workers: @pytest.mark.workers(prefill=1, decode=1)\n"
             f"  3. Skip GPU tests: SKIP_MODEL_POOL=1 pytest\n"
-            f"{'='*60}"
+            f"{sep}"
         )
 
     logger.info(

sgl-model-gateway/e2e_test/fixtures/router_manager.py

@@ -4,6 +4,7 @@ from dataclasses import dataclass
 from typing import Dict, List, Optional
 
 import requests
+from infra import wait_for_health
 
 from .ports import find_free_port
 
@@ -126,22 +127,9 @@ class RouterManager:
         proc = subprocess.Popen(cmd)
         self._children.append(proc)
         url = f"http://127.0.0.1:{port}"
-        self._wait_health(url)
+        wait_for_health(url, timeout=30.0, check_interval=0.2)
         return ProcHandle(process=proc, url=url)
 
-    def _wait_health(self, base_url: str, timeout: float = 30.0):
-        start = time.time()
-        with requests.Session() as s:
-            while time.time() - start < timeout:
-                try:
-                    r = s.get(f"{base_url}/health", timeout=2)
-                    if r.status_code == 200:
-                        return
-                except requests.RequestException:
-                    pass
-                time.sleep(0.2)
-        raise TimeoutError(f"Router at {base_url} did not become healthy")
-
     def add_worker(self, base_url: str, worker_url: str, timeout: float = 30.0) -> None:
         r = requests.post(f"{base_url}/workers", json={"url": worker_url})
         assert (

sgl-model-gateway/e2e_test/infra/__init__.py

@@ -17,6 +17,8 @@ from .constants import (  # Enums; Convenience sets; Fixture parameters; Default
     HEALTH_CHECK_INTERVAL,
     LOCAL_MODES,
     LOCAL_RUNTIMES,
+    LOG_SEPARATOR_WIDTH,
+    MAX_RETRY_ATTEMPTS,
     PARAM_BACKEND_ROUTER,
     PARAM_MODEL,
     PARAM_SETUP_BACKEND,
@@ -82,6 +84,8 @@ __all__ = [
     "DEFAULT_STARTUP_TIMEOUT",
     "DEFAULT_ROUTER_TIMEOUT",
     "HEALTH_CHECK_INTERVAL",
+    "MAX_RETRY_ATTEMPTS",
+    "LOG_SEPARATOR_WIDTH",
     # Env vars
     "ENV_MODELS",
     "ENV_BACKENDS",

sgl-model-gateway/e2e_test/infra/constants.py

@@ -58,3 +58,11 @@ DEFAULT_HOST = "127.0.0.1"
 DEFAULT_STARTUP_TIMEOUT = 300
 DEFAULT_ROUTER_TIMEOUT = 60
 HEALTH_CHECK_INTERVAL = 5
+
+# Retry configuration
+MAX_RETRY_ATTEMPTS = (
+    6  # Max retries with exponential backoff (total ~63s: 1+2+4+8+16+32)
+)
+
+# Display formatting
+LOG_SEPARATOR_WIDTH = 60  # Width for log separator lines (e.g., "="*60)

sgl-model-gateway/e2e_test/infra/gateway.py

@@ -328,22 +328,17 @@ class Gateway:
     # Worker Management APIs
     # -------------------------------------------------------------------------
 
-    def list_workers(self, timeout: float = 5.0) -> list[WorkerInfo]:
+    def _worker_from_api_response(self, w: dict) -> WorkerInfo:
-        """List all workers connected to the gateway.
+        """Convert API response dict to WorkerInfo.
+
+        Args:
+            w: Worker dict from API response.
 
         Returns:
-            List of WorkerInfo objects.
+            WorkerInfo object.
         """
-        try:
-            resp = httpx.get(f"{self.base_url}/workers", timeout=timeout)
-            if resp.status_code == 200:
-                data = resp.json()
-                workers = []
-                for w in data.get("workers", []):
-                    # Map API fields to WorkerInfo
         status = "healthy" if w.get("is_healthy", False) else "unhealthy"
-                    workers.append(
+        return WorkerInfo(
-                        WorkerInfo(
             id=w.get("id", ""),
             url=w.get("url", ""),
             model=w.get("model_id"),
@@ -356,8 +351,20 @@ class Gateway:
                 "cost": w.get("cost"),
             },
         )
-                    )
+
-                return workers
+    def list_workers(self, timeout: float = 5.0) -> list[WorkerInfo]:
+        """List all workers connected to the gateway.
+
+        Returns:
+            List of WorkerInfo objects.
+        """
+        try:
+            resp = httpx.get(f"{self.base_url}/workers", timeout=timeout)
+            if resp.status_code == 200:
+                data = resp.json()
+                return [
+                    self._worker_from_api_response(w) for w in data.get("workers", [])
+                ]
             return []
         except (httpx.RequestError, httpx.TimeoutException):
             return []
@@ -374,21 +381,7 @@ class Gateway:
         try:
             resp = httpx.get(f"{self.base_url}/workers/{worker_id}", timeout=timeout)
             if resp.status_code == 200:
-                w = resp.json()
+                return self._worker_from_api_response(resp.json())
-                status = "healthy" if w.get("is_healthy", False) else "unhealthy"
-                return WorkerInfo(
-                    id=w.get("id", ""),
-                    url=w.get("url", ""),
-                    model=w.get("model_id"),
-                    status=status,
-                    pending_requests=w.get("load", 0),
-                    metadata={
-                        "worker_type": w.get("worker_type"),
-                        "connection_mode": w.get("connection_mode"),
-                        "priority": w.get("priority"),
-                        "cost": w.get("cost"),
-                    },
-                )
             return None
         except (httpx.RequestError, httpx.TimeoutException):
             return None

sgl-model-gateway/e2e_test/infra/gpu_allocator.py

@@ -227,7 +227,7 @@ class GPUAllocator:
                     name = pynvml.nvmlDeviceGetName(handle)
                     # Handle bytes vs string return type (varies by pynvml version)
                     if isinstance(name, bytes):
-                        name = name.decode("utf-8")
+                        name = name.decode("utf-8", errors="replace")
                     mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
                     # Convert bytes to MB
                     memory_mb = mem_info.total // (1024 * 1024)

sgl-model-gateway/e2e_test/infra/model_pool.py

@@ -305,9 +305,6 @@ class ModelPool:
         if ib_device:
             logger.info("Detected InfiniBand device: %s", ib_device)
 
-        # Track bootstrap ports for PD groups (all PD workers of same model/mode share one)
-        pd_bootstrap_ports: dict[tuple[str, ConnectionMode], int] = {}
-
         deferred: list[str] = []
 
         # Process requirements in order - all workers treated uniformly
@@ -340,13 +337,8 @@ class ModelPool:
                 deferred.append(str(identity))
                 continue
 
-            # Get bootstrap port for PD workers (shared within model/mode group)
+            # Each prefill worker needs its own bootstrap port for PD communication
-            bootstrap_port = None
+            bootstrap_port = get_open_port() if identity.is_prefill else None
-            if identity.is_prefill or identity.is_decode:
-                pd_key = (identity.model_id, identity.mode)
-                if pd_key not in pd_bootstrap_ports:
-                    pd_bootstrap_ports[pd_key] = get_open_port()
-                bootstrap_port = pd_bootstrap_ports[pd_key]
 
             # Launch the worker
             self._launch_model(
@@ -354,7 +346,7 @@ class ModelPool:
                 mode=identity.mode,
                 gpu_slot=slots[0],
                 worker_type=identity.worker_type,
-                bootstrap_port=bootstrap_port if identity.is_prefill else None,
+                bootstrap_port=bootstrap_port,
                 ib_device=(
                     ib_device if (identity.is_prefill or identity.is_decode) else None
                 ),
@@ -888,25 +880,17 @@ class ModelPool:
         has_pd = any(w.is_prefill or w.is_decode for w in valid_workers)
         ib_device = detect_ib_device() if has_pd else None
 
-        # Track bootstrap ports for PD groups (shared within model/mode)
-        pd_bootstrap_ports: dict[tuple[str, ConnectionMode], int] = {}
-
         instances: list[ModelInstance] = []
         for w in valid_workers:
-            # Get bootstrap port for PD workers
+            # Each prefill worker needs its own bootstrap port for PD communication
-            bootstrap_port = None
+            bootstrap_port = get_open_port() if w.is_prefill else None
-            if w.is_prefill or w.is_decode:
-                pd_key = (w.model_id, w.mode)
-                if pd_key not in pd_bootstrap_ports:
-                    pd_bootstrap_ports[pd_key] = get_open_port()
-                bootstrap_port = pd_bootstrap_ports[pd_key]
 
             instance = self._launch_model(
                 model_id=w.model_id,
                 mode=w.mode,
                 gpu_slot=slot_map.get(w.key),
                 worker_type=w.worker_type,
-                bootstrap_port=bootstrap_port if w.is_prefill else None,
+                bootstrap_port=bootstrap_port,
                 ib_device=ib_device if (w.is_prefill or w.is_decode) else None,
                 instance_key=w.key,
             )

sgl-model-gateway/e2e_test/infra/run_eval.py

@@ -30,12 +30,10 @@ if TYPE_CHECKING:
     from .simple_eval_common import Eval
 
 from .simple_eval_common import ChatCompletionSampler, set_ulimit
+from .simple_eval_mmlu import MMLU_DATASET_URL
 
 logger = logging.getLogger(__name__)
 
-# MMLU dataset URL
-MMLU_DATASET_URL = "https://openaipublic.blob.core.windows.net/simple-evals/mmlu.csv"
-
 
 @dataclass
 class EvalConfig:

sgl-model-gateway/e2e_test/infra/simple_eval_common.py

@@ -20,6 +20,8 @@ import requests
 from openai import OpenAI
 from tqdm import tqdm
 
+from .constants import MAX_RETRY_ATTEMPTS
+
 logger = logging.getLogger(__name__)
 
 OPENAI_SYSTEM_MESSAGE_API = "You are a helpful assistant."
@@ -119,7 +121,6 @@ class ChatCompletionSampler(SamplerBase):
         image: str,
         encoding: str = "base64",
         format: str = "png",
-        fovea: int = 768,
     ):
         new_image = {
             "type": "image_url",
@@ -141,7 +142,7 @@ class ChatCompletionSampler(SamplerBase):
                 self._pack_message("system", self.system_message)
             ] + message_list
         trial = 0
-        while trial < 6:  # Max 63 seconds backoff (1+2+4+8+16+32)
+        while trial < MAX_RETRY_ATTEMPTS:
             try:
                 response = self.client.chat.completions.create(
                     model=self.model,
@@ -162,14 +163,15 @@ class ChatCompletionSampler(SamplerBase):
                 log_fn(
                     "Request failed (retry %d/%d, backoff %ds): %s",
                     trial + 1,
-                    6,
+                    MAX_RETRY_ATTEMPTS,
                     exception_backoff,
                     e,
                 )
                 time.sleep(exception_backoff)
                 trial += 1
         logger.warning(
-            "All retry attempts exhausted after 6 retries, returning empty response"
+            "All retry attempts exhausted after %d retries, returning empty response",
+            MAX_RETRY_ATTEMPTS,
         )
         return ""
 

sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py

@@ -25,6 +25,9 @@ from .simple_eval_common import (
 if TYPE_CHECKING:
     from .simple_eval_common import SamplerBase
 
+# MMLU dataset URL (hosted by OpenAI)
+MMLU_DATASET_URL = "https://openaipublic.blob.core.windows.net/simple-evals/mmlu.csv"
+
 SUBJECT_TO_CATEGORY = {
     "abstract_algebra": "stem",
     "anatomy": "other",