From dce8b0606c06d3a191a24c7b8cbe8e238ab316c9 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Tue, 6 Jan 2026 15:05:08 -0800 Subject: [PATCH] refactor(e2e): keep only benchmark tests in e2e_http, remove redundant tests (#16594) --- .../e2e_test/e2e_http/test_pd_router.py | 19 -- .../e2e_test/e2e_http/test_regular_router.py | 182 ------------------ .../e2e_test/e2e_http/test_tls.py | 153 --------------- 3 files changed, 354 deletions(-) delete mode 100644 sgl-model-gateway/e2e_test/e2e_http/test_tls.py diff --git a/sgl-model-gateway/e2e_test/e2e_http/test_pd_router.py b/sgl-model-gateway/e2e_test/e2e_http/test_pd_router.py index 4827cbb4f..ef7c593ab 100644 --- a/sgl-model-gateway/e2e_test/e2e_http/test_pd_router.py +++ b/sgl-model-gateway/e2e_test/e2e_http/test_pd_router.py @@ -8,8 +8,6 @@ from typing import Optional import pytest import requests -from sglang.test.run_eval import run_eval - logger = logging.getLogger(__name__) @@ -224,23 +222,6 @@ def pd_cluster(e2e_model: str): _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 def test_pd_genai_bench(e2e_model: str, pd_cluster, genai_bench_runner): """ diff --git a/sgl-model-gateway/e2e_test/e2e_http/test_regular_router.py b/sgl-model-gateway/e2e_test/e2e_http/test_regular_router.py index f4707f615..34a4eced0 100644 --- a/sgl-model-gateway/e2e_test/e2e_http/test_regular_router.py +++ b/sgl-model-gateway/e2e_test/e2e_http/test_regular_router.py @@ -1,12 +1,8 @@ -import threading import time -from types import SimpleNamespace import pytest import requests -from sglang.test.run_eval import run_eval - def _wait_for_workers( 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 def test_genai_bench( 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, ) - - -@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 diff --git a/sgl-model-gateway/e2e_test/e2e_http/test_tls.py b/sgl-model-gateway/e2e_test/e2e_http/test_tls.py deleted file mode 100644 index 8a1820b09..000000000 --- a/sgl-model-gateway/e2e_test/e2e_http/test_tls.py +++ /dev/null @@ -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)