[NPU] Add GitHub test summary and deduplicate test code. Part 1 (#23835)

Co-authored-by: Elizaveta Martirosian <elizaveta.martirosian@gmail.com>
Co-authored-by: root <root@localhost.localdomain>
Co-authored-by: Elizaveta Martirosian <you@example.com>
Co-authored-by: ronnie_zheng <zl19940307@163.com>

python/sglang/test/ascend/gsm8k_ascend_mixin.py

@@ -1,19 +1,22 @@
 import os
+import subprocess
 from abc import ABC
 from types import SimpleNamespace
 
 from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.test_ascend_utils import write_results_to_github_step_summary
 from sglang.test.few_shot_gsm8k import run_eval
 from sglang.test.test_utils import (
     DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
     DEFAULT_URL_FOR_TEST,
     popen_launch_server,
+    write_github_step_summary,
 )
 
 
 class GSM8KAscendMixin(ABC):
     model = ""
-    accuracy = 0.00
+
     timeout_for_server_launch = DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
     other_args = [
         "--trust-remote-code",
@@ -23,48 +26,80 @@ class GSM8KAscendMixin(ABC):
         "ascend",
         "--disable-cuda-graph",
     ]
+    server_cmd = ""
     gsm8k_num_shots = 5
+    num_questions = 200
+
+    env = {
+        **os.environ,
+        "PYTORCH_NPU_ALLOC_CONF": "expandable_segments:True",
+        "ASCEND_MF_STORE_URL": "tcp://127.0.0.1:24666",
+        "HCCL_BUFFSIZE": "200",
+        "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": "24",
+        "USE_VLLM_CUSTOM_ALLREDUCE": "1",
+        "HCCL_EXEC_TIMEOUT": "200",
+        "STREAMS_PER_DEVICE": "32",
+        "SGLANG_ENBLE_TORCH_COMILE": "1",
+        "AUTO_USE_UC_MEMORY": "0",
+        "P2P_HCCL_BUFFSIZE": "20",
+    }
 
     @classmethod
     def setUpClass(cls):
         cls.base_url = DEFAULT_URL_FOR_TEST
-        os.environ["PYTORCH_NPU_ALLOC_CONF"] = "expandable_segments:True"
+        try:
-        os.environ["ASCEND_MF_STORE_URL"] = "tcp://127.0.0.1:24666"
-        os.environ["HCCL_BUFFSIZE"] = "200"
-        os.environ["SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK"] = "24"
-        os.environ["USE_VLLM_CUSTOM_ALLREDUCE"] = "1"
-        os.environ["HCCL_EXEC_TIMEOUT"] = "200"
-        os.environ["STREAMS_PER_DEVICE"] = "32"
-        os.environ["SGLANG_ENBLE_TORCH_COMILE"] = "1"
-        os.environ["AUTO_USE_UC_MEMORY"] = "0"
-        os.environ["P2P_HCCL_BUFFSIZE"] = "20"
-        env = os.environ.copy()
-
             cls.process = popen_launch_server(
                 cls.model,
                 cls.base_url,
                 timeout=cls.timeout_for_server_launch,
                 other_args=cls.other_args,
-            env=env,
+                env=cls.env,
             )
+            cls.server_cmd = subprocess.list2cmdline(cls.process.args)
+        except Exception as e:
+            write_github_step_summary(f"Failed to launch server for {cls.model}: {e}")
+            raise AssertionError(f"Test failed for {cls.model}: {e}")
 
     @classmethod
     def tearDownClass(cls):
         kill_process_tree(cls.process.pid)
 
     def test_gsm8k(self):
+        accuracy_threshold = getattr(self, "accuracy", 0.00)
+        output_throughput_threshold = getattr(self, "output_throughput", 0.00)
+
+        model_metrics = {
+            "server": self.server_cmd,
+            "client": "few_shot_gsm8k",
+            "accuracy_threshold": getattr(self, "accuracy", "N/A"),
+            "output_throughput_threshold": getattr(self, "output_throughput", "N/A"),
+        }
+
+        try:
             args = SimpleNamespace(
                 num_shots=self.gsm8k_num_shots,
                 data_path=None,
-            num_questions=200,
+                num_questions=self.num_questions,
                 max_new_tokens=512,
                 parallel=128,
                 host="http://127.0.0.1",
                 port=int(self.base_url.split(":")[-1]),
             )
             metrics = run_eval(args)
+            model_metrics["accuracy"] = metrics["accuracy"]
+            model_metrics["output_throughput"] = metrics["output_throughput"]
             self.assertGreaterEqual(
                 metrics["accuracy"],
-            self.accuracy,
+                accuracy_threshold,
-            f'Accuracy of {self.model} is {str(metrics["accuracy"])}, is lower than {self.accuracy}',
+                f'Accuracy of {self.model} is {str(metrics["accuracy"])}, is lower than {accuracy_threshold}',
             )
+            self.assertGreaterEqual(
+                metrics["output_throughput"],
+                output_throughput_threshold,
+                f'Output throughput of {self.model} is {str(metrics["output_throughput"])}, is lower than {output_throughput_threshold}',
+            )
+        except Exception as e:
+            model_metrics["error"] = e
+            self.fail(f"Test failed for {self.model}: {e}")
+        finally:
+            write_results_to_github_step_summary({self.model: model_metrics})

python/sglang/test/ascend/test_ascend_utils.py

@@ -24,7 +24,9 @@ from sglang.test.test_utils import (
     DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
     DEFAULT_URL_FOR_TEST,
     auto_config_device,
+    is_in_ci,
     popen_launch_server,
+    write_github_step_summary,
 )
 
 # Model weights storage directory
@@ -90,6 +92,9 @@ LLAMA_3_2_1B_WEIGHTS_PATH = os.path.join(MODEL_WEIGHTS_DIR, "LLM-Research/Llama-
 LLAMA_4_SCOUT_17B_16E_INSTRUCT_WEIGHTS_PATH = os.path.join(
     MODEL_WEIGHTS_DIR, "meta-llama/Llama-4-Scout-17B-16E-Instruct"
 )
+LLaDA2_0_MINI_WEIGHTS_PATH = os.path.join(
+    MODEL_WEIGHTS_DIR, "inclusionAI/LLaDA2.0-mini"
+)
 META_LLAMA_3_1_8B_INSTRUCT = os.path.join(
     MODEL_WEIGHTS_DIR, "LLM-Research/Meta-Llama-3.1-8B-Instruct"
 )
@@ -555,3 +560,46 @@ def run_bench_serving(
 
     assert res["completed"] == num_prompts
     return res
+
+
+HEADER = """
+### Models
+| Model | Server | Client | Output Throughput | Expected Output Throughput | Latency | Expected Latency | Accuracy | Expected Accuracy | Status |
+| ----- | ------ | ------ | -------- | ------------------ | ------- | ---------------- | -------- | --------- | ------ |
+"""
+
+
+def write_results_to_github_step_summary(results: dict):
+    if not is_in_ci():
+        return
+
+    write_github_step_summary_once(HEADER)
+
+    get_float = lambda metrics, item, precision: (
+        f"{metrics[item]:.{precision}f}"
+        if isinstance(metrics.get(item, "-"), (int, float))
+        else metrics.get(item, "-")
+    )
+
+    summary = ""
+    for model, metrics in results.items():
+        model = model.replace(MODEL_WEIGHTS_DIR, "").replace(HF_MODEL_WEIGHTS_DIR, "")
+        output_throughput = get_float(metrics, "output_throughput", 2)
+        output_throughput_threshold = metrics.get("output_throughput_threshold", "N/A")
+        accuracy = get_float(metrics, "accuracy", 4)
+        accuracy_threshold = metrics.get("accuracy_threshold", "N/A")
+        latency = get_float(metrics, "latency", 4)
+        latency_threshold = metrics.get("latency_threshold", "N/A")
+        server = metrics.get("server", "N/A")
+        client = metrics.get("client", "N/A")
+        error = metrics.get("error", "")
+        status = "✅" if error == "" else "❌ " + str(error)
+        summary += f"| {model} | {server} | {client} | {output_throughput} | {output_throughput_threshold} | {latency} | {latency_threshold} | {accuracy} | {accuracy_threshold} | {status} |\n"
+    write_github_step_summary(summary)
+
+
+def write_github_step_summary_once(summary: str):
+    if getattr(write_github_step_summary_once, "has_written", False):
+        return
+    write_github_step_summary_once.has_written = True
+    write_github_step_summary(summary)

python/sglang/test/ascend/test_mmlu.py

@@ -0,0 +1,37 @@
+import subprocess
+from types import SimpleNamespace
+
+from sglang.test.ascend.test_ascend_utils import write_results_to_github_step_summary
+from sglang.test.run_eval import run_eval
+
+
+class TestMMLU:
+
+    def test_mmlu(self):
+        accuracy_mmlu_threshold = getattr(self, "accuracy_mmlu", 0.00)
+
+        model_metrics = {
+            "server": getattr(
+                self, "server_cmd", subprocess.list2cmdline(map(str, self.other_args))
+            ),
+            "client": "simple_eval_mmlu",
+            "accuracy_threshold": getattr(self, "accuracy_mmlu", "N/A"),
+        }
+
+        try:
+            args = SimpleNamespace(
+                base_url=self.base_url,
+                model=self.model,
+                eval_name="mmlu",
+                num_examples=128,
+                num_threads=32,
+            )
+            print("Starting mmlu test...")
+            metrics = run_eval(args)
+            model_metrics["accuracy"] = metrics["score"]
+            self.assertGreater(metrics["score"], accuracy_mmlu_threshold)
+        except Exception as e:
+            model_metrics["error"] = e
+            self.fail(f"Test failed for {self.model}: {e}")
+        finally:
+            write_results_to_github_step_summary({self.model: model_metrics})

python/sglang/test/ascend/vlm_utils.py

@@ -4,6 +4,7 @@ import os
 import subprocess
 
 from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.test_ascend_utils import write_results_to_github_step_summary
 from sglang.test.test_utils import (
     DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
     DEFAULT_URL_FOR_TEST,
@@ -96,6 +97,8 @@ class TestVLMModels(CustomTestCase):
             timeout=3600,
         )
 
+        return subprocess.list2cmdline(cmd)  # Return the command for logging purposes
+
     def _run_vlm_mmmu_test(
         self,
         output_path="./logs",
@@ -115,8 +118,15 @@ class TestVLMModels(CustomTestCase):
         """
         print(f"\nTesting model: {self.model}{test_name}")
 
+        model_metrics = {
+            "server": subprocess.list2cmdline(map(str, self.other_args)),
+            "client": "mmmu_eval",
+            "accuracy_threshold": self.mmmu_accuracy,
+        }
+
         process = None
         server_output = ""
+        mmmu_accuracy = None
 
         try:
             # Prepare environment variables
@@ -143,8 +153,10 @@ class TestVLMModels(CustomTestCase):
                 ),
             )
 
+            model_metrics["server"] = subprocess.list2cmdline(process.args)
+
             # Run evaluation
-            self.run_mmmu_eval(self.model, output_path, limit)
+            model_metrics["client"] = self.run_mmmu_eval(self.model, output_path, limit)
 
             # Get the result file
             result_file_path = glob.glob(f"{output_path}/*.json")[0]
@@ -163,6 +175,8 @@ class TestVLMModels(CustomTestCase):
             if capture_output and process:
                 server_output = self._read_output_from_files()
 
+            model_metrics["accuracy"] = mmmu_accuracy
+
             # Assert performance meets expected threshold
             self.assertGreaterEqual(
                 mmmu_accuracy,
@@ -173,10 +187,12 @@ class TestVLMModels(CustomTestCase):
             return server_output
 
         except Exception as e:
+            model_metrics["error"] = e
             print(f"Error testing {self.model}{test_name}: {e}")
             self.fail(f"Test failed for {self.model}{test_name}: {e}")
-
         finally:
+            write_results_to_github_step_summary({self.model: model_metrics})
+
             # Ensure process cleanup happens regardless of success/failure
             if process is not None and process.poll() is None:
                 print(f"Cleaning up process {process.pid}")

test/registered/ascend/basic_function/dllm/test_npu_llada2_mini.py

@@ -1,17 +1,13 @@
 import os
 import unittest
-from types import SimpleNamespace
 
-from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.gsm8k_ascend_mixin import GSM8KAscendMixin
+from sglang.test.ascend.test_ascend_utils import LLaDA2_0_MINI_WEIGHTS_PATH
 from sglang.test.ci.ci_register import register_npu_ci
-from sglang.test.few_shot_gsm8k import run_eval as run_eval_few_shot_gsm8k
 from sglang.test.send_one import BenchArgs, send_one_prompt
 from sglang.test.test_utils import (
-    DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
-    DEFAULT_URL_FOR_TEST,
     CustomTestCase,
     is_in_ci,
-    popen_launch_server,
     write_github_step_summary,
 )
 
@@ -19,14 +15,8 @@ register_npu_ci(est_time=400, suite="stage-b-test-4-npu-a3", nightly=False)
 register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
 
 
-class TestLLaDA2Mini(CustomTestCase):
+class TestLLaDA2Mini(GSM8KAscendMixin, CustomTestCase):
-    @classmethod
+    model = LLaDA2_0_MINI_WEIGHTS_PATH
-    def setUpClass(cls):
-        cls._old_disable_acl = os.environ.get("SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT")
-        os.environ["SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT"] = "1"
-
-        cls.model = "/root/.cache/modelscope/hub/models/inclusionAI/LLaDA2.0-mini"
-        cls.base_url = DEFAULT_URL_FOR_TEST
 
     other_args = [
         "--trust-remote-code",
@@ -40,38 +30,12 @@ class TestLLaDA2Mini(CustomTestCase):
         "--dllm-algorithm",
         "LowConfidence",  # TODO: Add dLLM configurations
     ]
-
+    env = {
-        cls.process = popen_launch_server(
+        **os.environ,
-            cls.model,
+        "SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT": "1",  # Need to avoid OOM issue
-            cls.base_url,
+    }
-            timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
+    accuracy = 0.88
-            other_args=other_args,
+    output_throughput = 70
-        )
-
-    @classmethod
-    def tearDownClass(cls):
-        kill_process_tree(cls.process.pid)
-
-        if cls._old_disable_acl is None:
-            os.environ.pop("SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT", None)
-        else:
-            os.environ["SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT"] = cls._old_disable_acl
-
-    def test_gsm8k(self):
-        args = SimpleNamespace(
-            num_shots=5,
-            data_path=None,
-            num_questions=200,
-            max_new_tokens=512,
-            parallel=128,
-            host="http://127.0.0.1",
-            port=int(self.base_url.split(":")[-1]),
-        )
-        metrics = run_eval_few_shot_gsm8k(args)
-        print(f"{metrics=}")
-
-        self.assertGreater(metrics["accuracy"], 0.88)
-        self.assertGreater(metrics["output_throughput"], 70)
 
     def test_bs_1_speed(self):
         args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)

test/registered/ascend/basic_function/optimization_debug/test_npu_piecewise_graph_prefill.py

@@ -1,38 +1,27 @@
+import subprocess
 import unittest
-from urllib.parse import urlparse
 
-from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.gsm8k_ascend_mixin import GSM8KAscendMixin
+from sglang.test.ascend.test_ascend_utils import (
+    QWEN2_5_7B_INSTRUCT_WEIGHTS_PATH,
+    write_results_to_github_step_summary,
+)
 from sglang.test.ci.ci_register import register_npu_ci
-from sglang.test.few_shot_gsm8k import run_eval as run_eval_few_shot_gsm8k
 from sglang.test.test_utils import (
-    DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
-    DEFAULT_URL_FOR_TEST,
     CustomTestCase,
-    SimpleNamespace,
-    popen_launch_server,
     run_bench_one_batch,
 )
 
 register_npu_ci(est_time=400, suite="stage-b-test-1-npu-a2", nightly=False)
 register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
 
-MODEL = "/root/.cache/modelscope/hub/models/Qwen/Qwen2.5-7B-Instruct"
+
-GSM8K_EXP_ACCURACY = 0.84
-EXP_PREFILL_LATENCY = 0.045
 TOKENS_TO_CAPTURE = [i for i in range(128, 4096, 128)]
 
 
-class TestPiecewiseGraphPrefillCorrectness(CustomTestCase):
+class TestPiecewiseGraphPrefillCorrectness(GSM8KAscendMixin, CustomTestCase):
-    @classmethod
+    model = QWEN2_5_7B_INSTRUCT_WEIGHTS_PATH
-    def setUpClass(cls):
+    other_args = [
-        cls.model = MODEL
-        cls.base_url = DEFAULT_URL_FOR_TEST
-        cls.url = urlparse(DEFAULT_URL_FOR_TEST)
-        cls.process = popen_launch_server(
-            cls.model,
-            cls.base_url,
-            timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
-            other_args=[
         "--trust-remote-code",
         "--mem-fraction-static",
         0.8,
@@ -43,39 +32,14 @@ class TestPiecewiseGraphPrefillCorrectness(CustomTestCase):
         "--enforce-piecewise-cuda-graph",
         "--piecewise-cuda-graph-tokens",
         *TOKENS_TO_CAPTURE,
-            ],
+    ]
-        )
+    accuracy = 0.84
-
+    num_questions = 1319
-    @classmethod
-    def tearDownClass(cls):
-        kill_process_tree(cls.process.pid)
-
-    def test_gsm8k(self):
-        print(f"##=== Testing accuracy: {self.model} ===##")
-        args = SimpleNamespace(
-            num_shots=5,
-            data_path=None,
-            num_questions=1319,
-            max_new_tokens=512,
-            parallel=128,
-            host=f"http://{self.url.hostname}",
-            port=int(self.url.port),
-        )
-
-        metrics = run_eval_few_shot_gsm8k(args)
-        self.assertGreaterEqual(
-            metrics["accuracy"],
-            GSM8K_EXP_ACCURACY,
-        )
 
 
 class TestPiecewiseGraphPrefillBenchmark(CustomTestCase):
-
+    model = QWEN2_5_7B_INSTRUCT_WEIGHTS_PATH
-    def test_latency(self):
+    other_args = [
-        print(f"##=== Testing prefill latency: {MODEL} ===##")
-        prefill_latency, _, _ = run_bench_one_batch(
-            MODEL,
-            other_args=[
         "--trust-remote-code",
         "--mem-fraction-static",
         0.8,
@@ -83,10 +47,30 @@ class TestPiecewiseGraphPrefillBenchmark(CustomTestCase):
         "ascend",
         "--enforce-piecewise-cuda-graph",
         "--piecewise-cuda-graph-tokens",
-            ]
+    ] + TOKENS_TO_CAPTURE
-            + TOKENS_TO_CAPTURE,
+
+    latency = 0.045
+
+    def test_latency(self):
+        print(f"##=== Testing prefill latency: {self.model} ===##")
+        model_metrics = {
+            "server": subprocess.list2cmdline(map(str, self.other_args)),
+            "client": "bench_one_batch",
+            "latency_threshold": self.latency,
+        }
+        try:
+            prefill_latency, _, _ = run_bench_one_batch(
+                self.model,
+                other_args=self.other_args,
             )
-        self.assertLess(prefill_latency, EXP_PREFILL_LATENCY)
+            model_metrics["latency"] = float(prefill_latency)
+            self.assertLess(prefill_latency, self.latency)
+        except Exception as e:
+            model_metrics["error"] = e
+            print(f"Error testing {self.model}: {e}")
+            self.fail(f"Test failed for {self.model}: {e}")
+        finally:
+            write_results_to_github_step_summary({self.model: model_metrics})
 
 
 if __name__ == "__main__":

test/registered/ascend/basic_function/parallel_strategy/expert_parallelism/test_npu_deepep_auto_deepseek_v3_2_w8a8.py

@@ -1,22 +1,16 @@
 import os
 import unittest
-from types import SimpleNamespace
 
-from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.gsm8k_ascend_mixin import GSM8KAscendMixin
 from sglang.test.ascend.test_ascend_utils import DEEPSEEK_V3_2_W8A8_WEIGHTS_PATH
+from sglang.test.ascend.test_mmlu import TestMMLU
 from sglang.test.ci.ci_register import register_npu_ci
-from sglang.test.few_shot_gsm8k import run_eval as run_gsm8k
+from sglang.test.test_utils import CustomTestCase
-from sglang.test.run_eval import run_eval
-from sglang.test.test_utils import (
-    DEFAULT_URL_FOR_TEST,
-    CustomTestCase,
-    popen_launch_server,
-)
 
 register_npu_ci(est_time=400, suite="nightly-16-npu-a3", nightly=True)
 
 
-class TestDeepEpDeepseekV32(CustomTestCase):
+class TestDeepEpDeepseekV32(GSM8KAscendMixin, TestMMLU, CustomTestCase):
     """Testcase: Verify that for the DeepSeek V3.2 model in the single-machine colocation scenario,
     its inference accuracy on the MMLU and GSM8K dataset meets the preset standard when the parameter --deepep-mode auto is configured.
 
@@ -24,15 +18,10 @@ class TestDeepEpDeepseekV32(CustomTestCase):
     [Test Target] --moe-a2a-backend deepep;--deepep-mode
     """
 
-    @classmethod
+    model = DEEPSEEK_V3_2_W8A8_WEIGHTS_PATH
-    def setUpClass(cls):
+
-        cls.model = DEEPSEEK_V3_2_W8A8_WEIGHTS_PATH
+    timeout_for_server_launch = 60000
-        cls.base_url = DEFAULT_URL_FOR_TEST
+    other_args = [
-        cls.process = popen_launch_server(
-            cls.model,
-            cls.base_url,
-            timeout=6000,
-            other_args=[
         "--trust-remote-code",
         "--tp-size",
         "16",
@@ -52,8 +41,10 @@ class TestDeepEpDeepseekV32(CustomTestCase):
         40960,
         "--max-total-tokens",
         40960,
-            ],
+    ]
-            env={
+
+    env = {
+        **os.environ,
         "PYTORCH_NPU_ALLOC_CONF": "expandable_segments:True",
         "STREAMS_PER_DEVICE": "32",
         "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": "16",
@@ -62,46 +53,10 @@ class TestDeepEpDeepseekV32(CustomTestCase):
         "SGLANG_NPU_USE_MLAPO": "0",
         "SGLANG_NPU_USE_MULTI_STREAM": "1",
         "TASK_QUEUE_ENABLE": "0",
-                **os.environ,
+    }
-            },
-        )
 
-    @classmethod
+    accuracy = 0.95  # Test GSM8K accuracy ≥0.95
-    def tearDownClass(cls):
+    accuracy_mmlu = 0.85  # Test MMLU accuracy ≥0.85
-        kill_process_tree(cls.process.pid)
-
-    def test_mmlu(self):
-        expect_score = 0.85
-        args = SimpleNamespace(
-            base_url=self.base_url,
-            model=self.model,
-            eval_name="mmlu",
-            num_examples=128,
-            num_threads=32,
-        )
-        print("Starting mmlu test...")
-        metrics = run_eval(args)
-        self.assertGreater(metrics["score"], expect_score)
-
-    def test_gsm8k(self):
-        expect_accuracy = 0.95
-        args = SimpleNamespace(
-            num_shots=8,
-            data_path=None,
-            timeout=60000,
-            num_questions=200,
-            max_new_tokens=512,
-            parallel=128,
-            host="http://127.0.0.1",
-            port=int(self.base_url.split(":")[-1]),
-        )
-        print("Starting gsm8k test...")
-        metrics = run_gsm8k(args)
-        self.assertGreaterEqual(
-            metrics["accuracy"],
-            expect_accuracy,
-            f'Accuracy of {self.model} is {str(metrics["accuracy"])}, is lower than {expect_accuracy}',
-        )
 
 
 if __name__ == "__main__":

test/registered/ascend/basic_function/speculative_inference/test_npu_eagle3.py

@@ -1,39 +1,27 @@
 import os
 import unittest
-from types import SimpleNamespace
-from urllib.parse import urlparse
 
-from sglang.srt.utils import kill_process_tree
+from sglang.test.ascend.gsm8k_ascend_mixin import GSM8KAscendMixin
 from sglang.test.ascend.test_ascend_utils import (
     QWEN3_8B_EAGLE3_WEIGHTS_PATH,
     QWEN3_8B_WEIGHTS_PATH,
 )
 from sglang.test.ci.ci_register import register_npu_ci
-from sglang.test.few_shot_gsm8k import run_eval as run_eval_few_shot_gsm8k
+from sglang.test.test_utils import CustomTestCase
-from sglang.test.test_utils import (
-    DEFAULT_URL_FOR_TEST,
-    CustomTestCase,
-    popen_launch_server,
-)
 
 register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
 
 
-class TestNpuEagle3(CustomTestCase):
+class TestNpuEagle3(GSM8KAscendMixin, CustomTestCase):
     """Testcase: Verify GSM8K inference accuracy ≥0.81 for model with specified EAGLE3 speculative inference parameters.
 
     [Test Category] Speculative Decoding
     [Test Target] --speculative-draft-model-quantization; --speculative-algorithm; --speculative-draft-model-path; --speculative-num-steps; --speculative-eagle-topk; --speculative-num-draft-tokens; --speculative-attention-mode
     """
 
-    @classmethod
+    model = QWEN3_8B_WEIGHTS_PATH
-    def setUpClass(cls):
+    timeout_for_server_launch = 1500
-        cls.model = QWEN3_8B_WEIGHTS_PATH
+    other_args = [
-        cls.accuracy = 0.81
-        cls.base_url = DEFAULT_URL_FOR_TEST
-        cls.url = urlparse(DEFAULT_URL_FOR_TEST)
-
-        cls.common_args = [
         "--trust-remote-code",
         "--attention-backend",
         "ascend",
@@ -61,39 +49,13 @@ class TestNpuEagle3(CustomTestCase):
         "bfloat16",
     ]
 
-        cls.extra_envs = {
+    env = {
+        **os.environ,
         "SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
     }
-        os.environ.update(cls.extra_envs)
 
-    def test_gsm8k(self):
+    accuracy = 0.81
-        process = popen_launch_server(
+    num_questions = 1319
-            self.model,
-            self.base_url,
-            timeout=1500,
-            other_args=[
-                *self.common_args,
-            ],
-        )
-
-        try:
-            args = SimpleNamespace(
-                num_shots=5,
-                data_path=None,
-                num_questions=1319,
-                max_new_tokens=512,
-                parallel=128,
-                host=f"http://{self.url.hostname}",
-                port=int(self.url.port),
-            )
-
-            metrics = run_eval_few_shot_gsm8k(args)
-            self.assertGreaterEqual(
-                metrics["accuracy"],
-                self.accuracy,
-            )
-        finally:
-            kill_process_tree(process.pid)
 
 
 if __name__ == "__main__":