""" Config-driven diffusion generation test with pytest parametrization. Each collected request prints a performance log before validation. """ from __future__ import annotations import json import os import queue import threading import time from pathlib import Path from typing import Any, Callable import numpy as np import openai import pytest import requests from openai import OpenAI from sglang.multimodal_gen.runtime.platforms import current_platform from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord from sglang.multimodal_gen.test.server import conftest from sglang.multimodal_gen.test.server.realtime_consistency import ( pop_realtime_key_frames, pop_realtime_perf_stats, validate_realtime_perf_stats, ) from sglang.multimodal_gen.test.server.test_server_utils import ( VALIDATOR_REGISTRY, PerformanceValidator, ServerContext, ServerManager, get_generate_fn, ) from sglang.multimodal_gen.test.server.testcase_configs import ( BASELINE_CONFIG, DiffusionTestCase, PerformanceSummary, ScenarioConfig, get_model_task_type_for_server_args, get_perf_baseline_path, ) from sglang.multimodal_gen.test.test_utils import ( SGL_TEST_FILES_CI_DATA_REVISION, _consistency_gt_filenames, _get_consistency_gt_dir, action_gt_exists, compare_with_gt, extract_key_frames_from_video, get_action_consistency_gt_candidates, get_action_consistency_gt_remote_files, get_consistency_gt_candidates, get_consistency_gt_remote_files, get_consistency_threshold_path, get_consistency_thresholds, get_dynamic_server_port, gt_exists, image_bytes_to_numpy, load_action_consistency_gt, load_consistency_gt, save_consistency_failure_artifact, wait_for_req_perf_record, ) logger = init_logger(__name__) # Track test cases missing estimated_full_test_time_s for time measurement output _MISSING_ESTIMATED_TIME_CASES: set[str] = set() _PENDING_BASELINE_DUMPS: dict[str, tuple[PerformanceSummary, bool]] = {} _OPENAI_REQUEST_TIMEOUT_SECS = float( os.environ.get("SGLANG_TEST_OPENAI_REQUEST_TIMEOUT_SECS", "600") ) _SERVER_EXIT_POLL_INTERVAL_SECS = float( os.environ.get("SGLANG_TEST_SERVER_EXIT_POLL_INTERVAL_SECS", "1") ) _CONTROL_API_TIMEOUT_SECS = float( os.environ.get("SGLANG_TEST_CONTROL_API_TIMEOUT_SECS", "300") ) _SERVER_FATAL_LOG_PATTERNS = ( "terminate called after throwing an instance of", "Fatal Python error:", "Segmentation fault", "Aborted (core dumped)", ) _CASE_LOG_SEPARATOR = "=" * 88 def _print_case_log_separator(case_id: str, state: str) -> None: print( f"\n{_CASE_LOG_SEPARATOR}\n" f"[server-test] {state}: {case_id}\n" f"{_CASE_LOG_SEPARATOR}", flush=True, ) @pytest.fixture def diffusion_server(case: DiffusionTestCase) -> ServerContext: """Start a diffusion server for a single case and tear it down afterwards.""" _fixture_start_time = time.perf_counter() server_args = case.server_args _print_case_log_separator(case.id, "BEGIN diffusion testcase") # Skip ring attention tests on AMD/ROCm - Ring Attention requires Flash Attention # which is not available on AMD. Use Ulysses parallelism instead. if ( current_platform.is_hip() and server_args.ring_degree is not None and server_args.ring_degree > 1 ): pytest.skip( f"Skipping {case.id}: Ring Attention (ring_degree={server_args.ring_degree}) " "requires Flash Attention which is not available on AMD/ROCm" ) default_port = get_dynamic_server_port() port = int(os.environ.get("SGLANG_TEST_SERVER_PORT", default_port)) extra_args = os.environ.get("SGLANG_TEST_SERVE_ARGS", "") extra_args = f"--model-type diffusion {extra_args}".strip() extra_args += f" --num-gpus {server_args.num_gpus}" if server_args.tp_size is not None: extra_args += f" --tp-size {server_args.tp_size}" if server_args.ulysses_degree is not None: extra_args += f" --ulysses-degree {server_args.ulysses_degree}" if server_args.dit_layerwise_offload: extra_args += " --dit-layerwise-offload true" if server_args.dit_offload_prefetch_size: extra_args += ( f" --dit-offload-prefetch-size {server_args.dit_offload_prefetch_size}" ) if server_args.text_encoder_cpu_offload: extra_args += " --text-encoder-cpu-offload" if server_args.ring_degree is not None: extra_args += f" --ring-degree {server_args.ring_degree}" if server_args.cfg_parallel: extra_args += " --enable-cfg-parallel" # LoRA support if server_args.lora_path: extra_args += f" --lora-path {server_args.lora_path}" # Strict ports: fail immediately if port is occupied instead of silently # picking another one (which causes the test client to connect to the wrong server). extra_args += " --strict-ports" # Shape-only mesh cases (e.g. hunyuan3d_shape_gen) validate geometry via # mesh-correctness and must NOT run the paint/texture stages, whose # verification checks texture artifacts (paint_mesh/normal_maps/renderer) # that the shape-only path never produces. Inject a pipeline-config override # disabling paint for these cases. if server_args.custom_validator == "mesh": import json as _json import tempfile as _tempfile _paint_off_cfg = os.path.join( _tempfile.gettempdir(), f"{case.id}_paint_off.json" ) with open(_paint_off_cfg, "w") as _f: _json.dump({"paint_enable": False}, _f) extra_args += f" --config {_paint_off_cfg}" for arg in server_args.extras: extra_args += f" {arg}" # Build custom environment variables env_vars = {} if server_args.enable_cache_dit: env_vars["SGLANG_CACHE_DIT_ENABLED"] = "true" env_vars.update(server_args.env_vars) # start server wait_deadline = float(os.environ.get("SGLANG_TEST_WAIT_SECS", "1200")) logger.info( "[server-test] Starting server for test case: %s\n" " Model: %s\n" " Port: %s\n" " Wait deadline: %ss\n" " Extra args: %s\n" " Num GPUs: %s", case.id, server_args.model_path, port, wait_deadline, extra_args, server_args.num_gpus, ) manager = ServerManager( model=server_args.model_path, port=port, wait_deadline=wait_deadline, extra_args=extra_args, env_vars=env_vars, ) try: ctx = manager.start() except (RuntimeError, TimeoutError) as exc: # Auto-skip when the installed diffusers version lacks the required # pipeline class. This avoids hard failures when a model needs a # newer diffusers release than what is currently installed in CI. msg = str(exc) if "not found in diffusers" in msg or ( "has no attribute" in msg and "diffusers" in msg.lower() ): pytest.skip( f"Skipping {case.id}: required diffusers pipeline class " f"is not available in the installed version. " f"Upgrade diffusers to enable this test." ) _print_case_log_separator(case.id, "FAILED during server startup") raise try: yield ctx finally: ctx.cleanup() _fixture_end_time = time.perf_counter() _measured_full_time = _fixture_end_time - _fixture_start_time is_baseline_generation_mode = os.environ.get("SGLANG_GEN_BASELINE", "0") == "1" pending_dump = _PENDING_BASELINE_DUMPS.pop(case.id, None) if pending_dump is not None: summary, missing_scenario = pending_dump DiffusionServerBase()._dump_baseline_for_testcase( case, summary, missing_scenario=missing_scenario, measured_full_time=_measured_full_time, ) scenario = BASELINE_CONFIG.scenarios.get(case.id) needs_estimated_time = ( scenario is None or scenario.estimated_full_test_time_s is None ) if needs_estimated_time and not is_baseline_generation_mode: _MISSING_ESTIMATED_TIME_CASES.add(case.id) logger.error( f'\n{"=" * 60}\n' f'Add "estimated_full_test_time_s" to scenario "{case.id}":\n\n' f"File: {get_perf_baseline_path()}\n\n" f' "{case.id}": {{\n' f" ...\n" f' "estimated_full_test_time_s": {_measured_full_time:.1f}\n' f" }}\n" f'{"=" * 60}\n' ) _print_case_log_separator(case.id, "END diffusion testcase") class DiffusionServerBase: """Performance tests for all diffusion models/scenarios. This single test class runs against all cases defined in ONE_GPU_CASES. Each case gets its own server instance via the parametrized fixture. """ _perf_results: list[dict[str, Any]] = [] _pytest_config = None # Store pytest config for stash access @classmethod def setup_class(cls): cls._perf_results = [] @classmethod def teardown_class(cls): print( f"\n[DEBUG teardown_class] Called for {cls.__name__}, _perf_results has {len(cls._perf_results)} entries" ) if cls._pytest_config: # Add results to pytest stash (shared across all import contexts) for result in cls._perf_results: result["class_name"] = cls.__name__ conftest.add_perf_results(cls._pytest_config, cls._perf_results) print( f"[DEBUG teardown_class] Added {len(cls._perf_results)} results to stash" ) else: print( "[DEBUG teardown_class] No pytest_config available, skipping stash update" ) @pytest.fixture(autouse=True) def _capture_pytest_config(self, request): """Capture pytest config for use in teardown_class.""" self.__class__._pytest_config = request.config def _client(self, ctx: ServerContext) -> OpenAI: """Get OpenAI client for the server.""" return OpenAI( api_key="sglang-anything", base_url=f"http://localhost:{ctx.port}/v1", timeout=_OPENAI_REQUEST_TIMEOUT_SECS, max_retries=0, ) def _fail_if_server_stopped_or_crashed( self, ctx: ServerContext, case_id: str ) -> None: returncode = ctx.process.poll() if returncode is None: tail = ctx.log_tail() for pattern in _SERVER_FATAL_LOG_PATTERNS: if pattern in tail: pytest.fail( f"{case_id}: server reported a fatal backend error during " f"generation: {pattern}\n\nServer log tail:\n{tail}", pytrace=False, ) return tail = ctx.log_tail() message = ( f"{case_id}: server process exited during generation " f"(code {returncode})." ) if tail: message += f"\n\nServer log tail:\n{tail}" pytest.fail(message, pytrace=False) def _run_generation_with_server_watchdog( self, ctx: ServerContext, case_id: str, generate_fn: Callable[[str, openai.Client], tuple[str, bytes]], client: openai.Client, ) -> tuple[str, bytes]: result_queue: queue.Queue[tuple[str, tuple[str, bytes] | BaseException]] = ( queue.Queue(maxsize=1) ) def _target() -> None: try: result_queue.put(("ok", generate_fn(case_id, client))) except BaseException as exc: result_queue.put(("error", exc)) # native backend crashes can leave the HTTP client blocked until its read # timeout; keep the request in a daemon thread so the main test thread can # fail as soon as the server subprocess exits thread = threading.Thread( target=_target, name=f"diffusion-generation-{case_id}", daemon=True, ) thread.start() while True: try: state, payload = result_queue.get( timeout=_SERVER_EXIT_POLL_INTERVAL_SECS ) except queue.Empty: self._fail_if_server_stopped_or_crashed(ctx, case_id) continue if state == "ok": if isinstance(payload, BaseException): raise payload return payload self._fail_if_server_stopped_or_crashed(ctx, case_id) if not isinstance(payload, BaseException): pytest.fail(f"{case_id}: invalid generation result state: {state}") raise payload def run_and_collect( self, ctx: ServerContext, case_id: str, generate_fn: Callable[[str, openai.Client], tuple[str, bytes]], collect_perf: bool = True, ) -> tuple[RequestPerfRecord | None, bytes]: """Run generation and optionally collect performance records. Returns: Tuple of (performance_record, content_bytes) """ client = self._client(ctx) rid, content = self._run_generation_with_server_watchdog( ctx, case_id, generate_fn, client ) if not collect_perf: return None, content log_path = ctx.perf_log_path log_wait_timeout = 30 req_perf_record = wait_for_req_perf_record( rid, log_path, timeout=log_wait_timeout, ) return (req_perf_record, content) def _validate_and_record( self, case: DiffusionTestCase, perf_record: RequestPerfRecord, ) -> None: """Validate metrics and record results.""" is_baseline_generation_mode = os.environ.get("SGLANG_GEN_BASELINE", "0") == "1" scenario = BASELINE_CONFIG.scenarios.get(case.id) missing_scenario = False if scenario is None: # Create dummy scenario to allow metric collection scenario = type( "DummyScenario", (), { "expected_e2e_ms": 0, "expected_avg_denoise_ms": 0, "expected_median_denoise_ms": 0, "stages_ms": {}, "denoise_step_ms": {}, }, )() if not is_baseline_generation_mode: missing_scenario = True if ( not missing_scenario and not is_baseline_generation_mode and scenario.estimated_full_test_time_s is None ): _MISSING_ESTIMATED_TIME_CASES.add(case.id) validator_name = case.server_args.custom_validator or "default" validator_class = VALIDATOR_REGISTRY.get(validator_name, PerformanceValidator) validator = validator_class( scenario=scenario, tolerances=BASELINE_CONFIG.tolerances, step_fractions=BASELINE_CONFIG.step_fractions, ) summary = validator.collect_metrics(perf_record) self._print_performance_log(case, summary, scenario) if case.run_perf_check: if is_baseline_generation_mode: _PENDING_BASELINE_DUMPS[case.id] = (summary, missing_scenario) return if missing_scenario: self._dump_baseline_for_testcase(case, summary, missing_scenario) if missing_scenario: pytest.fail( f"Testcase '{case.id}' not found in {get_perf_baseline_path()}" ) return # only run performance validation if run_perf_check is True try: validator.validate(perf_record, case.sampling_params.num_frames) except AssertionError as e: logger.error(f"Performance validation failed for {case.id}:\n{e}") self._dump_baseline_for_testcase(case, summary, missing_scenario) raise result = { "test_name": case.id, "modality": case.server_args.modality, "e2e_ms": summary.e2e_ms, "avg_denoise_ms": summary.avg_denoise_ms, "median_denoise_ms": summary.median_denoise_ms, "stage_metrics": summary.stage_metrics, "sampled_steps": summary.sampled_steps, } # video-specific metrics if summary.frames_per_second: result.update( { "frames_per_second": summary.frames_per_second, "total_frames": summary.total_frames, "avg_frame_time_ms": summary.avg_frame_time_ms, } ) self.__class__._perf_results.append(result) print( f"[DEBUG _validate_and_record] Appended result for {case.id}, class {self.__class__.__name__} now has {len(self.__class__._perf_results)} results" ) def _print_performance_log( self, case: DiffusionTestCase, summary: PerformanceSummary, scenario: ScenarioConfig | None, ) -> None: lines = [ "", f"--- Performance Log: {case.id} ---", ( f" e2e={summary.e2e_ms:.2f}ms, " f"avg_denoise={summary.avg_denoise_ms:.2f}ms, " f"median_denoise={summary.median_denoise_ms:.2f}ms" ), ] if scenario is not None: lines.append( " baseline: " f"e2e={scenario.expected_e2e_ms:.2f}ms, " f"avg_denoise={scenario.expected_avg_denoise_ms:.2f}ms, " f"median_denoise={scenario.expected_median_denoise_ms:.2f}ms" ) if summary.stage_metrics: stages = ", ".join( f"{name}={duration:.2f}ms" for name, duration in summary.stage_metrics.items() ) lines.append(f" stages: {stages}") if summary.all_denoise_steps: # ci retries need the exact outlier, not only sampled checkpoints steps = ", ".join( f"{idx}={duration:.2f}ms" for idx, duration in sorted(summary.all_denoise_steps.items()) ) lines.append(f" denoise_steps: {steps}") lines.append(f"--- End Performance Log: {case.id} ---") print("\n".join(lines), flush=True) def _dump_baseline_for_testcase( self, case: DiffusionTestCase, summary: PerformanceSummary, missing_scenario: bool = False, measured_full_time: float | None = None, ) -> None: """Dump performance metrics as a JSON scenario for baselines.""" import json denoise_steps_formatted = { str(k): round(v, 2) for k, v in summary.all_denoise_steps.items() } stages_formatted = {k: round(v, 2) for k, v in summary.stage_metrics.items()} baseline = { "stages_ms": stages_formatted, "denoise_step_ms": denoise_steps_formatted, "expected_e2e_ms": round(summary.e2e_ms, 2), "expected_avg_denoise_ms": round(summary.avg_denoise_ms, 2), "expected_median_denoise_ms": round(summary.median_denoise_ms, 2), } if measured_full_time is not None: baseline["estimated_full_test_time_s"] = round(measured_full_time, 1) # Video-specific metrics if case.server_args.modality == "video": if "per_frame_generation" not in baseline["stages_ms"]: baseline["stages_ms"]["per_frame_generation"] = ( round(summary.avg_frame_time_ms, 2) if summary.avg_frame_time_ms else None ) action = "add" if missing_scenario else "update" output = f""" {action} this baseline in the "scenarios" section of {get_perf_baseline_path()}: "{case.id}": {json.dumps(baseline, indent=4)} """ logger.error(output) def _validate_consistency( self, case: DiffusionTestCase, content: bytes, ) -> None: """Validate output consistency against ground truth using CLIP similarity.""" if os.environ.get("SGLANG_SKIP_CONSISTENCY", "0") == "1": logger.info( f"[Consistency] Skipping consistency check for {case.id} (SGLANG_SKIP_CONSISTENCY=1)" ) return if not content: logger.warning( f"[Consistency] Skipping consistency check for {case.id}: " "content is empty (generation may have timed out)" ) return if case.server_args.modality == "action": self._validate_action_consistency(case, content) return num_gpus = case.server_args.num_gpus is_video = case.server_args.modality == "video" output_format = case.sampling_params.output_format if not gt_exists( case.id, num_gpus, is_video=is_video, output_format=output_format ): if _get_consistency_gt_dir() is not None: names = ", ".join( get_consistency_gt_candidates( case.id, num_gpus, is_video, output_format ) ) else: names = ", ".join( _consistency_gt_filenames( case.id, num_gpus, is_video, output_format ) ) logger.error(f""" --- MISSING GROUND TRUTH DETECTED --- GT image(s) not found for '{case.id}'. Add the expected file(s) to sgl-project/ci-data in diffusion-ci/consistency_gt/sglang_generated/ with naming (n=num_gpus). Image: {case.id}_{{n}}gpu. (ext from output_format: png, jpg, webp) Video: {case.id}_{{n}}gpu_frame_0.png, {case.id}_{{n}}gpu_frame_mid.png, {case.id}_{{n}}gpu_frame_last.png For this case, expected file(s): {names} Repository: https://github.com/sgl-project/ci-data (path: diffusion-ci/consistency_gt/sglang_generated/, with optional platform subdirectories such as 5090/) Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} (Optional) Per-case override in {get_consistency_threshold_path()}: "cases": {{ "{case.id}": {{ "clip_threshold": 0.92, "ssim_threshold": 0.95, "psnr_threshold": 28.0, "mean_abs_diff_threshold": 8.0 }} }} """) pytest.fail( f"GT not found for {case.id}. See logs for instructions to add GT." ) gt_data = load_consistency_gt( case.id, num_gpus, is_video=is_video, output_format=output_format ) thresholds = get_consistency_thresholds(case.id, is_video=is_video) if is_video: output_frames = pop_realtime_key_frames(case.id) if output_frames is None: output_frames = extract_key_frames_from_video(content) else: output_frames = [image_bytes_to_numpy(content)] result = compare_with_gt( output_frames=output_frames, gt_data=gt_data, thresholds=thresholds, case_id=case.id, ) if not result.passed: failed_frames = [] gt_remote_files = get_consistency_gt_remote_files( case.id, num_gpus, is_video=is_video, output_format=output_format, ) artifact_path = save_consistency_failure_artifact( artifact_dir=os.environ.get("SGLANG_DIFFUSION_ARTIFACT_DIR"), case_id=case.id, num_gpus=num_gpus, output_frames=output_frames, gt_data=gt_data, result=result, is_video=is_video, output_format=output_format, gt_remote_files=gt_remote_files, ) if artifact_path is not None: logger.info( "[Artifact] Saved consistency failure comparison: %s", artifact_path, ) gt_remote_info = "\n".join( f" - {filename}: {url}" for filename, url in gt_remote_files ) for metric in result.frame_metrics: failed_metrics = [] if not metric.clip_passed: failed_metrics.append("clip") if not metric.ssim_passed: failed_metrics.append("ssim") if not metric.psnr_passed: failed_metrics.append("psnr") if not metric.mean_abs_diff_passed: failed_metrics.append("mean_abs_diff") if failed_metrics: failed_frames.append( f" - f{metric.frame_index} " f"[{', '.join(failed_metrics)}] " f"clip={metric.clip_similarity:.4f} " f"ssim={metric.ssim:.4f} " f"psnr={metric.psnr:.4f} " f"mean_abs_diff={metric.mean_abs_diff:.4f}" ) pytest.fail( f"Consistency check failed for {case.id}:\n" f" Metrics: sim={result.min_similarity:.4f}, " f"ssim={result.min_ssim:.4f}, " f"psnr={result.min_psnr:.4f}, " f"mean_abs_diff={result.max_mean_abs_diff:.4f}\n" f" Thresholds: clip>={result.thresholds.clip_threshold}, " f"ssim>={result.thresholds.ssim_threshold}, " f"psnr>={result.thresholds.psnr_threshold}, " f"mean_abs_diff<={result.thresholds.mean_abs_diff_threshold}\n" f" Failed frames:\n" + "\n".join(failed_frames) + f"\n Compared GT files and links:\n{gt_remote_info}" ) logger.info( f"[Consistency] {case.id}: PASSED " f"(min_similarity={result.min_similarity:.4f}, " f"min_ssim={result.min_ssim:.4f}, " f"min_psnr={result.min_psnr:.4f}, " f"max_mean_abs_diff={result.max_mean_abs_diff:.4f})" ) def _extract_action_array( self, payload: dict[str, Any], expected_horizon: int, expected_dim: int, ) -> np.ndarray: action = payload["data"][0]["action"] values = action["values"] assert action["shape"] == [expected_horizon, expected_dim] array = np.asarray(values, dtype=np.float32) assert array.shape == (expected_horizon, expected_dim) assert np.isfinite(array).all() return array def _validate_action_consistency( self, case: DiffusionTestCase, content: bytes, ) -> None: payload = json.loads(content.decode("utf-8")) expected_horizon = int(case.sampling_params.extras.get("action_horizon", 50)) expected_dim = int(case.sampling_params.extras.get("action_dim", 32)) output = self._extract_action_array(payload, expected_horizon, expected_dim) num_gpus = case.server_args.num_gpus if not action_gt_exists(case.id, num_gpus): names = ", ".join(get_action_consistency_gt_candidates(case.id, num_gpus)) logger.error(f""" --- MISSING ACTION GROUND TRUTH DETECTED --- GT action JSON not found for '{case.id}'. Add the expected file to sgl-project/ci-data in diffusion-ci/consistency_gt/sglang_generated/ with naming: Action: {case.id}_{{n}}gpu.json For this case, expected file(s): {names} Repository: https://github.com/sgl-project/ci-data (path: diffusion-ci/consistency_gt/sglang_generated/, with optional platform subdirectories such as 5090/) Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} """) pytest.fail( f"GT action JSON not found for {case.id}. See logs for instructions to add GT." ) gt_payload = load_action_consistency_gt(case.id, num_gpus) gt = self._extract_action_array(gt_payload, expected_horizon, expected_dim) abs_diff = np.abs(output - gt) max_abs_diff = float(abs_diff.max()) mean_abs_diff = float(abs_diff.mean()) max_abs_threshold = float( case.sampling_params.extras.get("action_max_abs_diff_threshold", 0.05) ) mean_abs_threshold = float( case.sampling_params.extras.get("action_mean_abs_diff_threshold", 0.005) ) if max_abs_diff > max_abs_threshold or mean_abs_diff > mean_abs_threshold: gt_remote_info = "\n".join( f" - {filename}: {url}" for filename, url in get_action_consistency_gt_remote_files( case.id, num_gpus, ) ) pytest.fail( f"Action consistency check failed for {case.id}:\n" f" max_abs_diff={max_abs_diff:.6f} " f"(threshold {max_abs_threshold:.6f})\n" f" mean_abs_diff={mean_abs_diff:.6f} " f"(threshold {mean_abs_threshold:.6f})\n" f" Compared GT files and links:\n{gt_remote_info}" ) logger.info( "[Consistency] %s: PASSED action GT check " "(shape=%sx%s, max_abs_diff=%.6f, mean_abs_diff=%.6f)", case.id, expected_horizon, expected_dim, max_abs_diff, mean_abs_diff, ) def _save_gt_output( self, case: DiffusionTestCase, content: bytes, ) -> None: """Save generated content as ground truth files. Args: case: Test case configuration content: Generated content bytes (image or video) """ gt_output_dir = os.environ.get("SGLANG_GT_OUTPUT_DIR") if not gt_output_dir: logger.error("SGLANG_GT_OUTPUT_DIR not set, cannot save GT output") return out_dir = Path(gt_output_dir) out_dir.mkdir(parents=True, exist_ok=True) num_gpus = case.server_args.num_gpus is_video = case.server_args.modality == "video" if case.server_args.modality == "action": output_path = out_dir / f"{case.id}_{num_gpus}gpu.json" output_path.write_bytes(content) logger.info(f"Saved GT action JSON: {output_path}") return if is_video: # realtime consistency uses websocket raw frames to avoid lossy mp4 drift frames = pop_realtime_key_frames(case.id) if frames is None: frames = extract_key_frames_from_video( content, num_frames=case.sampling_params.num_frames ) if len(frames) != 3: logger.warning( f"{case.id}: expected 3 frames, got {len(frames)}, skipping frame save" ) return # Save frames (reuse naming from _consistency_gt_filenames) filenames = _consistency_gt_filenames(case.id, num_gpus, is_video=True) from PIL import Image for frame, fn in zip(frames, filenames): frame_path = out_dir / fn Image.fromarray(frame).save(frame_path) logger.info(f"Saved GT frame: {frame_path}") else: # Save image from sglang.multimodal_gen.test.test_utils import detect_image_format detected_format = detect_image_format(content) filenames = _consistency_gt_filenames( case.id, num_gpus, is_video=False, output_format=detected_format ) output_path = out_dir / filenames[0] output_path.write_bytes(content) logger.info(f"Saved GT image: {output_path} (format: {detected_format})") def _validate_lora_consistency( self, case: DiffusionTestCase, content: bytes, operation: str ) -> None: if not case.run_consistency_check: logger.info( "[LoRA Consistency] Skipping %s consistency for %s: disabled for case", operation, case.id, ) return logger.info( "[LoRA Consistency] Validating %s output for %s", operation, case.id ) self._validate_consistency(case, content) def _test_lora_api_functionality( self, ctx: ServerContext, case: DiffusionTestCase, generate_fn: Callable[[str, openai.Client], tuple[str, bytes]], ) -> None: """ Test LoRA API functionality with end-to-end validation: merge, unmerge, and set_lora. This test verifies that each API call succeeds AND that generation works after each operation. """ base_url = f"http://localhost:{ctx.port}/v1" client = self._client(ctx) # Test 1: unmerge_lora_weights - API should succeed and generation should work logger.info("[LoRA E2E] Testing unmerge_lora_weights for %s", case.id) resp = requests.post( f"{base_url}/unmerge_lora_weights", timeout=_CONTROL_API_TIMEOUT_SECS ) assert resp.status_code == 200, f"unmerge_lora_weights failed: {resp.text}" logger.info("[LoRA E2E] Verifying generation after unmerge for %s", case.id) rid_after_unmerge, _ = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid_after_unmerge is not None, "Generation after unmerge failed" logger.info("[LoRA E2E] Generation after unmerge succeeded") # Test 2: merge_lora_weights - API should succeed and generation should work logger.info("[LoRA E2E] Testing merge_lora_weights for %s", case.id) resp = requests.post( f"{base_url}/merge_lora_weights", timeout=_CONTROL_API_TIMEOUT_SECS ) assert resp.status_code == 200, f"merge_lora_weights failed: {resp.text}" logger.info("[LoRA E2E] Verifying generation after re-merge for %s", case.id) rid_after_merge, content_after_merge = ( self._run_generation_with_server_watchdog(ctx, case.id, generate_fn, client) ) assert rid_after_merge is not None, "Generation after merge failed" self._validate_lora_consistency(case, content_after_merge, "merge_lora_weights") logger.info("[LoRA E2E] Generation after merge succeeded") # Test 3: set_lora (re-set the same adapter) - API should succeed and generation should work logger.info("[LoRA E2E] Testing set_lora for %s", case.id) resp = requests.post( f"{base_url}/set_lora", json={"lora_nickname": "default"}, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert resp.status_code == 200, f"set_lora failed: {resp.text}" logger.info("[LoRA E2E] Verifying generation after set_lora for %s", case.id) rid_after_set, content_after_set = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid_after_set is not None, "Generation after set_lora failed" self._validate_lora_consistency(case, content_after_set, "set_lora") logger.info("[LoRA E2E] Generation after set_lora succeeded") # Test 4: list_loras - API should return the expected list of LoRA adapters logger.info("[LoRA E2E] Testing list_loras for %s", case.id) resp = requests.get(f"{base_url}/list_loras", timeout=_CONTROL_API_TIMEOUT_SECS) assert resp.status_code == 200, f"list_loras failed: {resp.text}" lora_info = resp.json() logger.info("[LoRA E2E] list_loras returned %s", lora_info) assert ( isinstance(lora_info["loaded_adapters"], list) and len(lora_info["loaded_adapters"]) > 0 ), "loaded_adapters should be a non-empty list" assert any( a.get("nickname") == "default" for a in lora_info["loaded_adapters"] ), f"nickname 'default' not found in loaded_adapters: {lora_info['loaded_adapters']}" logger.info("[LoRA E2E] list_loras returned expected LoRA adapters") logger.info("[LoRA E2E] All LoRA API E2E tests passed for %s", case.id) def _test_lora_dynamic_switch_e2e( self, ctx: ServerContext, case: DiffusionTestCase, generate_fn: Callable[[str, openai.Client], tuple[str, bytes]], second_lora_path: str, ) -> None: """ Test dynamic LoRA switching with end-to-end validation. This test verifies that switching between LoRA adapters works correctly and generation succeeds after each switch. """ base_url = f"http://localhost:{ctx.port}/v1" client = self._client(ctx) # Test 1: Generate with initial LoRA logger.info( "[LoRA Switch E2E] Testing generation with initial LoRA for %s", case.id ) rid_initial, content_initial = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid_initial is not None, "Generation with initial LoRA failed" self._validate_lora_consistency( case, content_initial, "dynamic switch initial LoRA" ) logger.info("[LoRA Switch E2E] Generation with initial LoRA succeeded") # Test 2: Switch to second LoRA and generate logger.info( "[LoRA Switch E2E] Switching to second LoRA adapter for %s", case.id ) resp = requests.post( f"{base_url}/set_lora", json={"lora_nickname": "lora2", "lora_path": second_lora_path}, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 200 ), f"set_lora to second adapter failed: {resp.text}" logger.info( "[LoRA Switch E2E] Verifying generation with second LoRA for %s", case.id ) rid_second, _ = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid_second is not None, "Generation with second LoRA failed" logger.info("[LoRA Switch E2E] Generation with second LoRA succeeded") # Test 3: Switch back to original LoRA and generate logger.info("[LoRA Switch E2E] Switching back to original LoRA for %s", case.id) resp = requests.post( f"{base_url}/set_lora", json={"lora_nickname": "default"}, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert resp.status_code == 200, f"set_lora back to default failed: {resp.text}" logger.info( "[LoRA Switch E2E] Verifying generation after switching back for %s", case.id, ) rid_switched_back, content_switched_back = ( self._run_generation_with_server_watchdog(ctx, case.id, generate_fn, client) ) assert rid_switched_back is not None, "Generation after switching back failed" self._validate_lora_consistency( case, content_switched_back, "dynamic switch default LoRA" ) logger.info("[LoRA Switch E2E] Generation after switching back succeeded") logger.info( "[LoRA Switch E2E] All dynamic switch E2E tests passed for %s", case.id ) def _test_dynamic_lora_loading( self, ctx: ServerContext, case: DiffusionTestCase, ) -> None: """ Test dynamic LoRA loading after server startup. This test reproduces the LayerwiseOffload + set_lora issue: - Server starts WITHOUT lora_path (LayerwiseOffloadManager initializes first) - Then set_lora is called via API to load LoRA dynamically - This tests the interaction between layerwise offload and dynamic LoRA loading """ base_url = f"http://localhost:{ctx.port}/v1" dynamic_lora_path = case.server_args.dynamic_lora_path # Call set_lora to load LoRA dynamically after server startup logger.info( "[Dynamic LoRA] Loading LoRA dynamically via set_lora API for %s", case.id ) logger.info("[Dynamic LoRA] LoRA path: %s", dynamic_lora_path) resp = requests.post( f"{base_url}/set_lora", json={"lora_nickname": "default", "lora_path": dynamic_lora_path}, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert resp.status_code == 200, f"Dynamic set_lora failed: {resp.text}" logger.info("[Dynamic LoRA] set_lora succeeded for %s", case.id) def _test_multi_lora_e2e( self, ctx: ServerContext, case: DiffusionTestCase, generate_fn: Callable[[str, openai.Client], tuple[str, bytes]], first_lora_path: str, second_lora_path: str, ) -> None: """ Test multiple LoRA adapters with different set_lora input scenarios. Tests: basic multi-LoRA, different strengths, cached adapters, switch back to single. """ base_url = f"http://localhost:{ctx.port}/v1" client = self._client(ctx) # Test 1: Basic multi-LoRA with list format resp = requests.post( f"{base_url}/set_lora", json={ "lora_nickname": ["default", "lora2"], "lora_path": [first_lora_path, second_lora_path], "target": "all", "strength": [0.5, 0.5], }, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 200 ), f"set_lora with multiple adapters failed: {resp.text}" rid, _ = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid is not None # Test 2: Different strengths resp = requests.post( f"{base_url}/set_lora", json={ "lora_nickname": ["default", "lora2"], "lora_path": [first_lora_path, second_lora_path], "target": "all", "strength": [0.6, 0.35], }, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 200 ), f"set_lora with different strengths failed: {resp.text}" rid, _ = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid is not None # Test 3: Different targets requests.post( f"{base_url}/set_lora", json={"lora_nickname": "default"}, timeout=_CONTROL_API_TIMEOUT_SECS, ) resp = requests.post( f"{base_url}/set_lora", json={ "lora_nickname": ["default", "lora2"], "lora_path": [first_lora_path, second_lora_path], "target": ["transformer", "transformer_2"], "strength": [0.6, 0.35], }, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 200 ), f"set_lora with cached adapters failed: {resp.text}" rid, _ = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid is not None # Test 4: Switch back to single LoRA resp = requests.post( f"{base_url}/set_lora", json={"lora_nickname": "default"}, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 200 ), f"set_lora back to single adapter failed: {resp.text}" rid, content = self._run_generation_with_server_watchdog( ctx, case.id, generate_fn, client ) assert rid is not None self._validate_lora_consistency(case, content, "multi-LoRA default adapter") logger.info("[Multi-LoRA] All multi-LoRA tests passed for %s", case.id) def _test_v1_models_endpoint( self, ctx: ServerContext, case: DiffusionTestCase ) -> None: """ Test /v1/models endpoint returns OpenAI-compatible response. This endpoint is required for sgl-model-gateway router compatibility. """ base_url = f"http://localhost:{ctx.port}" # Test GET /v1/models logger.info("[Models API] Testing GET /v1/models for %s", case.id) resp = requests.get(f"{base_url}/v1/models", timeout=_CONTROL_API_TIMEOUT_SECS) assert resp.status_code == 200, f"/v1/models failed: {resp.text}" data = resp.json() assert ( data["object"] == "list" ), f"Expected object='list', got {data.get('object')}" assert len(data["data"]) >= 1, "Expected at least one model in response" model = data["data"][0] assert "id" in model, "Model missing 'id' field" assert ( model["object"] == "model" ), f"Expected object='model', got {model.get('object')}" assert ( model["id"] == case.server_args.model_path ), f"Model ID mismatch: expected {case.server_args.model_path}, got {model['id']}" # Verify extended diffusion-specific fields assert "num_gpus" in model, "Model missing 'num_gpus' field" assert "task_type" in model, "Model missing 'task_type' field" assert "dit_precision" in model, "Model missing 'dit_precision' field" assert "vae_precision" in model, "Model missing 'vae_precision' field" assert ( model["num_gpus"] == case.server_args.num_gpus ), f"num_gpus mismatch: expected {case.server_args.num_gpus}, got {model['num_gpus']}" expected_task_type = get_model_task_type_for_server_args(case.server_args).name assert model["task_type"] == expected_task_type, ( f"task_type mismatch: expected {expected_task_type}, " f"got {model['task_type']}" ) logger.info( "[Models API] GET /v1/models returned valid response with extended fields" ) # Test GET /v1/models/{model_path} model_path = model["id"] logger.info("[Models API] Testing GET /v1/models/%s", model_path) resp = requests.get( f"{base_url}/v1/models/{model_path}", timeout=_CONTROL_API_TIMEOUT_SECS ) assert resp.status_code == 200, f"/v1/models/{model_path} failed: {resp.text}" single_model = resp.json() assert single_model["id"] == model_path, "Single model ID mismatch" assert single_model["object"] == "model", "Single model object type mismatch" # Verify extended fields on single model endpoint too assert "num_gpus" in single_model, "Single model missing 'num_gpus' field" assert "task_type" in single_model, "Single model missing 'task_type' field" assert single_model["task_type"] == expected_task_type, ( f"Single model task_type mismatch: expected {expected_task_type}, " f"got {single_model['task_type']}" ) logger.info( "[Models API] GET /v1/models/{model_path} returned valid response with extended fields" ) # Test GET /v1/models/{non_existent_model} returns 404 logger.info("[Models API] Testing GET /v1/models/non_existent_model") resp = requests.get( f"{base_url}/v1/models/non_existent_model", timeout=_CONTROL_API_TIMEOUT_SECS, ) assert resp.status_code == 404, f"Expected 404, got {resp.status_code}" error_data = resp.json() assert "error" in error_data, "404 response missing 'error' field" assert ( error_data["error"]["code"] == "model_not_found" ), f"Incorrect error code: {error_data['error'].get('code')}" logger.info("[Models API] GET /v1/models/non_existent returns 404 as expected") logger.info("[Models API] All /v1/models tests passed for %s", case.id) def _test_t2v_rejects_input_reference( self, ctx: ServerContext, case: DiffusionTestCase ) -> None: if case.server_args.modality != "video": return base_url = f"http://localhost:{ctx.port}" resp = requests.get(f"{base_url}/v1/models", timeout=_CONTROL_API_TIMEOUT_SECS) assert resp.status_code == 200, f"/v1/models failed: {resp.text}" data = resp.json().get("data", []) if not data: pytest.fail("/v1/models returned empty model list") task_type = data[0].get("task_type") if task_type != "T2V": return prompt = case.sampling_params.prompt or "test" payload = {"prompt": prompt, "input_reference": "dummy"} if case.sampling_params.output_size: payload["size"] = case.sampling_params.output_size resp = requests.post( f"{base_url}/v1/videos", json=payload, timeout=_CONTROL_API_TIMEOUT_SECS, ) assert ( resp.status_code == 400 ), f"Expected 400 for T2V input_reference, got {resp.status_code}: {resp.text}" detail = resp.json().get("detail", "") assert ( "input_reference is not supported" in detail ), f"Unexpected error detail for T2V input_reference: {detail}" def test_diffusion_generation( self, case: DiffusionTestCase, diffusion_server: ServerContext, ): """Single parametrized test that runs for all cases. This test performs: 1. Generation 2. Performance validation against baselines 3. Consistency validation against ground truth Pytest will execute this test once per case in ONE_GPU_CASES, with test IDs like: - test_diffusion_generation[qwen_image_text] - test_diffusion_generation[qwen_image_edit] - etc. """ try: self._test_diffusion_generation_impl(case, diffusion_server) except pytest.skip.Exception: _print_case_log_separator(case.id, "SKIPPED diffusion testcase") raise except (KeyboardInterrupt, SystemExit): raise except BaseException: _print_case_log_separator(case.id, "FAILED diffusion testcase") raise else: _print_case_log_separator(case.id, "PASSED diffusion testcase") def _test_diffusion_generation_impl( self, case: DiffusionTestCase, diffusion_server: ServerContext, ): # Check if we're in GT generation mode is_gt_gen_mode = os.environ.get("SGLANG_GEN_GT", "0") == "1" # GT generation also needs the dynamic set_lora step before generation. if case.run_lora_dynamic_load_check: self._test_dynamic_lora_loading(diffusion_server, case) generate_fn = get_generate_fn( model_path=case.server_args.model_path, modality=case.server_args.modality, sampling_params=case.sampling_params, ) # Single generation - output is reused for both validations is_realtime_case = case.sampling_params.realtime_num_chunks is not None perf_record, content = self.run_and_collect( diffusion_server, case.id, generate_fn, collect_perf=not is_gt_gen_mode and not is_realtime_case, ) if is_gt_gen_mode: # GT generation mode: save output and skip all validations/tests self._save_gt_output(case, content) return failures: list[tuple[str, str]] = [] def run_case_check(name: str, fn: Callable[[], None]) -> None: try: fn() except BaseException as exc: if isinstance(exc, (KeyboardInterrupt, SystemExit)): raise failures.append((name, str(exc))) if is_realtime_case: run_case_check( "performance", lambda: validate_realtime_perf_stats( case.id, pop_realtime_perf_stats(case.id), case.sampling_params.realtime_perf_thresholds, ignore_initial_chunks=( case.sampling_params.realtime_perf_ignore_initial_chunks ), ), ) else: run_case_check( "performance", lambda: self._validate_and_record(case, perf_record), ) if case.server_args.custom_validator == "mesh": from sglang.multimodal_gen.test.server.test_server_utils import ( MESH_OUTPUT_PATHS, validate_mesh_correctness, ) def validate_mesh_output() -> None: mesh_path = MESH_OUTPUT_PATHS.pop(case.id, None) if mesh_path: validate_mesh_correctness(mesh_path) run_case_check("mesh correctness", validate_mesh_output) if case.run_models_api_check: run_case_check( "/v1/models endpoint", lambda: self._test_v1_models_endpoint(diffusion_server, case), ) if case.run_t2v_input_reference_check: run_case_check( "t2v input_reference rejection", lambda: self._test_t2v_rejects_input_reference(diffusion_server, case), ) if case.run_consistency_check: run_case_check( "consistency", lambda: self._validate_consistency(case, content), ) if case.run_lora_basic_api_check: run_case_check( "LoRA basic API", lambda: self._test_lora_api_functionality( diffusion_server, case, generate_fn ), ) if case.run_lora_dynamic_switch_check: run_case_check( "LoRA dynamic switch", lambda: self._test_lora_dynamic_switch_e2e( diffusion_server, case, generate_fn, case.server_args.second_lora_path, ), ) if case.run_multi_lora_api_check: run_case_check( "multi-LoRA API", lambda: self._test_multi_lora_e2e( diffusion_server, case, generate_fn, case.server_args.lora_path, case.server_args.second_lora_path, ), ) if failures: formatted_failures = [] for name, message in failures: if "\n" in message: formatted_failures.append(f"[{name}]\n{message}") else: formatted_failures.append(f"[{name}] {message}") pytest.fail( f"Diffusion testcase '{case.id}' failed {len(failures)} check(s):\n\n" + "\n\n".join(formatted_failures), pytrace=False, )