From 2b4eeb8343ccd6ee5676aa3a9b1f73d789a7f46a Mon Sep 17 00:00:00 2001 From: Bingxu Chen Date: Thu, 23 Apr 2026 13:28:20 +0800 Subject: [PATCH] [AMD] Restore test_zimage_turbo.py and test_int4fp8_moe.py with __main__ entry (#23455) Co-authored-by: Claude Opus 4 (1M context) --- test/registered/amd/test_zimage_turbo.py | 156 ++++++++++++++++++++++ test/registered/quant/test_int4fp8_moe.py | 64 +++++++++ 2 files changed, 220 insertions(+) create mode 100644 test/registered/amd/test_zimage_turbo.py create mode 100644 test/registered/quant/test_int4fp8_moe.py diff --git a/test/registered/amd/test_zimage_turbo.py b/test/registered/amd/test_zimage_turbo.py new file mode 100644 index 000000000..2a1481b3c --- /dev/null +++ b/test/registered/amd/test_zimage_turbo.py @@ -0,0 +1,156 @@ +"""AMD nightly test for Z-Image-Turbo diffusion model (text-to-image).""" + +import io +import logging +import os + +import pytest + +from sglang.multimodal_gen.test.server.test_server_common import ( # noqa: F401 + DiffusionServerBase, + diffusion_server, +) +from sglang.multimodal_gen.test.server.test_server_utils import ( + ServerContext, + get_generate_fn, +) +from sglang.multimodal_gen.test.server.testcase_configs import ( + DiffusionSamplingParams, + DiffusionServerArgs, + DiffusionTestCase, +) +from sglang.test.ci.ci_register import register_amd_ci + +logger = logging.getLogger(__name__) + +register_amd_ci(est_time=1800, suite="nightly-amd-1-gpu-zimage-turbo", nightly=True) + +AMD_ZIMAGE_CASES = [ + DiffusionTestCase( + "zimage_image_t2i", + DiffusionServerArgs(model_path="Tongyi-MAI/Z-Image-Turbo", modality="image"), + DiffusionSamplingParams( + prompt="Doraemon is eating dorayaki", + output_size="1024x1024", + ), + ), +] + +CLIP_SCORE_THRESHOLD = 0.20 + + +ARTIFACT_DIR = os.environ.get( + "SGLANG_DIFFUSION_ARTIFACT_DIR", "/tmp/diffusion-artifacts" +) + + +def _save_image_and_write_summary( + case_id: str, prompt: str, image_bytes: bytes, clip_score: float | None = None +): + """Save generated image to artifact dir and write summary.""" + ext = "jpg" if image_bytes[:2] == b"\xff\xd8" else "png" + os.makedirs(ARTIFACT_DIR, exist_ok=True) + img_path = os.path.join(ARTIFACT_DIR, f"{case_id}.{ext}") + with open(img_path, "wb") as f: + f.write(image_bytes) + logger.info("Saved image artifact: %s (%d bytes)", img_path, len(image_bytes)) + + summary_file = os.environ.get("GITHUB_STEP_SUMMARY") + if not summary_file: + return + + clip_line = "" + if clip_score is not None: + status = "PASS" if clip_score >= CLIP_SCORE_THRESHOLD else "FAIL" + clip_line = f"| CLIP Score | {clip_score:.4f} ({status}, threshold: {CLIP_SCORE_THRESHOLD}) |\n" + + md = ( + f"### Z-Image-Turbo — `{case_id}`\n\n" + f"| | |\n|---|---|\n" + f"| Prompt | {prompt} |\n" + f"| Size | {len(image_bytes):,} bytes |\n" + f"{clip_line}" + f"| Artifact | `{case_id}.{ext}` (download from Artifacts section above) |\n\n" + ) + + with open(summary_file, "a") as f: + f.write(md) + + +def _compute_clip_score(image_bytes: bytes, prompt: str) -> float | None: + """Compute CLIP cosine similarity between the image and prompt.""" + try: + import torch + from PIL import Image + from transformers import CLIPModel, CLIPProcessor + + model_name = "openai/clip-vit-base-patch32" + processor = CLIPProcessor.from_pretrained(model_name) + model = CLIPModel.from_pretrained(model_name) + model.eval() + + image = Image.open(io.BytesIO(image_bytes)).convert("RGB") + inputs = processor(text=[prompt], images=image, return_tensors="pt") + + with torch.no_grad(): + outputs = model(**inputs) + score = outputs.logits_per_image.item() / 100.0 + + logger.info("CLIP score for '%s': %.4f", prompt, score) + return score + except Exception as e: + logger.warning("CLIP score computation failed: %s", e) + return None + + +class TestZImageTurboAMD(DiffusionServerBase): + """AMD nightly test for Z-Image-Turbo text-to-image generation.""" + + @classmethod + def teardown_class(cls): + try: + super().teardown_class() + except AttributeError: + pass + + @pytest.fixture(params=AMD_ZIMAGE_CASES, ids=lambda c: c.id) + def case(self, request) -> DiffusionTestCase: + return request.param + + def test_diffusion_generation( + self, + case: DiffusionTestCase, + diffusion_server: ServerContext, + ): + generate_fn = get_generate_fn( + model_path=case.server_args.model_path, + modality=case.server_args.modality, + sampling_params=case.sampling_params, + ) + + perf_record, content = self.run_and_collect( + diffusion_server, case.id, generate_fn + ) + + self._validate_and_record(case, perf_record) + self._test_v1_models_endpoint(diffusion_server, case) + + prompt = case.sampling_params.prompt or "" + clip_score = _compute_clip_score(content, prompt) + + if clip_score is not None: + logger.info( + "CLIP score: %.4f (threshold: %.2f)", clip_score, CLIP_SCORE_THRESHOLD + ) + assert clip_score >= CLIP_SCORE_THRESHOLD, ( + f"CLIP score {clip_score:.4f} below threshold {CLIP_SCORE_THRESHOLD} " + f"for prompt '{prompt}'" + ) + + _save_image_and_write_summary(case.id, prompt, content, clip_score) + + +if __name__ == "__main__": + import sys + + sys.exit(pytest.main([__file__, "-v"])) diff --git a/test/registered/quant/test_int4fp8_moe.py b/test/registered/quant/test_int4fp8_moe.py new file mode 100644 index 000000000..bc92a4ed7 --- /dev/null +++ b/test/registered/quant/test_int4fp8_moe.py @@ -0,0 +1,64 @@ +from types import SimpleNamespace + +from sglang.srt.utils import kill_process_tree +from sglang.test.ci.ci_register import register_amd_ci +from sglang.test.run_eval import run_eval +from sglang.test.test_utils import ( + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + +register_amd_ci(est_time=313, suite="stage-b-test-2-gpu-large-amd") + + +class TestMixtralAccuracy(CustomTestCase): + @classmethod + def setUpClass(cls): + cls.model = "mistralai/Mixtral-8x7B-Instruct-v0.1" + cls.base_url = DEFAULT_URL_FOR_TEST + + other_args = [ + "--tp", + "2", + "--mem-fraction-static", + "0.9", + "--context-length", + "38768", + "--quantization", + "quark_int4fp8_moe", + "--attention-backend", + "triton", + ] + + cls.process = popen_launch_server( + cls.model, + cls.base_url, + timeout=45 * 60, + other_args=other_args, + ) + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + def test_gsm8k(self): + args = SimpleNamespace( + base_url=self.base_url, + model=self.model, + eval_name="gsm8k", + api="completion", + max_tokens=512, + num_examples=1400, + num_threads=128, + num_shots=8, + ) + metrics = run_eval(args) + print(f"{metrics=}") + self.assertGreater(metrics["score"], 0.56) + + +if __name__ == "__main__": + import unittest + + unittest.main()