[diffusion] CI: fix generate mode and add cli test (#16174)

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

python/sglang/multimodal_gen/runtime/entrypoints/cli/generate.py

@@ -91,7 +91,7 @@ def generate_cmd(args: argparse.Namespace):
     server_args = ServerArgs.from_cli_args(args)
     sampling_params_kwargs = SamplingParams.get_cli_args(args)
     generator = DiffGenerator.from_pretrained(
-        model_path=server_args.model_path, server_args=server_args
+        model_path=server_args.model_path, server_args=server_args, local_mode=True
     )
 
     results = generator.generate(sampling_params_kwargs=sampling_params_kwargs)

python/sglang/multimodal_gen/runtime/entrypoints/diffusion_generator.py

@@ -77,6 +77,7 @@ class DiffGenerator:
     @classmethod
     def from_pretrained(
         cls,
+        local_mode: bool = True,
         **kwargs,
     ) -> "DiffGenerator":
         """
@@ -100,10 +101,12 @@ class DiffGenerator:
         else:
             server_args = ServerArgs.from_kwargs(**kwargs)
 
-        return cls.from_server_args(server_args)
+        return cls.from_server_args(server_args, local_mode=local_mode)
 
     @classmethod
-    def from_server_args(cls, server_args: ServerArgs) -> "DiffGenerator":
+    def from_server_args(
+        cls, server_args: ServerArgs, local_mode: bool = True
+    ) -> "DiffGenerator":
         """
         Create a DiffGenerator with the specified arguments.
 
@@ -116,9 +119,8 @@ class DiffGenerator:
         instance = cls(
             server_args=server_args,
         )
-        is_local_mode = server_args.is_local_mode
+        logger.info(f"Local mode: {local_mode}")
-        logger.info(f"Local mode: {is_local_mode}")
+        if local_mode:
-        if is_local_mode:
             instance.local_scheduler_process = instance._start_local_server_if_needed()
         else:
             # In remote mode, we just need to connect and check.

python/sglang/multimodal_gen/runtime/utils/perf_logger.py

@@ -227,9 +227,9 @@ class PerformanceLogger:
             os.makedirs(os.path.dirname(abs_path), exist_ok=True)
             with open(abs_path, "w", encoding="utf-8") as f:
                 json.dump(report, f, indent=2)
-            logger.info(f"[Performance] Metrics dumped to: {abs_path}")
+            logger.info(f"Metrics dumped to: {abs_path}")
         except IOError as e:
-            logger.error(f"[Performance] Failed to dump metrics to {abs_path}: {e}")
+            logger.error(f"Failed to dump metrics to {abs_path}: {e}")
 
     @classmethod
     def log_request_summary(

python/sglang/multimodal_gen/test/cli/test_generate_common.py

@@ -3,105 +3,138 @@
 """
     Common generate cli test, one test for image and video each
 """
-
+import dataclasses
 import os
+import shlex
+import subprocess
+import sys
 import unittest
-from pathlib import Path
+from typing import Optional
 
 from PIL import Image
 
-from sglang.multimodal_gen.test.test_utils import (
+from sglang.multimodal_gen.configs.sample.sampling_params import DataType
-    TestCLIBase,
+from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
-    check_image_size,
+from sglang.multimodal_gen.test.test_utils import check_image_size
-    is_mp4,
+
-    run_command,
+logger = init_logger(__name__)
-)
 
 
-class TestGenerate(TestCLIBase):
+@dataclasses.dataclass
-    model_path = "black-forest-labs/FLUX.1-dev"
+class TestResult:
-    launch_file_name = "launch_flux.json"
+    name: str
-    output_name = "FLUX.1-dev, single gpu"
+    key: str
-    ext = "jpg"
+    duration: Optional[float]
+    succeed: bool
 
-    def test_generate_with_config(self):
+    @property
-        test_dir = Path(__file__).parent
+    def duration_str(self):
-        config_path = (
+        return f"{self.duration:.4f}" if self.duration else "NA"
-            (test_dir / ".." / "test_files" / self.launch_file_name)
+
-            .resolve()
+
-            .as_posix()
+def run_command(command) -> Optional[float]:
+    """Runs a command and returns the execution time and status."""
+    print(f"Running command: {shlex.join(command)}")
+
+    duration = None
+    with subprocess.Popen(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        encoding="utf-8",
+    ) as process:
+        for line in process.stdout:
+            sys.stdout.write(line)
+            if "Pixel data generated" in line:
+                words = line.split(" ")
+                duration = float(words[-2])
+
+    if process.returncode == 0:
+        return duration
+    else:
+        print(f"Command failed with exit code {process.returncode}")
+        return None
+
+
+class CLIBase(unittest.TestCase):
+    model_path: str = None
+    extra_args = []
+    data_type: DataType = None
+    # tested on h100
+
+    width: int = 720
+    height: int = 720
+    output_path: str = "test_outputs"
+
+    def get_base_command(self):
+        return [
+            "sglang",
+            "generate",
+            "--prompt",
+            "A curious raccoon",
+            "--save-output",
+            "--log-level=debug",
+            f"--width={self.width}",
+            f"--height={self.height}",
+            f"--output-path={self.output_path}",
+        ]
+
+    results = []
+
+    @classmethod
+    def setUpClass(cls):
+        cls.results = []
+
+    def _run_command(self, name: str, model_path: str, test_key: str = "", args=[]):
+        command = (
+            self.get_base_command()
+            + [f"--model-path={model_path}"]
+            + shlex.split(args or "")
+            + ["--output-file-name", f"{name}"]
+            + self.extra_args
         )
-        command = [
-            "sgl_diffusion",
-            "generate",
-            f"--config={config_path}",
-        ]
         duration = run_command(command)
+        status = "Success" if duration else "Failed"
+        succeed = duration is not None
 
-        self.assertIsNotNone(duration, f"Run command failed: {command}")
+        duration = float(duration) if succeed else None
+        self.results.append(TestResult(name, test_key, duration, succeed))
 
-        # verify
+        return name, duration, status
-        self.verify_image(self.output_name)
 
-    def test_generate_multiple_outputs(self):
+    def _run_test(self, name: str, args, model_path: str, test_key: str):
-        command = [
+        name, duration, status = self._run_command(
-            "sglang",
+            name, args=args, model_path=model_path, test_key=test_key
-            "generate",
+        )
-            "--prompt",
+        self.verify(status, name, duration)
-            "A curious raccoon",
-            "--output-path=outputs",
-            f"--model-path={self.model_path}",
-            "--save-output",
-            f"--output-file-name={self.output_name}",
-            "--num-outputs-per-prompt=2",
-            "--width=720",
-            "--height=720",
-        ]
-        duration = run_command(command)
-        self.assertIsNotNone(duration, f"Run command failed: {command}")
 
-        self.verify_image(f"{self.output_name}_0.{self.ext}")
+    def verify(self, status, name, duration):
-        self.verify_image(f"{self.output_name}_1.{self.ext}")
+        print("-" * 80)
+        print("\n" * 3)
 
-    def verify_image(self, output_name):
+        # test task status
-        path = os.path.join("outputs", output_name)
+        self.assertEqual(status, "Success", f"{name} command failed")
+        self.assertIsNotNone(duration, f"Could not parse duration for {name}")
+
+        # test output file
+        path = os.path.join(
+            self.output_path, f"{name}.{self.data_type.get_default_extension()}"
+        )
+        self.assertTrue(os.path.exists(path), f"Output file not exist for {path}")
+        if self.data_type == DataType.IMAGE:
             with Image.open(path) as image:
-            check_image_size(self, image, 720, 720)
+                check_image_size(self, image, self.width, self.height)
+        logger.info(f"{name} passed in {duration:.4f}s")
 
-    def verify_video(self, output_name):
+    def model_name(self):
-        path = os.path.join("outputs", output_name)
+        return self.model_path.split("/")[-1]
-        with open(path, "rb") as f:
-            header = f.read(12)
-            assert is_mp4(header)
 
-
+    def test_single_gpu(self):
-class TestWanGenerate(TestGenerate):
+        """single gpu"""
-    model_path = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"
+        self._run_test(
-    launch_file_name = "launch_wan.json"
+            name=f"{self.model_name()}_single_gpu",
-    output_name = "Wan2.1-T2V-1.3B-Diffusers, single gpu"
+            args=None,
-    ext = "mp4"
+            model_path=self.model_path,
-
+            test_key="test_single_gpu",
-    def test_generate_multiple_outputs(self):
+        )
-        command = [
-            "sglang",
-            "generate",
-            "--prompt",
-            "A curious raccoon",
-            "--output-path=outputs",
-            f"--model-path={self.model_path}",
-            "--save-output",
-            f"--output-file-name={self.output_name}",
-            "--num-outputs-per-prompt=2",
-            "--width=720",
-            "--height=720",
-        ]
-        duration = run_command(command)
-        self.assertIsNotNone(duration, f"Run command failed: {command}")
-
-        self.verify_video(f"{self.output_name}_0.{self.ext}")
-        # FIXME: second video is a meaningless output
-        self.verify_video(f"{self.output_name}_1.{self.ext}")
-
-
-if __name__ == "__main__":
-    unittest.main()

python/sglang/multimodal_gen/test/cli/test_generate_t2i_perf.py

@@ -1,132 +1,22 @@
 # Copied and adapted from: https://github.com/hao-ai-lab/FastVideo
 
 import unittest
-from pathlib import Path
 
 from sglang.multimodal_gen.configs.sample.sampling_params import DataType
 from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
-from sglang.multimodal_gen.test.test_utils import TestGenerateBase
+from sglang.multimodal_gen.test.cli.test_generate_common import CLIBase
 
 logger = init_logger(__name__)
 
 
-class TestFlux_T2V(TestGenerateBase):
+class TestFlux_T2V(CLIBase):
     model_path = "black-forest-labs/FLUX.1-dev"
     extra_args = []
     data_type: DataType = DataType.IMAGE
-    thresholds = {
-        "test_single_gpu": 6.5 * 1.05,
-        "test_usp": 8.3 * 1.05,
-    }
-
-    def test_cfg_parallel(self):
-        pass
-
-    def test_mixed(self):
-        pass
 
 
-class TestQwenImage(TestGenerateBase):
+del CLIBase
-    model_path = "Qwen/Qwen-Image"
-    extra_args = []
-    data_type: DataType = DataType.IMAGE
-    thresholds = {
-        "test_single_gpu": 10.4 * 1.05,
-        "test_usp": 20.2 * 1.05,
-    }
-
-    def test_cfg_parallel(self):
-        pass
-
-    def test_mixed(self):
-        pass
-
-
-class TestQwenImageEdit(TestGenerateBase):
-    model_path = "Qwen/Qwen-Image-Edit"
-    extra_args = []
-    data_type: DataType = DataType.IMAGE
-    thresholds = {
-        "test_single_gpu": 33.4 * 1.05,
-        "test_usp": 26.9 * 1.05,
-    }
-
-    prompt: str | None = (
-        "Change the rabbit's color to purple, with a flash light background."
-    )
-
-    def setUp(self):
-        test_dir = Path(__file__).parent
-        img_path = (test_dir / ".." / "test_files" / "rabbit.jpg").resolve().as_posix()
-        self.base_command = [
-            "sglang",
-            "generate",
-            "--text-encoder-cpu-offload",
-            "--pin-cpu-memory",
-            f"--prompt",
-            f"{self.prompt}",
-            "--save-output",
-            "--log-level=debug",
-            f"--width={self.width}",
-            f"--height={self.height}",
-            f"--output-path={self.output_path}",
-        ] + [f"--image-path={img_path}"]
-
-    def test_cfg_parallel(self):
-        pass
-
-    def test_mixed(self):
-        pass
-
-
-class TestQwenImageEditPlusMultiImageURL(TestGenerateBase):
-    """CLI-level test for multi-image URL input with Qwen-Image-Edit."""
-
-    model_path = "Qwen/Qwen-Image-Edit-2509"
-    extra_args = []
-    data_type: DataType = DataType.IMAGE
-
-    thresholds = {
-        "test_single_gpu": 33.4 * 1.05,
-    }
-
-    prompt: str | None = (
-        "The magician bear is on the left, the alchemist bear is on the right, facing each other in the central park square."
-    )
-
-    def setUp(self):
-        super().setUp()
-
-        img_urls = [
-            "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen-Image/edit2509/edit2509_1.jpg",
-            "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen-Image/edit2509/edit2509_2.jpg",
-        ]
-
-        self.base_command = [
-            "sglang",
-            "generate",
-            "--text-encoder-cpu-offload",
-            "--pin-cpu-memory",
-            "--prompt",
-            f"{self.prompt}",
-            "--save-output",
-            "--log-level=debug",
-            f"--width={self.width}",
-            f"--height={self.height}",
-            f"--output-path={self.output_path}",
-        ]
-        self.base_command += [
-            "--image-path",
-            *img_urls,
-        ]
-
-    def test_cfg_parallel(self):
-        pass
-
-    def test_mixed(self):
-        pass
 
 
 if __name__ == "__main__":
-    del TestGenerateBase
     unittest.main()

python/sglang/multimodal_gen/test/cli/test_generate_t2v_perf.py

@@ -1,81 +0,0 @@
-# Copied and adapted from: https://github.com/hao-ai-lab/FastVideo
-
-import unittest
-
-from sglang.multimodal_gen.configs.sample.sampling_params import DataType
-from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
-from sglang.multimodal_gen.test.test_utils import TestGenerateBase
-
-logger = init_logger(__name__)
-
-
-class TestFastWan2_1_T2V(TestGenerateBase):
-    model_path = "FastVideo/FastWan2.1-T2V-1.3B-Diffusers"
-    extra_args = ["--attention-backend=video_sparse_attn"]
-    data_type: DataType = DataType.VIDEO
-    thresholds = {
-        "test_single_gpu": 13.0,
-        "test_cfg_parallel": 15.0,
-        "test_usp": 15.0,
-        "test_mixed": 15.0 * 1.05,
-    }
-
-    # disabled for vsa
-    def test_usp(self):
-        pass
-
-
-class TestFastWan2_2_T2V(TestGenerateBase):
-    model_path = "FastVideo/FastWan2.2-TI2V-5B-FullAttn-Diffusers"
-    extra_args = []
-    data_type: DataType = DataType.VIDEO
-    thresholds = {
-        "test_single_gpu": 25.0,
-        "test_cfg_parallel": 30.0,
-        "test_usp": 30.0,
-        "test_mixed": 30.0,
-    }
-
-
-class TestWan2_1_T2V(TestGenerateBase):
-    model_path = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"
-    extra_args = []
-    data_type: DataType = DataType.VIDEO
-    thresholds = {
-        "test_single_gpu": 76.0 * 1.05,
-        "test_cfg_parallel": 46.5 * 1.05,
-        "test_usp": 39.8 * 1.05,
-        "test_mixed": 37.3 * 1.05,
-    }
-
-    def test_mixed(self):
-        pass
-
-    def test_cfg_parallel(self):
-        pass
-
-
-class TestWan2_2_T2V(TestGenerateBase):
-    model_path = "Wan-AI/Wan2.2-T2V-A14B-Diffusers"
-    extra_args = []
-    data_type: DataType = DataType.VIDEO
-    thresholds = {
-        "test_single_gpu": 904.3 * 1.05,
-        "test_cfg_parallel": 446,
-        "test_usp": 316 * 1.05,
-        "test_mixed": 159,
-    }
-
-    def test_single_gpu(self):
-        pass
-
-    def test_mixed(self):
-        pass
-
-    def test_cfg_parallel(self):
-        pass
-
-
-if __name__ == "__main__":
-    del TestGenerateBase
-    unittest.main()

python/sglang/multimodal_gen/test/cli/test_generate_ti2v_perf.py

@@ -1,71 +0,0 @@
-import unittest
-
-from sglang.multimodal_gen.configs.sample.sampling_params import DataType
-from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
-from sglang.multimodal_gen.test.test_utils import TestGenerateBase
-
-logger = init_logger(__name__)
-
-
-class TestGenerateTI2VBase(TestGenerateBase):
-    data_type: DataType = DataType.VIDEO
-
-    @classmethod
-    def setUpClass(cls):
-        cls.base_command = [
-            "sglang",
-            "generate",
-            "--prompt",
-            "Summer beach vacation style, a white cat wearing sunglasses sits on a surfboard. The fluffy-furred feline gazes directly at the camera with a relaxed expression. Blurred beach scenery forms the background featuring crystal-clear waters, distant green hills, and a blue sky dotted with white clouds. The cat assumes a naturally relaxed posture, as if savoring the sea breeze and warm sunlight. A close-up shot highlights the feline's intricate details and the refreshing atmosphere of the seaside.",
-            "--image-path",
-            "https://github.com/Wan-Video/Wan2.2/blob/990af50de458c19590c245151197326e208d7191/examples/i2v_input.JPG?raw=true",
-            "--save-output",
-            "--log-level=debug",
-            f"--output-path={cls.output_path}",
-        ] + cls.extra_args
-
-    def test_single_gpu(self):
-        pass
-
-    def test_cfg_parallel(self):
-        pass
-
-    def test_mixed(self):
-        pass
-
-
-class TestWan2_1_I2V_14B_480P(TestGenerateTI2VBase):
-    model_path = "Wan-AI/Wan2.1-I2V-14B-480P-Diffusers"
-    thresholds = {
-        "test_usp": 557.9 * 1.05,
-    }
-
-
-class TestWan2_1_I2V_14B_720P(TestGenerateTI2VBase):
-    model_path = "Wan-AI/Wan2.1-I2V-14B-720P-Diffusers"
-    thresholds = {
-        "test_usp": 558.4 * 1.05,
-    }
-
-
-class TestWan2_2_TI2V_5B(TestGenerateTI2VBase):
-    model_path = "Wan-AI/Wan2.2-TI2V-5B-Diffusers"
-    # FIXME: doesn't work with vsa at the moment
-    # extra_args = ["--attention-backend=video_sparse_attn"]
-    thresholds = {
-        "test_usp": 82.3 * 1.05,
-    }
-
-
-# OOM
-# class TestWan2_2_I2V_A14B(TestGenerateTI2VBase):
-#     model_path = "Wan-AI/Wan2.2-I2V-A14B-Diffusers"
-#     # FIXME: doesn't work with vsa at the moment
-#     thresholds = {
-#         "test_usp": 66.3 * 1.05,
-#     }
-
-
-if __name__ == "__main__":
-    del TestGenerateTI2VBase, TestGenerateBase
-    unittest.main()

python/sglang/multimodal_gen/test/cli/test_serve.py

@@ -1,301 +0,0 @@
-# Copied and adapted from: https://github.com/hao-ai-lab/FastVideo
-
-import asyncio
-import base64
-import subprocess
-import tempfile
-import time
-import unittest
-import uuid
-from contextlib import contextmanager
-from pathlib import Path
-from urllib.request import urlopen
-
-from openai import OpenAI
-
-from sglang.multimodal_gen.runtime.utils.common import kill_process_tree
-from sglang.multimodal_gen.test.test_utils import is_mp4, is_png, wait_for_port
-
-
-@contextmanager
-def downloaded_temp_file(url: str, prefix: str = "i2v_input_", suffix: str = ".jpg"):
-    tmp_path = Path(tempfile.gettempdir()) / f"{prefix}{uuid.uuid4().hex}{suffix}"
-    with urlopen(url) as resp:
-        tmp_path.write_bytes(resp.read())
-    try:
-        yield tmp_path
-    finally:
-        try:
-            tmp_path.unlink(missing_ok=True)
-        except Exception:
-            pass
-
-
-def wait_for_video_completion(client, video_id, timeout=300, check_interval=3):
-    start = time.time()
-    video = client.videos.retrieve(video_id)
-
-    while video.status not in ("completed", "failed"):
-        time.sleep(check_interval)
-        video = client.videos.retrieve(video_id)
-        assert time.time() - start < timeout, "video generate timeout"
-
-    return video
-
-
-class TestVideoHttpServer(unittest.TestCase):
-    model_name = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"
-    timeout = 500
-    extra_args = []
-
-    def _create_wait_and_download(
-        self, client: OpenAI, prompt: str, size: str
-    ) -> bytes:
-
-        video = client.videos.create(prompt=prompt, size=size)
-        video_id = video.id
-        self.assertEqual(video.status, "queued")
-
-        video = wait_for_video_completion(client, video_id, timeout=self.timeout)
-        self.assertEqual(video.status, "completed", "video generate failed")
-
-        response = client.videos.download_content(
-            video_id=video_id,
-        )
-        content = response.read()
-        return content
-
-    @classmethod
-    def setUpClass(cls):
-        cls.base_command = [
-            "sglang",
-            "serve",
-            "--model-path",
-            f"{cls.model_name}",
-            "--port",
-            "30010",
-        ]
-
-        process = subprocess.Popen(
-            cls.base_command + cls.extra_args,
-            # stdout=subprocess.PIPE,
-            # stderr=subprocess.PIPE,
-            text=True,
-            bufsize=1,
-        )
-        cls.pid = process.pid
-        wait_for_port(host="127.0.0.1", port=30010)
-
-    @classmethod
-    def tearDownClass(cls):
-        kill_process_tree(cls.pid)
-
-    def test_http_server_basic(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30010/v1"
-        )
-        content = self._create_wait_and_download(
-            client, "A plane is taking off.", "832x480"
-        )
-        self.assertTrue(is_mp4(content))
-
-    def test_concurrent_requests(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30010/v1"
-        )
-
-        num_requests = 2
-
-        async def generate_and_check_video(prompt, size):
-            content = await asyncio.to_thread(
-                self._create_wait_and_download, client, prompt, size
-            )
-            self.assertTrue(is_mp4(content))
-
-        async def send_concurrent_requests():
-            tasks = [
-                generate_and_check_video(
-                    "A ship is beside the port.",
-                    "832x480",
-                )
-                for _ in range(num_requests)
-            ]
-            await asyncio.gather(*tasks)
-
-        asyncio.run(send_concurrent_requests())
-
-
-class TestImage2VideoHttpServer(unittest.TestCase):
-    model_name = "Wan-AI/Wan2.2-I2V-A14B-Diffusers"
-    timeout = 1200
-    extra_args = []
-
-    def _create_wait_and_download(
-        self, client: OpenAI, prompt: str, size: str
-    ) -> bytes:
-
-        image_url = "https://github.com/Wan-Video/Wan2.2/blob/990af50de458c19590c245151197326e208d7191/examples/i2v_input.JPG?raw=true"
-        with downloaded_temp_file(
-            image_url, prefix="i2v_input_", suffix=".jpg"
-        ) as tmp_path:
-            video = client.videos.create(
-                prompt=prompt,
-                input_reference=tmp_path,
-                size=size,
-                seconds=10,
-                extra_body={"fps": 16, "num_frames": 125},
-            )
-        # TODO: Some combinations of num_frames and fps may cause errors and need further investigation.
-        video_id = video.id
-        self.assertEqual(video.status, "queued")
-
-        video = wait_for_video_completion(client, video_id, timeout=self.timeout)
-        self.assertEqual(video.status, "completed", "video generate failed")
-
-        response = client.videos.download_content(
-            video_id=video_id,
-        )
-        content = response.read()
-        return content
-
-    @classmethod
-    def setUpClass(cls):
-        cls.base_command = [
-            "sglang",
-            "serve",
-            "--model-path",
-            f"{cls.model_name}",
-            "--num-gpus",
-            "4",
-            "--ulysses-degree",
-            "4",
-            "--port",
-            "30010",
-        ]
-
-        process = subprocess.Popen(
-            cls.base_command + cls.extra_args,
-            text=True,
-            bufsize=1,
-        )
-        cls.pid = process.pid
-        wait_for_port(host="127.0.0.1", port=30010)
-
-    @classmethod
-    def tearDownClass(cls):
-        kill_process_tree(cls.pid)
-
-    def test_http_server_basic(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30010/v1"
-        )
-        content = self._create_wait_and_download(
-            client, "A cat surfing on the sea.", "832x480"
-        )
-        self.assertTrue(is_mp4(content))
-
-    def test_concurrent_requests(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30010/v1"
-        )
-
-        num_requests = 2
-
-        async def generate_and_check_video(prompt, size):
-            content = await asyncio.to_thread(
-                self._create_wait_and_download, client, prompt, size
-            )
-            self.assertTrue(is_mp4(content))
-
-        async def send_concurrent_requests():
-            tasks = [
-                generate_and_check_video(
-                    "A cat surfing on the sea.",
-                    "832x480",
-                )
-                for _ in range(num_requests)
-            ]
-            await asyncio.gather(*tasks)
-
-        asyncio.run(send_concurrent_requests())
-
-
-class TestImageHttpServer(unittest.TestCase):
-
-    def _create_wait_and_download(
-        self, client: OpenAI, prompt: str, size: str
-    ) -> bytes:
-        img = client.images.generate(
-            model="gpt-image-1",
-            prompt=prompt,
-            n=1,
-            size=size,
-            response_format="b64_json",
-            output_format="png",
-        )
-        image_bytes = base64.b64decode(img.data[0].b64_json)
-        return image_bytes
-
-    @classmethod
-    def setUpClass(cls):
-        cls.base_command = [
-            "sglang",
-            "serve",
-            "--model-path",
-            "Qwen/Qwen-Image",
-            "--port",
-            "30020",
-        ]
-
-        process = subprocess.Popen(
-            cls.base_command,
-            # stdout=subprocess.PIPE,
-            # stderr=subprocess.PIPE,
-            text=True,
-            bufsize=1,
-        )
-        cls.pid = process.pid
-        wait_for_port(host="127.0.0.1", port=30020)
-
-    @classmethod
-    def tearDownClass(cls):
-        kill_process_tree(cls.pid)
-
-    def test_http_server_basic(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30020/v1"
-        )
-        content = self._create_wait_and_download(
-            client, "A calico cat playing a piano on stage", "832x480"
-        )
-        self.assertTrue(is_png(content))
-
-    def test_concurrent_requests(self):
-        client = OpenAI(
-            api_key="sk-proj-1234567890", base_url="http://localhost:30020/v1"
-        )
-
-        num_requests = 2
-
-        async def generate_and_check_image(prompt, size):
-            content = await asyncio.to_thread(
-                self._create_wait_and_download, client, prompt, size
-            )
-            self.assertTrue(is_png(content))
-
-        async def send_concurrent_requests():
-            tasks = [
-                generate_and_check_image(
-                    "A dog playing a piano on stage",
-                    "832x480",
-                )
-                for _ in range(num_requests)
-            ]
-            await asyncio.gather(*tasks)
-
-        asyncio.run(send_concurrent_requests())
-
-
-if __name__ == "__main__":
-    # del TestPerform·anceBase
-    unittest.main()

python/sglang/multimodal_gen/test/run_suite.py

@@ -23,6 +23,8 @@ SUITES = {
         "test_server_a.py",
         "test_server_b.py",
         "test_lora_format_adapter.py",
+        # cli test
+        "../cli/test_generate_t2i_perf.py",
         # add new 1-gpu test files here
     ],
     "2-gpu": [

python/sglang/multimodal_gen/test/test_offline_api.py

@@ -1,75 +0,0 @@
-# Copied and adapted from: https://github.com/hao-ai-lab/FastVideo
-
-"""
-    Testing the performance of generate command of sgl_diffusion' CLI
-"""
-
-import unittest
-
-import torch
-
-from sglang.multimodal_gen.runtime.entrypoints.diffusion_generator import DiffGenerator
-from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
-
-logger = init_logger(__name__)
-
-
-class TestGeneratorAPIBase(unittest.TestCase):
-    # server args
-    server_kwargs = {}
-
-    # sampling
-    output_path: str = "test_outputs"
-
-    results = []
-
-    @classmethod
-    def setUpClass(cls):
-        cls.results = []
-
-    def verify_single_generation_result(self, result):
-        self.assertIsNotNone(result, "Generation failed")
-        self.assertTrue(
-            "samples" in result and isinstance(result["samples"], torch.Tensor),
-            f"Incorrect Generation result",
-        )
-
-    def _run_test(self, name, server_kwargs, test_key: str):
-        generator = DiffGenerator.from_pretrained(**server_kwargs)
-        result = generator.generate(prompt="A curious raccoon")
-        self.verify_single_generation_result(result)
-
-    def test_single_gpu(self):
-        self._run_test(
-            name=self.server_kwargs["model_path"],
-            server_kwargs=self.server_kwargs | dict(num_gpus=1),
-            test_key="test_single_gpu",
-        )
-
-    def test_cfg_parallel(self):
-        self._run_test(
-            name=self.server_kwargs["model_path"],
-            server_kwargs=self.server_kwargs
-            | dict(num_gpus=2, enable_cfg_parallel=True),
-            test_key="test_cfg_parallel",
-        )
-
-    def test_multiple_prompts(self):
-        generator = DiffGenerator.from_pretrained(
-            **self.server_kwargs | dict(num_gpus=2, enable_cfg_parallel=True)
-        )
-        prompts = ["A curious raccoon", "A curious cat"]
-        results = generator.generate(prompt=prompts)
-
-        self.assertEqual(len(results), len(prompts), "Some generation tasks fail")
-        for result in results:
-            self.verify_single_generation_result(result)
-
-
-class TestWan2_1_T2V(TestGeneratorAPIBase):
-    server_kwargs = {"model_path": "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"}
-
-
-if __name__ == "__main__":
-    del TestGeneratorAPIBase
-    unittest.main()

python/sglang/multimodal_gen/test/test_utils.py

@@ -1,21 +1,14 @@
 # Copied and adapted from: https://github.com/hao-ai-lab/FastVideo
 import base64
-import dataclasses
 import json
 import os
-import shlex
 import socket
-import subprocess
-import sys
 import time
-import unittest
 from pathlib import Path
-from typing import Optional
 
 import cv2
 from PIL import Image
 
-from sglang.multimodal_gen.configs.sample.sampling_params import DataType
 from sglang.multimodal_gen.runtime.utils.common import get_bool_env_var
 from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
 from sglang.multimodal_gen.runtime.utils.perf_logger import (
@@ -35,31 +28,6 @@ def is_image_url(image_path: str | Path | None) -> bool:
     )
 
 
-def run_command(command) -> Optional[float]:
-    """Runs a command and returns the execution time and status."""
-    print(f"Running command: {shlex.join(command)}")
-
-    duration = None
-    with subprocess.Popen(
-        command,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        text=True,
-        encoding="utf-8",
-    ) as process:
-        for line in process.stdout:
-            sys.stdout.write(line)
-            if "Pixel data generated" in line:
-                words = line.split(" ")
-                duration = float(words[-2])
-
-    if process.returncode == 0:
-        return duration
-    else:
-        print(f"Command failed with exit code {process.returncode}")
-        return None
-
-
 def probe_port(host="127.0.0.1", port=30010, timeout=2.0) -> bool:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         s.settimeout(timeout)
@@ -397,199 +365,3 @@ def validate_video_file(
         assert (
             actual_height == expected_height
         ), f"Video height mismatch: expected {expected_height}, got {actual_height}"
-
-
-@dataclasses.dataclass
-class TestResult:
-    name: str
-    key: str
-    duration: Optional[float]
-    succeed: bool
-
-    @property
-    def duration_str(self):
-        return f"{self.duration:.4f}" if self.duration else "NA"
-
-
-class TestCLIBase(unittest.TestCase):
-    model_path: str = None
-    extra_args = []
-    data_type: DataType = None
-    # tested on h100
-    thresholds = {}
-
-    width: int = 720
-    height: int = 720
-    output_path: str = "test_outputs"
-
-    base_command = [
-        "sglang",
-        "generate",
-        "--text-encoder-cpu-offload",
-        "--pin-cpu-memory",
-        "--prompt",
-        "A curious raccoon",
-        "--save-output",
-        "--log-level=debug",
-        f"--width={width}",
-        f"--height={height}",
-        f"--output-path={output_path}",
-    ]
-
-    results = []
-
-    @classmethod
-    def setUpClass(cls):
-        cls.results = []
-
-    def _run_command(self, name: str, model_path: str, test_key: str = "", args=[]):
-        command = (
-            self.base_command
-            + [f"--model-path={model_path}"]
-            + shlex.split(args or "")
-            + ["--output-file-name", f"{name}"]
-            + self.extra_args
-        )
-        duration = run_command(command)
-        status = "Success" if duration else "Failed"
-        succeed = duration is not None
-
-        duration = float(duration) if succeed else None
-        self.results.append(TestResult(name, test_key, duration, succeed))
-
-        return name, duration, status
-
-
-class TestGenerateBase(TestCLIBase):
-    model_path: str = None
-    extra_args = []
-    data_type: DataType = None
-    # tested on h100
-    thresholds = {}
-
-    width: int = 720
-    height: int = 720
-    output_path: str = "test_outputs"
-    image_path: str | None = None
-    prompt: str | None = "A curious raccoon"
-
-    base_command = [
-        "sglang",
-        "generate",
-        # "--text-encoder-cpu-offload",
-        # "--pin-cpu-memory",
-        f"--prompt",
-        f"{prompt}",
-        "--save-output",
-        "--log-level=debug",
-        f"--width={width}",
-        f"--height={height}",
-        f"--output-path={output_path}",
-    ]
-
-    results: list[TestResult] = []
-
-    @classmethod
-    def setUpClass(cls):
-        cls.results = []
-
-    @classmethod
-    def tearDownClass(cls):
-        # Print markdown table
-        print("\n## Test Results\n")
-        print("| Test Case                      | Duration | Status  |")
-        print("|--------------------------------|----------|---------|")
-        test_keys = ["test_single_gpu", "test_cfg_parallel", "test_usp", "test_mixed"]
-        test_key_to_order = {
-            test_key: order for order, test_key in enumerate(test_keys)
-        }
-
-        ordered_results: list[TestResult] = [None] * len(test_keys)
-        for result in cls.results:
-            order = test_key_to_order[result.key]
-            ordered_results[order] = result
-
-        for result in ordered_results:
-            if not result:
-                continue
-            status = (
-                "Succeed"
-                if (
-                    result.succeed
-                    and float(result.duration) <= float(cls.thresholds[result.key])
-                )
-                else "Failed"
-            )
-            print(f"| {result.name:<30} | {result.duration_str:<8} | {status:<7} |")
-        print()
-        durations = [result.duration_str for result in cls.results]
-        print(" | ".join([""] + durations + [""]))
-
-    def _run_test(self, name: str, args, model_path: str, test_key: str):
-        time_threshold = self.thresholds[test_key]
-        name, duration, status = self._run_command(
-            name, args=args, model_path=model_path, test_key=test_key
-        )
-        self.verify(status, name, duration, time_threshold)
-
-    def verify(self, status, name, duration, time_threshold):
-        print("-" * 80)
-        print("\n" * 3)
-
-        # test task status
-        self.assertEqual(status, "Success", f"{name} command failed")
-        self.assertIsNotNone(duration, f"Could not parse duration for {name}")
-        self.assertLessEqual(
-            duration,
-            time_threshold,
-            f"{name} failed with {duration:.4f}s > {time_threshold}s",
-        )
-
-        # test output file
-        path = os.path.join(
-            self.output_path, f"{name}.{self.data_type.get_default_extension()}"
-        )
-        self.assertTrue(os.path.exists(path), f"Output file not exist for {path}")
-        if self.data_type == DataType.IMAGE:
-            with Image.open(path) as image:
-                check_image_size(self, image, self.width, self.height)
-        logger.info(f"{name} passed in {duration:.4f}s (threshold: {time_threshold}s)")
-
-    def model_name(self):
-        return self.model_path.split("/")[-1]
-
-    def test_single_gpu(self):
-        """single gpu"""
-        self._run_test(
-            name=f"{self.model_name()}_single_gpu",
-            args=None,
-            model_path=self.model_path,
-            test_key="test_single_gpu",
-        )
-
-    def test_cfg_parallel(self):
-        """cfg parallel"""
-        self._run_test(
-            name=f"{self.model_name()}_cfg_parallel",
-            args="--num-gpus 2 --enable-cfg-parallel",
-            model_path=self.model_path,
-            test_key="test_cfg_parallel",
-        )
-
-    def test_usp(self):
-        """usp"""
-        self._run_test(
-            name=f"{self.model_name()}_usp",
-            args="--num-gpus 4 --ulysses-degree=2 --ring-degree=2",
-            model_path=self.model_path,
-            test_key="test_usp",
-        )
-
-    def test_mixed(self):
-        """mixed"""
-        self._run_test(
-            name=f"{self.model_name()}_mixed",
-            args="--num-gpus 4 --ulysses-degree=2 --ring-degree=1 --enable-cfg-parallel",
-            model_path=self.model_path,
-            test_key="test_mixed",
-        )