minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(e2e): keep only benchmark tests in e2e_http, remove redundant tests (#16594)

Browse Source
This commit is contained in:
Simo Lin
2026-01-06 15:05:08 -08:00
committed by GitHub
parent 399d5283f8
commit dce8b0606c
3 changed files with 0 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sgl-model-gateway/e2e_test/e2e_http/test_pd_router.py
-19
View File
@@ -8,8 +8,6 @@ from typing import Optional
import pytest import pytest
import requests import requests
from sglang.test.run_eval import run_eval
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -224,23 +222,6 @@ def pd_cluster(e2e_model: str):
_terminate(w.proc) _terminate(w.proc)
@pytest.mark.e2e
def test_pd_mmlu(e2e_model: str, pd_cluster):
"""
Launch 4 workers, start a PD router (2 prefill + 2 decode), then run MMLU.
"""
args = SimpleNamespace(
base_url=pd_cluster.router_url,
model=e2e_model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
temperature=0.1,
)
metrics = run_eval(args)
assert metrics["score"] >= 0.65
@pytest.mark.e2e @pytest.mark.e2e
def test_pd_genai_bench(e2e_model: str, pd_cluster, genai_bench_runner): def test_pd_genai_bench(e2e_model: str, pd_cluster, genai_bench_runner):
""" """
sgl-model-gateway/e2e_test/e2e_http/test_regular_router.py
-182
View File
@@ -1,12 +1,8 @@
import threading
import time import time
from types import SimpleNamespace
import pytest import pytest
import requests import requests
from sglang.test.run_eval import run_eval
def _wait_for_workers( def _wait_for_workers(
base_url: str, expected_count: int, timeout: float = 60.0, headers: dict = None base_url: str, expected_count: int, timeout: float = 60.0, headers: dict = None
@@ -29,43 +25,6 @@ def _wait_for_workers(
) )
def _get_worker_id_by_url(base_url: str, worker_url: str, headers: dict = None) -> str:
r = requests.get(f"{base_url}/workers", headers=headers, timeout=10)
r.raise_for_status()
workers = r.json().get("workers", [])
worker_id = next((w.get("id") for w in workers if w.get("url") == worker_url), None)
if not worker_id:
raise RuntimeError(
f"Could not find worker_id for url={worker_url}. workers={workers}"
)
return worker_id
@pytest.mark.e2e
def test_mmlu(e2e_router_only_rr, e2e_two_workers_dp2, e2e_model):
# Attach two dp=2 workers (total 4 GPUs) to a fresh router-only instance
base = e2e_router_only_rr.url
for w in e2e_two_workers_dp2:
r = requests.post(f"{base}/workers", json={"url": w.url}, timeout=180)
assert (
r.status_code == 202
), f"Expected 202 ACCEPTED, got {r.status_code}: {r.text}"
# Wait for workers to be registered
_wait_for_workers(base, expected_count=2, timeout=60.0)
args = SimpleNamespace(
base_url=base,
model=e2e_model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
temperature=0.1,
)
metrics = run_eval(args)
assert metrics["score"] >= 0.65
@pytest.mark.e2e @pytest.mark.e2e
def test_genai_bench( def test_genai_bench(
e2e_router_only_rr, e2e_two_workers_dp2, e2e_model, genai_bench_runner e2e_router_only_rr, e2e_two_workers_dp2, e2e_model, genai_bench_runner
@@ -95,144 +54,3 @@ def test_genai_bench(
}, },
kill_procs=e2e_two_workers_dp2, kill_procs=e2e_two_workers_dp2,
) )
@pytest.mark.e2e
def test_add_and_remove_worker_live(e2e_router_only_rr, e2e_primary_worker, e2e_model):
base = e2e_router_only_rr.url
worker_url = e2e_primary_worker.url
r = requests.post(f"{base}/workers", json={"url": worker_url}, timeout=180)
assert r.status_code == 202, f"Expected 202 ACCEPTED, got {r.status_code}: {r.text}"
# Wait for worker to be registered
_wait_for_workers(base, expected_count=1, timeout=60.0)
with requests.Session() as s:
for i in range(8):
r = s.post(
f"{base}/v1/completions",
json={
"model": e2e_model,
"prompt": f"x{i}",
"max_tokens": 1,
"stream": False,
},
timeout=120,
)
r.raise_for_status()
# Remove the worker
worker_id = _get_worker_id_by_url(base, worker_url)
r = requests.delete(f"{base}/workers/{worker_id}", timeout=60)
assert r.status_code == 202, f"Expected 202 ACCEPTED, got {r.status_code}: {r.text}"
@pytest.mark.e2e
def test_lazy_fault_tolerance_live(e2e_router_only_rr, e2e_primary_worker, e2e_model):
base = e2e_router_only_rr.url
worker = e2e_primary_worker
r = requests.post(f"{base}/workers", json={"url": worker.url}, timeout=180)
assert r.status_code == 202, f"Expected 202 ACCEPTED, got {r.status_code}: {r.text}"
# Wait for worker to be registered
_wait_for_workers(base, expected_count=1, timeout=60.0)
def killer():
time.sleep(10)
try:
worker.proc.terminate()
except Exception:
pass
t = threading.Thread(target=killer, daemon=True)
t.start()
args = SimpleNamespace(
base_url=base,
model=e2e_model,
eval_name="mmlu",
num_examples=32,
num_threads=16,
temperature=0.0,
)
metrics = run_eval(args)
assert 0.0 <= metrics["score"] <= 1.0
@pytest.mark.e2e
def test_dp_aware_worker_expansion_and_api_key(
e2e_model,
e2e_router_only_rr_dp_aware_api,
e2e_worker_dp2_api,
):
"""
Launch a router-only instance in dp_aware mode and a single worker with dp_size=2
and API key protection. Verify expansion, auth enforcement, and basic eval.
"""
import os
router_url = e2e_router_only_rr_dp_aware_api.url
worker_url = e2e_worker_dp2_api.url
api_key = e2e_router_only_rr_dp_aware_api.api_key
# Attach worker; router should expand to dp_size logical workers
r = requests.post(
f"{router_url}/workers",
json={"url": worker_url, "api_key": api_key},
headers={"Authorization": f"Bearer {api_key}"},
timeout=180,
)
assert r.status_code == 202, f"Expected 202 ACCEPTED, got {r.status_code}: {r.text}"
# Wait for workers to be registered and expanded
_wait_for_workers(
router_url,
expected_count=2,
timeout=60.0,
headers={"Authorization": f"Bearer {api_key}"},
)
# Verify the expanded workers have correct URLs
r = requests.get(
f"{router_url}/workers",
headers={"Authorization": f"Bearer {api_key}"},
timeout=30,
)
r.raise_for_status()
workers = r.json().get("workers", [])
urls = [w["url"] for w in workers]
assert len(urls) == 2
assert set(urls) == {f"{worker_url}@0", f"{worker_url}@1"}
# Verify API key enforcement
# 1) Without Authorization -> Should get 401 Unauthorized
r = requests.post(
f"{router_url}/v1/completions",
json={"model": e2e_model, "prompt": "hi", "max_tokens": 1},
timeout=60,
)
assert r.status_code == 401
# 2) With correct Authorization -> 200
r = requests.post(
f"{router_url}/v1/completions",
json={"model": e2e_model, "prompt": "hi", "max_tokens": 1},
headers={"Authorization": f"Bearer {api_key}"},
timeout=60,
)
assert r.status_code == 200
# Finally, run MMLU eval through the router with auth
os.environ["OPENAI_API_KEY"] = api_key
args = SimpleNamespace(
base_url=router_url,
model=e2e_model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
temperature=0.1,
)
metrics = run_eval(args)
assert metrics["score"] >= 0.65
sgl-model-gateway/e2e_test/e2e_http/test_tls.py
-153
View File
@@ -1,153 +0,0 @@
import datetime
import os
import socket
import subprocess
import sys
import time
from contextlib import closing
import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from urllib3.exceptions import InsecureRequestWarning
# Suppress insecure request warnings due to self-signed cert
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
def find_free_port() -> int:
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1]
def generate_self_signed_cert(cert_path: str, key_path: str) -> None:
"""Generate a self-signed certificate and private key for localhost."""
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
subject = issuer = x509.Name(
[
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "California"),
x509.NameAttribute(NameOID.LOCALITY_NAME, "San Francisco"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SGLang Test"),
x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
]
)
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(datetime.datetime.utcnow())
.not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=10))
.add_extension(
x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False
)
.sign(key, hashes.SHA256())
)
with open(key_path, "wb") as f:
f.write(
key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
)
with open(cert_path, "wb") as f:
f.write(cert.public_bytes(serialization.Encoding.PEM))
def test_tls_server() -> None:
"""End-to-end test for TLS-enabled router startup and basic endpoints."""
cert_path = "cert.pem"
key_path = "key.pem"
generate_self_signed_cert(cert_path, key_path)
port = find_free_port()
cmd = [
sys.executable,
"-m",
"sglang_router.launch_router",
"--worker-urls",
"http://127.0.0.1:9999", # Dummy worker
"--host",
"127.0.0.1",
"--port",
str(port),
"--tls-cert-path",
cert_path,
"--tls-key-path",
key_path,
"--log-level",
"info",
]
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
try:
# Wait for server to start and respond to health check
start_time = time.time()
while time.time() - start_time < 15:
try:
response = requests.get(
f"https://localhost:{port}/health", verify=False, timeout=2
)
if response.status_code == 200:
break
except requests.RequestException:
pass
if proc.poll() is not None:
stdout, stderr = proc.communicate()
raise RuntimeError(
f"Router process died early.\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}"
)
time.sleep(0.5)
else:
raise TimeoutError("Server did not become healthy within 15 seconds")
# Verify basic endpoints work over TLS
models_resp = requests.get(
f"https://localhost:{port}/v1/models", verify=False, timeout=2
)
assert models_resp.status_code in (
200,
503,
) # 503 expected with no healthy workers
# Minimal generate request (should be rejected or queued)
gen_payload = {"model": "dummy", "prompt": "test", "max_new_tokens": 1}
gen_resp = requests.post(
f"https://localhost:{port}/generate",
json=gen_payload,
verify=False,
timeout=2,
)
assert gen_resp.status_code in (
200,
400,
503,
) # Various valid responses with dummy worker
finally:
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
for path in (cert_path, key_path):
if os.path.exists(path):
os.remove(path)