Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 7 additions & 24 deletions e2e_test/fixtures/hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import pytest
from infra import get_runtime

from .markers import resolve_class_marker

# ---------------------------------------------------------------------------
# Marker registration
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -95,32 +97,13 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
# ---------------------------------------------------------------------------


def _get_own_class_marker(item: pytest.Item, name: str):
"""Get a marker defined on the item's own class, not inherited from parents.

When a test class subclasses another test class, pytest's pytestmark list
includes parent markers first and child markers last. get_closest_marker()
therefore returns the parent's marker, which can be wrong when the child
intentionally overrides a marker (e.g. engine("sglang") overriding
engine("sglang","vllm","trtllm")).
def _get_marker(item: pytest.Item, name: str):
"""Get the most specific marker, preferring child class over parent.

This helper checks the child class's __dict__ directly.
Delegates to resolve_class_marker() which walks the class MRO (child-first)
so that a child class marker overrides a parent class marker.
"""
cls = getattr(item, "cls", None)
if cls is None:
return None
own_marks = cls.__dict__.get("pytestmark", [])
if not isinstance(own_marks, list):
own_marks = [own_marks]
for mark in own_marks:
if getattr(mark, "name", None) == name:
return mark
return None


def _get_marker(item: pytest.Item, name: str):
"""Get the most specific marker, preferring child class over parent."""
return _get_own_class_marker(item, name) or item.get_closest_marker(name)
return resolve_class_marker(item, name)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Honor method markers over inherited class markers

Switching _get_marker() to resolve_class_marker() changes precedence when a test method has an engine/vendor/gpu marker but its class only inherits a parent class marker: resolve_class_marker() returns the inherited class marker before consulting item.get_closest_marker(), so method-level overrides are ignored. Under E2E_ENGINE/E2E_VENDOR/E2E_GPU_TIER filtering this can incorrectly include/exclude tests in subclass hierarchies where only the parent class carries the broad marker.

Useful? React with 👍 / 👎.



def pytest_collection_modifyitems(
Expand Down
2 changes: 0 additions & 2 deletions e2e_test/infra/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,6 @@ def get_runtime() -> str:
"""
global _RUNTIME_CACHE
if _RUNTIME_CACHE is None:
import os

_RUNTIME_CACHE = os.environ.get(ENV_RUNTIME, DEFAULT_RUNTIME)
return _RUNTIME_CACHE

Expand Down
7 changes: 3 additions & 4 deletions e2e_test/infra/gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import logging
import os
import subprocess
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any
Expand Down Expand Up @@ -373,10 +374,8 @@ def add_worker(
worker_id = data.get("worker_id")

if wait_ready and worker_id:
import time

start = time.time()
while time.time() - start < ready_timeout:
start = time.perf_counter()
while time.perf_counter() - start < ready_timeout:
workers = self.list_workers()
for w in workers:
if w.id == worker_id:
Expand Down
26 changes: 13 additions & 13 deletions e2e_test/infra/gpu_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,10 @@
logger = logging.getLogger(__name__)


def _percentile(samples: list[float], p: float) -> float:
"""Calculate percentile from sorted samples."""
if not samples:
def _percentile(sorted_samples: list[float], p: float) -> float:
"""Calculate percentile from pre-sorted samples."""
if not sorted_samples:
return 0.0
sorted_samples = sorted(samples)
idx = max(
0,
min(len(sorted_samples) - 1, int(round((p / 100.0) * (len(sorted_samples) - 1)))),
Expand Down Expand Up @@ -64,17 +63,18 @@ def _compute_stats(
if not trimmed:
trimmed = samples # Fallback to original if trimming removes all samples

sorted_trimmed = sorted(trimmed)
return {
"mean": sum(trimmed) / len(trimmed),
"min": min(trimmed),
"max": max(trimmed),
"p5": _percentile(trimmed, 5),
"p10": _percentile(trimmed, 10),
"p25": _percentile(trimmed, 25),
"p50": _percentile(trimmed, 50),
"p75": _percentile(trimmed, 75),
"p90": _percentile(trimmed, 90),
"p95": _percentile(trimmed, 95),
"min": sorted_trimmed[0],
"max": sorted_trimmed[-1],
"p5": _percentile(sorted_trimmed, 5),
"p10": _percentile(sorted_trimmed, 10),
"p25": _percentile(sorted_trimmed, 25),
"p50": _percentile(sorted_trimmed, 50),
"p75": _percentile(sorted_trimmed, 75),
"p90": _percentile(sorted_trimmed, 90),
"p95": _percentile(sorted_trimmed, 95),
"count": len(trimmed),
}

Expand Down
31 changes: 20 additions & 11 deletions e2e_test/infra/model_specs.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,22 +110,31 @@ def get_models_with_feature(feature: str) -> list[str]:
]


def _parse_tp_overrides() -> dict | None:
"""Parse E2E_MODEL_TP_OVERRIDES env var once at import time."""
raw = os.environ.get("E2E_MODEL_TP_OVERRIDES")
if raw:
try:
parsed = json.loads(raw)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
pass
return None


_TP_OVERRIDES = _parse_tp_overrides()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep TP override env lookup dynamic

Caching E2E_MODEL_TP_OVERRIDES at import time freezes the override map for the lifetime of the process, so any later environment updates are silently ignored by get_model_spec(). This is a behavioral regression from the previous per-call lookup and breaks workflows/tests that set or mutate this env var after module import (for example via monkeypatch.setenv) to control GPU parallelism per run.

Useful? React with 👍 / 👎.



def get_model_spec(model_id: str) -> dict:
"""Get spec for a specific model, raising KeyError if not found."""
if model_id not in MODEL_SPECS:
raise KeyError(f"Unknown model: {model_id}. Available: {list(MODEL_SPECS.keys())}")
spec = dict(MODEL_SPECS[model_id])
tp_overrides_json = os.environ.get("E2E_MODEL_TP_OVERRIDES")
if tp_overrides_json:
try:
tp_overrides = json.loads(tp_overrides_json)
if isinstance(tp_overrides, dict):
override = tp_overrides.get(model_id)
if isinstance(override, int) and override > 0:
spec["tp"] = override
except json.JSONDecodeError:
# Ignore malformed override config and fall back to canonical specs.
pass
if _TP_OVERRIDES is not None:
override = _TP_OVERRIDES.get(model_id)
if isinstance(override, int) and override > 0:
spec["tp"] = override
return spec


Expand Down
40 changes: 18 additions & 22 deletions e2e_test/infra/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -324,37 +324,33 @@ def _wait_grpc_healthy(self, timeout: float) -> None:
)

start = time.perf_counter()
while time.perf_counter() - start < timeout:
if not self.is_alive():
pid = self.process.pid if self.process else "unknown"
raise RuntimeError(f"Worker {self.model_id} (PID {pid}) died during startup")
try:
channel = grpc.insecure_channel(f"{DEFAULT_HOST}:{self.port}")
channel = grpc.insecure_channel(f"{DEFAULT_HOST}:{self.port}")
try:
while time.perf_counter() - start < timeout:
if not self.is_alive():
pid = self.process.pid if self.process else "unknown"
raise RuntimeError(f"Worker {self.model_id} (PID {pid}) died during startup")
try:
stub = health_pb2_grpc.HealthStub(channel)
request = health_pb2.HealthCheckRequest(service="")
response = stub.Check(request, timeout=5.0)
if response.status == health_pb2.HealthCheckResponse.SERVING:
return
finally:
channel.close()
except grpc.RpcError as e:
# UNIMPLEMENTED means server is up but doesn't have health service;
# fall back to channel connectivity check
if hasattr(e, "code") and e.code() == grpc.StatusCode.UNIMPLEMENTED:
try:
channel = grpc.insecure_channel(f"{DEFAULT_HOST}:{self.port}")
except grpc.RpcError as e:
# UNIMPLEMENTED means server is up but doesn't have health service;
# fall back to channel connectivity check
if hasattr(e, "code") and e.code() == grpc.StatusCode.UNIMPLEMENTED:
try:
grpc.channel_ready_future(channel).result(timeout=5.0)
return
finally:
channel.close()
except Exception:
pass
except Exception:
pass

time.sleep(HEALTH_CHECK_INTERVAL)
except Exception:
pass
except Exception:
pass

time.sleep(HEALTH_CHECK_INTERVAL)
finally:
channel.close()

raise TimeoutError(
f"gRPC worker {self.model_id} on port {self.port} "
Expand Down