diff --git a/test/registered/unit/checkpoint_engine/test_checkpoint_engine_worker.py b/test/registered/unit/checkpoint_engine/test_checkpoint_engine_worker.py deleted file mode 100644 index 50ece8d34ab3..000000000000 --- a/test/registered/unit/checkpoint_engine/test_checkpoint_engine_worker.py +++ /dev/null @@ -1,141 +0,0 @@ -"""Unit tests for srt/checkpoint_engine/checkpoint_engine_worker.py — no server, no model loading. - -Focus: device resolution so the ZMQ handshake key matches checkpoint-engine's -ParameterServer (ps.py::_get_physical_gpu_id) on every backend -- ``GPU-`` -for CUDA/XPU and ``NPU-`` for NPU. These paths are pure namespace routing -(``get_device`` / ``get_device_module`` / ``is_npu``) and are fully mockable on CPU. - -Skipped entirely unless the ``checkpoint-engine`` extra is installed, since the -worker module refuses to import without it. -""" - -from sglang.test.ci.ci_register import register_cpu_ci - -# Unit tests may register CPU suites only (scripts/lint/check_registered_tests.py). -# TestWorkerDeviceUuidOnXpu below still self-skips off XPU, so it stays runnable -# by hand on an Intel GPU host. -register_cpu_ci(est_time=5, suite="base-a-test-cpu") - -import importlib.util -import unittest -from unittest.mock import MagicMock, patch - -import torch - -from sglang.srt.utils import is_xpu -from sglang.test.test_utils import CustomTestCase - -# checkpoint-engine is an optional extra (sglang[checkpoint-engine]) that CI does -# not install, and the worker module raises ImportError at import time without it. -# Probe first so this file stays importable: an unguarded import would fail -# collection and take the whole file down rather than skipping. -_HAS_CHECKPOINT_ENGINE = importlib.util.find_spec("checkpoint_engine") is not None - -if _HAS_CHECKPOINT_ENGINE: - from sglang.srt.checkpoint_engine.checkpoint_engine_worker import ( - SGLangCheckpointEngineWorkerExtensionImpl, - ) - -_WORKER_MOD = "sglang.srt.checkpoint_engine.checkpoint_engine_worker" -_NO_CKPT_ENGINE = "requires the checkpoint-engine optional dependency" - - -@unittest.skipUnless(_HAS_CHECKPOINT_ENGINE, _NO_CKPT_ENGINE) -class TestWorkerDeviceResolution(CustomTestCase): - """get_device_uuid / get_device_id must route through the active accelerator - namespace and emit the key the ParameterServer expects.""" - - def _make_worker(self): - # model_runner is unused by the device-resolution methods under test. - return SGLangCheckpointEngineWorkerExtensionImpl(model_runner=MagicMock()) - - def _fake_device_module(self, *, current=3): - mod = MagicMock() - mod.current_device.return_value = current - return mod - - def _fake_platform(self, *, uuid="abcd-1234"): - # The uuid now comes from the platform layer (current_platform), which - # already returns str(get_device_properties(id).uuid) for cuda/xpu. - plat = MagicMock() - plat.get_device_uuid.return_value = uuid - return plat - - def test_device_uuid_cuda(self): - worker = self._make_worker() - fake = self._fake_device_module(current=0) - with ( - patch(f"{_WORKER_MOD}.is_npu", return_value=False), - patch(f"{_WORKER_MOD}.get_device_module", return_value=fake), - patch( - f"{_WORKER_MOD}.current_platform", self._fake_platform(uuid="cuda-uuid") - ), - ): - self.assertEqual(worker.get_device_uuid(), "GPU-cuda-uuid") - self.assertEqual(worker.get_device_id(), 0) - - def test_device_uuid_xpu(self): - worker = self._make_worker() - fake = self._fake_device_module(current=2) - with ( - patch(f"{_WORKER_MOD}.is_npu", return_value=False), - patch(f"{_WORKER_MOD}.get_device_module", return_value=fake), - patch( - f"{_WORKER_MOD}.current_platform", self._fake_platform(uuid="xpu-uuid") - ), - ): - # XPU shares CUDA's GPU- format; only the namespace differs. - self.assertEqual(worker.get_device_uuid(), "GPU-xpu-uuid") - self.assertEqual(worker.get_device_id(), 2) - - def test_device_uuid_npu_uses_npu_prefix(self): - # NPU must NOT be treated as CUDA: the ParameterServer keys it as - # NPU-, so a GPU- key would never resolve. - worker = self._make_worker() - with ( - patch(f"{_WORKER_MOD}.is_npu", return_value=True), - patch( - "checkpoint_engine.device_utils.npu_generate_uuid", - return_value="1.2.3.4-0", - ), - ): - self.assertEqual(worker.get_device_uuid(), "NPU-1.2.3.4-0") - - def test_device_uuid_wraps_assertion_error(self): - worker = self._make_worker() - fake = self._fake_device_module(current=1) - plat = MagicMock() - plat.get_device_uuid.side_effect = AssertionError("no uuid") - with ( - patch(f"{_WORKER_MOD}.is_npu", return_value=False), - patch(f"{_WORKER_MOD}.get_device_module", return_value=fake), - patch(f"{_WORKER_MOD}.current_platform", plat), - self.assertRaises(ValueError), - ): - worker.get_device_uuid() - - -@unittest.skipUnless(_HAS_CHECKPOINT_ENGINE, _NO_CKPT_ENGINE) -@unittest.skipUnless(is_xpu(), "requires an Intel XPU") -class TestWorkerDeviceUuidOnXpu(CustomTestCase): - """Hardware-gated: the real XPU key must match what checkpoint-engine's - ParameterServer derives, or the ZMQ handshake silently fails on XPU.""" - - def test_real_uuid_matches_parameter_server(self): - from checkpoint_engine.device_utils import DeviceManager - from checkpoint_engine.ps import _get_physical_gpu_id - - worker = SGLangCheckpointEngineWorkerExtensionImpl(model_runner=MagicMock()) - key = worker.get_device_uuid() - - self.assertTrue(key.startswith("GPU-"), key) - self.assertEqual(worker.get_device_id(), torch.xpu.current_device()) - - # Independently derived by the ParameterServer side; the two must agree. - dm = DeviceManager() - self.assertEqual(dm.device_type, "xpu") - self.assertEqual(key, _get_physical_gpu_id(dm, torch.xpu.current_device())) - - -if __name__ == "__main__": - unittest.main(verbosity=3) diff --git a/test/registered/unit/entrypoints/test_http2_server_config.py b/test/registered/unit/entrypoints/test_http2_server_config.py deleted file mode 100644 index 07791f3c8263..000000000000 --- a/test/registered/unit/entrypoints/test_http2_server_config.py +++ /dev/null @@ -1,45 +0,0 @@ -import importlib.util -import unittest -from unittest.mock import patch - -from sglang.srt.entrypoints.http_server import _run_granian_server -from sglang.test.ci.ci_register import register_cpu_ci - -register_cpu_ci(est_time=12, suite="base-a-test-cpu") - - -@unittest.skipUnless( - importlib.util.find_spec("granian"), "granian is required for HTTP/2" -) -class TestGranianHTTP2Config(unittest.TestCase): - def test_passes_explicit_max_concurrent_streams(self): - configured = {} - - class FakeEmbeddedServer: - def __init__(self, **kwargs): - configured.update(kwargs) - - async def serve(self): - return None - - def stop(self): - return None - - with patch("granian.server.embed.Server", FakeEmbeddedServer): - _run_granian_server( - host="127.0.0.1", - port=30000, - log_level="info", - http2_max_concurrent_streams=37, - http2_initial_connection_window_size=8 * 1024 * 1024, - ) - - self.assertEqual(configured["http2_settings"].max_concurrent_streams, 37) - self.assertEqual( - configured["http2_settings"].initial_connection_window_size, - 8 * 1024 * 1024, - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/registered/unit/layers/test_moriep_mxfp8_dispatch.py b/test/registered/unit/layers/test_moriep_mxfp8_dispatch.py deleted file mode 100644 index 60fcc2ffc288..000000000000 --- a/test/registered/unit/layers/test_moriep_mxfp8_dispatch.py +++ /dev/null @@ -1,81 +0,0 @@ -"""Contract tests for the MoRI MXFP8 dispatch dtype. - -MXFP8 dispatch sends an fp8 payload with group-32 e8m0 microscales, which is -exactly what the per_1x32 (MXFP4-weight) MoE kernels consume. The value of the -mode rests on that byte layout being right: an fp8 payload with the wrong scale -group size or the wrong scale dtype still runs, but silently reintroduces the -upscale round trip the mode exists to remove, and the only symptom is lost -throughput. - -These pin the layout arithmetic and the env-var wiring, which are the parts that -can regress silently. They do not need a GPU. -""" - -import pytest - -torch = pytest.importorskip("torch") - -from sglang.srt.layers.moe.token_dispatcher.moriep import ( # noqa: E402 - MXFP4_BLOCK_SIZE, - DispatchDtype, -) -from sglang.test.ci.ci_register import register_cpu_ci # noqa: E402 - -register_cpu_ci(est_time=10, suite="base-a-test-cpu") - -HIDDEN = 7168 # DeepSeek-V4 - - -def test_mxfp8_member_exists_and_is_distinct(): - assert hasattr(DispatchDtype, "mxfp8") - values = {d.value for d in DispatchDtype} - assert len(values) == len(list(DispatchDtype)), "duplicate DispatchDtype value" - - -def test_scale_group_size_is_32(): - """per_1x32 is the whole point: group-128 scales would force the receiver - back through an fp8->bf16 upscale.""" - assert MXFP4_BLOCK_SIZE == 32 - - -def test_scale_dim_matches_group_32_layout(): - """One scale per 32 channels. A mismatch here under-allocates the scale - buffer and the kernels read past it.""" - assert HIDDEN % MXFP4_BLOCK_SIZE == 0 - assert HIDDEN // MXFP4_BLOCK_SIZE == 224 - - -def test_e8m0_scale_is_one_byte(): - """The dispatch buffer is sized from this. float32 scales would need 4x the - room and silently truncate the payload.""" - assert torch.float8_e8m0fnu.itemsize == 1 - - -@pytest.mark.parametrize( - "name,expected", - [ - ("bf16", DispatchDtype.bf16), - ("fp8", DispatchDtype.fp8), - ("fp4", DispatchDtype.fp4), - ("mxfp8", DispatchDtype.mxfp8), - ], -) -def test_env_override_maps_to_member(name, expected): - """SGLANG_MORI_DISPATCH_DTYPE is the only way to reach this mode, so an - unmapped string would leave it silently on the bf16 default.""" - assert DispatchDtype(expected.value) is expected - assert expected.name == name - - -def test_empty_token_batch_scale_shape(): - """Decode can hand a rank zero live tokens. The empty branch must still - produce a correctly shaped scale tensor or the all-to-all desyncs.""" - scale = torch.empty((0, HIDDEN // MXFP4_BLOCK_SIZE), dtype=torch.float8_e8m0fnu) - assert scale.shape == (0, 224) - assert scale.dtype == torch.float8_e8m0fnu - - -if __name__ == "__main__": - import sys - - sys.exit(pytest.main([__file__, "-v"])) diff --git a/test/registered/unit/managers/test_flat_raw_top_logprobs.py b/test/registered/unit/managers/test_flat_raw_top_logprobs.py index 9535d8e1bd19..2326fc645a12 100644 --- a/test/registered/unit/managers/test_flat_raw_top_logprobs.py +++ b/test/registered/unit/managers/test_flat_raw_top_logprobs.py @@ -4,10 +4,7 @@ import asyncio import base64 -import json -import os import pickle -import time import unittest from array import array from types import SimpleNamespace @@ -696,127 +693,5 @@ def test_fields_default_none(self): self.assertIsNone(output.input_top_logprobs_flat_null_prefix) -@unittest.skipUnless( - os.environ.get("SGLANG_BENCH_FLAT_RAW_TOP_LOGPROBS"), - "Serialization microbenchmark; set SGLANG_BENCH_FLAT_RAW_TOP_LOGPROBS=1 to run.", -) -class BenchFlatRawTopLogprobsSerialization(CustomTestCase): - """Round-trip cost of the formats: server assembly + json.dumps, then - client json.loads + reconstruction into [rows, k] arrays.""" - - def test_bench(self): - num_positions, k = 32768, 2 - rng = np.random.default_rng(0) - vals = rng.standard_normal((num_positions, k)).astype(np.float32) - idxs = rng.integers(0, 150000, size=(num_positions, k), dtype=np.int32) - val_rows = [None] + vals[1:].tolist() - idx_rows = [None] + idxs[1:].tolist() - - def best_of(fn, iters=5): - result = fn() - elapsed = min( - (lambda s=time.perf_counter(): (fn(), time.perf_counter() - s)[1])() - for _ in range(iters) - ) - return elapsed * 1e3, result - - def bench(name, build, decode): - encode_ms, payload = best_of(lambda: json.dumps(build())) - decode_ms, arrays = best_of(lambda: decode(payload)) - self.assertEqual(arrays[0].shape, (num_positions - 1, k)) - print( - f"{name}: encode {encode_ms:.1f} ms, decode {decode_ms:.1f} ms, " - f"{len(payload)} bytes" - ) - - def decode_nested(payload): - rows = [r for r in json.loads(payload) if r is not None] - return ( - np.array([[e[0] for e in r] for r in rows], dtype=np.float32), - np.array([[e[1] for e in r] for r in rows], dtype=np.int32), - ) - - def decode_flat(payload): - d = json.loads(payload) - shape = d["input_top_logprobs_shape"] - return ( - np.asarray(d["input_top_logprobs_val_flat"], np.float32).reshape(shape), - np.asarray(d["input_top_logprobs_idx_flat"], np.int32).reshape(shape), - ) - - bench( - "nested triples", - lambda: [ - (None if row is None else [(v, i, None) for v, i in zip(row, idx_row)]) - for row, idx_row in zip(val_rows, idx_rows) - ], - decode_nested, - ) - bench( - "flat lists", - lambda: _build_flat_input_top_logprobs_fields( - val_rows, idx_rows, top_logprobs_num=k - ), - decode_flat, - ) - - def decode_b64(payload): - d = json.loads(payload) - shape = d["input_top_logprobs_shape"] - return ( - np.frombuffer( - base64.b64decode(d["input_top_logprobs_val_flat_b64"]), - np.dtype(d["input_top_logprobs_val_flat_b64_dtype"]), - ).reshape(shape), - np.frombuffer( - base64.b64decode(d["input_top_logprobs_idx_flat_b64"]), - np.dtype(d["input_top_logprobs_idx_flat_b64_dtype"]), - ).reshape(shape), - ) - - bench( - "flat b64", - lambda: _build_flat_input_top_logprobs_fields( - val_rows, idx_rows, top_logprobs_num=k, return_b64=True - ), - decode_b64, - ) - - def test_bench_ipc_pickle(self): - """Inter-process cost of BatchTokenIDOutput input-top fields: nested - per-position rows vs scheduler-flat arrays (two ZMQ pickle hops each - pay dumps + loads).""" - num_positions, k = 32768, 2 - rng = np.random.default_rng(0) - vals = rng.standard_normal((num_positions, k)).astype(np.float32) - idxs = rng.integers(0, 150000, size=(num_positions, k), dtype=np.int32) - - def best_of(fn, iters=10): - return min( - (lambda s=time.perf_counter(): (fn(), time.perf_counter() - s)[1])() - for _ in range(iters) - ) - - nested = _make_batch_token_id_output( - input_top_logprobs_val=[[None] + vals[1:].tolist(), []], - input_top_logprobs_idx=[[None] + idxs[1:].tolist(), []], - ) - flat = _make_batch_token_id_output( - input_top_logprobs_val_flat=[vals[1:], None], - input_top_logprobs_idx_flat=[idxs[1:], None], - input_top_logprobs_flat_null_prefix=[1, None], - ) - for name, obj in (("nested rows", nested), ("flat arrays", flat)): - payload = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL) - dumps_ms = best_of( - lambda o=obj: pickle.dumps(o, protocol=pickle.HIGHEST_PROTOCOL) - ) - loads_ms = best_of(lambda p=payload: pickle.loads(p)) - print( - f"{name}: pickle.dumps {dumps_ms * 1e3:.2f} ms, " - f"pickle.loads {loads_ms * 1e3:.2f} ms, {len(payload) / 1e6:.2f} MB" - ) - - if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/test/registered/unit/managers/test_profile_merger_http_api.py b/test/registered/unit/managers/test_profile_merger_http_api.py deleted file mode 100644 index 1dd0836e24f3..000000000000 --- a/test/registered/unit/managers/test_profile_merger_http_api.py +++ /dev/null @@ -1,61 +0,0 @@ -import unittest - -from sglang.srt.managers.io_struct import ProfileReq -from sglang.test.ci.ci_register import register_cpu_ci -from sglang.test.test_utils import CustomTestCase - -register_cpu_ci(est_time=11, suite="base-a-test-cpu") - - -class TestProfileMergerHTTPAPI(CustomTestCase): - def test_profile_req_merge_profiles_json_deserialization(self): - # Test JSON data as would come from HTTP request - json_data = { - "output_dir": "/tmp/test", - "num_steps": 10, - "activities": ["CPU", "GPU", "MEM"], - "profile_by_stage": False, - "merge_profiles": True, - } - - # Create ProfileReq from dict (as HTTP server would do) - req = ProfileReq(**json_data) - - self.assertTrue(req.merge_profiles) - self.assertEqual(req.output_dir, "/tmp/test") - self.assertEqual(req.num_steps, 10) - self.assertEqual(req.activities, ["CPU", "GPU", "MEM"]) - self.assertFalse(req.profile_by_stage) - - def test_profile_req_merge_profiles_default_value(self): - # Test with minimal data - json_data = {"output_dir": "/tmp/test"} - - req = ProfileReq(**json_data) - self.assertFalse(req.merge_profiles) - - def test_profile_req_merge_profiles_explicit_false(self): - json_data = {"output_dir": "/tmp/test", "merge_profiles": False} - - req = ProfileReq(**json_data) - self.assertFalse(req.merge_profiles) - - def test_http_api_parameter_validation(self): - # Test with True - json_data = {"merge_profiles": True} - req = ProfileReq(**json_data) - self.assertTrue(req.merge_profiles) - - # Test with False - json_data = {"merge_profiles": False} - req = ProfileReq(**json_data) - self.assertFalse(req.merge_profiles) - - # Test with string "true" (should be converted by JSON parser) - json_data = {"merge_profiles": "true"} - req = ProfileReq(**json_data) - self.assertEqual(req.merge_profiles, "true") # String, not boolean - - -if __name__ == "__main__": - unittest.main() diff --git a/test/registered/unit/model_loader/test_modelopt_export.py b/test/registered/unit/model_loader/test_modelopt_export.py deleted file mode 100644 index 25313f28241c..000000000000 --- a/test/registered/unit/model_loader/test_modelopt_export.py +++ /dev/null @@ -1,316 +0,0 @@ -""" -Unit tests for ModelOpt export functionality in SGLang. - -These tests verify the integration of ModelOpt export API with SGLang's model loading -and quantization workflow. -""" - -import json -import os -import tempfile -import unittest -from unittest.mock import Mock, patch - -import torch - -from sglang.srt.configs.device_config import DeviceConfig -from sglang.srt.configs.load_config import LoadConfig -from sglang.srt.configs.model_config import ModelConfig -from sglang.srt.model_loader.loader import ModelOptModelLoader -from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci - -register_cuda_ci(est_time=11, stage="base-b", runner_config="1-gpu-large") -register_amd_ci(est_time=9, suite="stage-b-test-1-gpu-small-amd") - -# Note: PYTHONPATH=python should be set when running tests - -# Check if modelopt is available -try: - import modelopt # noqa: F401 - - MODELOPT_AVAILABLE = True -except ImportError: - MODELOPT_AVAILABLE = False - - -class TestModelOptExport(unittest.TestCase): - """Test suite for ModelOpt export functionality.""" - - def setUp(self): - """Set up test fixtures.""" - # Mock distributed functionality to avoid initialization errors - self.mock_tp_rank = patch( - "sglang.srt.distributed.parallel_state.get_tensor_model_parallel_rank", - return_value=0, - ) - self.mock_tp_rank.start() - - self.mock_rank0_log = patch("sglang.srt.model_loader.loader.rank0_log") - self.mock_rank0_log.start() - - # Mock logger to avoid issues - self.mock_logger = patch("sglang.srt.model_loader.loader.logger") - self.mock_logger.start() - - # Mock all distributed functions that might be called - self.mock_get_tp_group = patch("sglang.srt.distributed.parallel_state._TP") - self.mock_get_tp_group.start() - - # Mock model parallel initialization check - self.mock_mp_is_initialized = patch( - "sglang.srt.distributed.parallel_state.model_parallel_is_initialized", - return_value=True, - ) - self.mock_mp_is_initialized.start() - self.temp_dir = tempfile.mkdtemp() - self.export_dir = os.path.join(self.temp_dir, "exported_model") - self.checkpoint_dir = os.path.join(self.temp_dir, "checkpoint") - - # Mock model - self.mock_model = Mock(spec=torch.nn.Module) - self.mock_model.device = torch.device("cuda:0") - - # Mock tokenizer - self.mock_tokenizer = Mock() - - # Mock quantization config - self.mock_quant_cfg = Mock() - - # Create ModelOptModelLoader instance - self.load_config = LoadConfig() - self.model_loader = ModelOptModelLoader(self.load_config) - - def tearDown(self): - """Clean up test fixtures.""" - import shutil - - shutil.rmtree(self.temp_dir, ignore_errors=True) - - # Stop mocks - self.mock_tp_rank.stop() - self.mock_rank0_log.stop() - self.mock_logger.stop() - self.mock_get_tp_group.stop() - self.mock_mp_is_initialized.stop() - - def _create_mock_export_files(self, export_dir: str): - """Create mock export files for testing validation.""" - os.makedirs(export_dir, exist_ok=True) - - # Create config.json - config = { - "model_type": "test_model", - "architectures": ["TestModel"], - "quantization_config": { - "quant_method": "modelopt", - "bits": 8, - }, - } - with open(os.path.join(export_dir, "config.json"), "w") as f: - json.dump(config, f) - - # Create tokenizer_config.json - tokenizer_config = {"tokenizer_class": "TestTokenizer"} - with open(os.path.join(export_dir, "tokenizer_config.json"), "w") as f: - json.dump(tokenizer_config, f) - - # Create model file - with open(os.path.join(export_dir, "model.safetensors"), "w") as f: - f.write("mock_model_data") - - @unittest.skipIf(not MODELOPT_AVAILABLE, "nvidia-modelopt not available") - @patch("sglang.srt.model_loader.loader.os.makedirs") - @patch("modelopt.torch.export.export_hf_checkpoint") - def test_export_modelopt_checkpoint_success(self, mock_export, mock_makedirs): - """Test successful model export.""" - # Arrange - mock_export.return_value = None - mock_makedirs.return_value = None - - # Act - self.model_loader._export_modelopt_checkpoint(self.mock_model, self.export_dir) - - # Assert - mock_makedirs.assert_called_once_with(self.export_dir, exist_ok=True) - mock_export.assert_called_once_with(self.mock_model, export_dir=self.export_dir) - - @unittest.skipIf(not MODELOPT_AVAILABLE, "nvidia-modelopt not available") - @patch("modelopt.torch.opt.restore") - @patch("modelopt.torch.quantization.utils.is_quantized") - def test_setup_quantization_with_export_from_checkpoint( - self, mock_is_quantized, mock_restore - ): - """Test export functionality when restoring from checkpoint.""" - # Arrange - mock_is_quantized.return_value = False - mock_restore.return_value = None - - with patch.object( - self.model_loader, "_export_modelopt_checkpoint" - ) as mock_export: - # Act - self.model_loader._setup_modelopt_quantization( - self.mock_model, - self.mock_tokenizer, - self.mock_quant_cfg, - quantized_ckpt_restore_path=self.checkpoint_dir, - export_path=self.export_dir, - ) - - # Assert - mock_restore.assert_called_once_with(self.mock_model, self.checkpoint_dir) - mock_export.assert_called_once_with(self.mock_model, self.export_dir, None) - - @unittest.skipIf(not MODELOPT_AVAILABLE, "nvidia-modelopt not available") - @patch("modelopt.torch.quantization.quantize") - @patch("modelopt.torch.quantization.print_quant_summary") - @patch("modelopt.torch.quantization.utils.is_quantized") - @patch("modelopt.torch.utils.dataset_utils.get_dataset_dataloader") - @patch("modelopt.torch.utils.dataset_utils.create_forward_loop") - def test_setup_quantization_with_export_after_calibration( - self, - mock_create_loop, - mock_get_dataloader, - mock_is_quantized, - mock_print_summary, - mock_quantize, - ): - """Test export functionality after calibration-based quantization.""" - # Arrange - mock_is_quantized.return_value = False - mock_dataloader = Mock() - mock_get_dataloader.return_value = mock_dataloader - mock_calibrate_loop = Mock() - mock_create_loop.return_value = mock_calibrate_loop - mock_quantize.return_value = None - mock_print_summary.return_value = None - - with patch.object( - self.model_loader, "_export_modelopt_checkpoint" - ) as mock_export: - # Act - self.model_loader._setup_modelopt_quantization( - self.mock_model, - self.mock_tokenizer, - self.mock_quant_cfg, - export_path=self.export_dir, - ) - - # Assert - mock_quantize.assert_called_once_with( - self.mock_model, self.mock_quant_cfg, forward_loop=mock_calibrate_loop - ) - mock_export.assert_called_once_with(self.mock_model, self.export_dir, None) - - def test_quantize_and_serve_config_validation(self): - """Test that quantize_and_serve is properly disabled.""" - # Test that quantize-and-serve mode raises NotImplementedError - with self.assertRaises(NotImplementedError) as context: - ModelConfig( - model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", - quantization="modelopt_fp8", - quantize_and_serve=True, - ) - - # Verify the error message contains helpful instructions - error_msg = str(context.exception) - self.assertIn("disabled due to compatibility issues", error_msg) - self.assertIn("separate quantize-then-deploy workflow", error_msg) - - # Test invalid configuration - no quantization - with self.assertRaises(ValueError) as context: - ModelConfig( - model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", - quantize_and_serve=True, - ) - self.assertIn("requires ModelOpt quantization", str(context.exception)) - - @unittest.skipIf(not MODELOPT_AVAILABLE, "nvidia-modelopt not available") - def test_standard_workflow_selection(self): - """Test that standard workflow is selected by default.""" - with patch( - "modelopt.torch.quantization.utils.is_quantized", return_value=False - ): - with patch.object( - self.model_loader, "_standard_quantization_workflow" - ) as mock_standard: - with patch.object(self.model_loader, "_load_modelopt_base_model"): - mock_standard.return_value = Mock() - - # Create model config without quantize_and_serve - model_config = ModelConfig( - model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", - quantization="modelopt_fp8", - quantize_and_serve=False, - ) - device_config = DeviceConfig() - - # Act - self.model_loader.load_model( - model_config=model_config, - device_config=device_config, - ) - - # Assert - mock_standard.assert_called_once_with(model_config, device_config) - - -@unittest.skipIf(not MODELOPT_AVAILABLE, "nvidia-modelopt not available") -class TestModelOptExportIntegration(unittest.TestCase): - """Integration tests for ModelOpt export with full model loading workflow.""" - - def setUp(self): - """Set up integration test fixtures.""" - self.temp_dir = tempfile.mkdtemp() - self.export_dir = os.path.join(self.temp_dir, "exported_model") - - def tearDown(self): - """Clean up integration test fixtures.""" - import shutil - - shutil.rmtree(self.temp_dir, ignore_errors=True) - - @patch("sglang.srt.model_loader.loader.get_model_architecture") - @patch("transformers.AutoTokenizer.from_pretrained") - @patch("transformers.AutoModelForCausalLM.from_pretrained") - def test_full_workflow_with_export(self, mock_model, mock_tokenizer, mock_arch): - """Test the complete workflow from model config to export.""" - # Arrange - mock_arch.return_value = ("TestModel", "TestConfig") - mock_tokenizer.return_value = Mock() - mock_model.return_value = Mock(spec=torch.nn.Module) - - model_config = ModelConfig( - model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", - quantization="modelopt_fp8", - ) - - load_config = LoadConfig(modelopt_export_path=self.export_dir) - device_config = DeviceConfig() - - # Mock the quantization and export process - with patch.object( - ModelOptModelLoader, "_setup_modelopt_quantization" - ) as mock_setup: - with patch.object( - ModelOptModelLoader, "_load_modelopt_base_model" - ) as mock_load_base: - mock_load_base.return_value = mock_model.return_value - - # Act - model_loader = ModelOptModelLoader(load_config) - result = model_loader.load_model( - model_config=model_config, - device_config=device_config, - ) - - # Assert - self.assertIsNotNone(result) - mock_setup.assert_called_once() - # Verify export_path was passed to setup - args, kwargs = mock_setup.call_args - self.assertEqual(kwargs.get("export_path"), self.export_dir) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/registered/unit/model_loader/test_modelopt_loader.py b/test/registered/unit/model_loader/test_modelopt_loader.py index db6fa5ad73ed..86bdcce7d2ec 100644 --- a/test/registered/unit/model_loader/test_modelopt_loader.py +++ b/test/registered/unit/model_loader/test_modelopt_loader.py @@ -215,194 +215,6 @@ def test_calibration_workflow_integration(self, mock_logger, mock_auto_tokenizer # Note: We can't easily verify the exact calls due to dynamic imports, # but we can verify the workflow completed successfully - @patch("sglang.srt.model_loader.loader.QUANT_CFG_CHOICES", QUANT_CFG_CHOICES) - @patch("sglang.srt.model_loader.loader.AutoTokenizer") - @patch("sglang.srt.model_loader.loader.logger") - def test_quantized_checkpoint_restore(self, mock_logger, mock_auto_tokenizer): - """Test restoring from a quantized checkpoint.""" - - # Create model config with checkpoint restore path - config_with_restore = ModelConfig( - model_path=self.model_path, - quantization="modelopt_fp8", - ) - - # Create load config with checkpoint restore path - load_config_with_restore = LoadConfig( - modelopt_checkpoint_restore_path="/path/to/quantized/checkpoint" - ) - - loader = ModelOptModelLoader(load_config_with_restore) - - # Mock tokenizer - mock_tokenizer = MagicMock() - mock_auto_tokenizer.from_pretrained.return_value = mock_tokenizer - - # Mock modelopt modules - mock_mtq = MagicMock() - mock_mto = MagicMock() - - # Configure quantization config - mock_fp8_cfg = MagicMock() - mock_mtq.FP8_DEFAULT_CFG = mock_fp8_cfg - - # Configure model as not quantized initially - mock_is_quantized = MagicMock(return_value=False) - - with patch.object( - loader, "_load_modelopt_base_model", return_value=self.mock_base_model - ): - with patch.dict( - "sys.modules", - { - "modelopt": MagicMock(), - "modelopt.torch": MagicMock(), - "modelopt.torch.opt": mock_mto, - "modelopt.torch.quantization": mock_mtq, - "modelopt.torch.quantization.utils": MagicMock( - is_quantized=mock_is_quantized - ), - }, - ): - with patch.object(loader, "_setup_modelopt_quantization") as mock_setup: - # Mock the _setup_modelopt_quantization to simulate checkpoint restore - def mock_setup_quantization( - model, - tokenizer, - quant_cfg, - quantized_ckpt_restore_path=None, - **kwargs, - ): - if quantized_ckpt_restore_path: - mock_mto.restore(model, quantized_ckpt_restore_path) - print( - f"Restored quantized model from {quantized_ckpt_restore_path}" - ) - return - - mock_setup.side_effect = mock_setup_quantization - - # Execute the load_model method - result_model = loader.load_model( - model_config=config_with_restore, - device_config=self.device_config, - ) - - # Verify the setup was called with restore path - mock_setup.assert_called_once() - call_args = mock_setup.call_args - # Check that the restore path was passed correctly - self.assertIn("quantized_ckpt_restore_path", call_args[1]) - self.assertEqual( - call_args[1]["quantized_ckpt_restore_path"], - "/path/to/quantized/checkpoint", - ) - - # Verify restore was called - mock_mto.restore.assert_called_once_with( - self.mock_base_model, "/path/to/quantized/checkpoint" - ) - - # Verify we get the expected model back - self.assertEqual(result_model, self.mock_base_model) - - @patch("sglang.srt.model_loader.loader.QUANT_CFG_CHOICES", QUANT_CFG_CHOICES) - @patch("sglang.srt.model_loader.loader.AutoTokenizer") - @patch("sglang.srt.model_loader.loader.logger") - def test_quantized_checkpoint_save(self, mock_logger, mock_auto_tokenizer): - """Test saving quantized checkpoint after calibration.""" - - # Create model config with checkpoint save path - config_with_save = ModelConfig( - model_path=self.model_path, - quantization="modelopt_fp8", - ) - - # Create load config with checkpoint save path - load_config_with_save = LoadConfig( - modelopt_checkpoint_save_path="/path/to/save/checkpoint" - ) - - loader = ModelOptModelLoader(load_config_with_save) - - # Mock tokenizer - mock_tokenizer = MagicMock() - mock_auto_tokenizer.from_pretrained.return_value = mock_tokenizer - - # Mock modelopt modules - mock_mtq = MagicMock() - mock_mto = MagicMock() - mock_dataset_utils = MagicMock() - - # Configure quantization config - mock_fp8_cfg = MagicMock() - mock_mtq.FP8_DEFAULT_CFG = mock_fp8_cfg - - # Configure model as not quantized initially - mock_is_quantized = MagicMock(return_value=False) - - with patch.object( - loader, "_load_modelopt_base_model", return_value=self.mock_base_model - ): - with patch.dict( - "sys.modules", - { - "modelopt": MagicMock(), - "modelopt.torch": MagicMock(), - "modelopt.torch.opt": mock_mto, - "modelopt.torch.quantization": mock_mtq, - "modelopt.torch.quantization.utils": MagicMock( - is_quantized=mock_is_quantized - ), - "modelopt.torch.utils": MagicMock(), - "modelopt.torch.utils.dataset_utils": mock_dataset_utils, - }, - ): - with patch.object(loader, "_setup_modelopt_quantization") as mock_setup: - # Mock the _setup_modelopt_quantization to simulate checkpoint save - def mock_setup_quantization( - model, - tokenizer, - quant_cfg, - quantized_ckpt_save_path=None, - **kwargs, - ): - # Simulate calibration and quantization - mock_mtq.quantize(model, quant_cfg, forward_loop=MagicMock()) - mock_mtq.print_quant_summary(model) - - # Save checkpoint if path provided - if quantized_ckpt_save_path: - mock_mto.save(model, quantized_ckpt_save_path) - print( - f"Quantized model saved to {quantized_ckpt_save_path}" - ) - - mock_setup.side_effect = mock_setup_quantization - - # Execute the load_model method - result_model = loader.load_model( - model_config=config_with_save, device_config=self.device_config - ) - - # Verify the setup was called with save path - mock_setup.assert_called_once() - call_args = mock_setup.call_args - # Check that the save path was passed correctly - self.assertIn("quantized_ckpt_save_path", call_args[1]) - self.assertEqual( - call_args[1]["quantized_ckpt_save_path"], - "/path/to/save/checkpoint", - ) - - # Verify save was called - mock_mto.save.assert_called_once_with( - self.mock_base_model, "/path/to/save/checkpoint" - ) - - # Verify we get the expected model back - self.assertEqual(result_model, self.mock_base_model) - def test_unified_quantization_flag_support(self): """Test that ModelOptModelLoader supports unified quantization flags.""" # Test modelopt_fp8 @@ -422,6 +234,29 @@ def test_unified_quantization_flag_support(self): # Should default to fp8 when no config is detected self.assertEqual(config_auto._get_modelopt_quant_type(), "fp8") + def test_quantize_and_serve_config_validation(self): + """Test that quantize_and_serve is properly disabled.""" + # Test that quantize-and-serve mode raises NotImplementedError + with self.assertRaises(NotImplementedError) as context: + ModelConfig( + model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", + quantization="modelopt_fp8", + quantize_and_serve=True, + ) + + # Verify the error message contains helpful instructions + error_msg = str(context.exception) + self.assertIn("disabled due to compatibility issues", error_msg) + self.assertIn("separate quantize-then-deploy workflow", error_msg) + + # Test invalid configuration - no quantization + with self.assertRaises(ValueError) as context: + ModelConfig( + model_path="TinyLlama/TinyLlama-1.1B-Chat-v1.0", + quantize_and_serve=True, + ) + self.assertIn("requires ModelOpt quantization", str(context.exception)) + class TestModelOptLoaderIntegration(CustomTestCase): """Integration tests for ModelOptModelLoader with Engine API.""" diff --git a/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py b/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py deleted file mode 100644 index fa146e8bfa83..000000000000 --- a/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py +++ /dev/null @@ -1,72 +0,0 @@ -"""Guard the PaddleOCR-VL serving defaults that a refactor could silently drop. - -Both settings here live in allowlists keyed by model type / architecture, so -nothing in PaddleOCR-VL's own code path breaks if an entry disappears — the -model just quietly serves slower. - -A document page costs tens of milliseconds to resize + normalize + patchify, so -a single synchronous processor worker caps request throughput at -1 / preprocess_time no matter how much GPU is idle. Measured on an H200 with -1080p pages, opting into concurrent workers moved 32-way concurrent throughput -from 6.6 to 8.9 req/s and made single-stream TTFT stable (the single-worker -path alternated between ~282 ms and ~790 ms). - -The opt-in lives on the class, and `QwenVLImageProcessor` grants it only to an -explicit `model_type` allowlist that PaddleOCR-VL is not on — so it is exactly -the kind of setting a refactor can silently drop. -""" - -import pytest - -from sglang.srt.configs.model_config import ( - multimodal_breakable_cuda_graph_supported_model_archs, -) -from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor -from sglang.srt.multimodal.processors.paddleocr_vlm import PaddleOCRVLImageProcessor -from sglang.test.ci.ci_register import register_cpu_ci - -register_cpu_ci(est_time=12, suite="base-a-test-cpu") - - -def test_processor_preprocesses_pages_concurrently(): - assert PaddleOCRVLImageProcessor.supports_mm_processor_concurrency is True - assert PaddleOCRVLImageProcessor.auto_mm_processor_worker_num > 1 - assert PaddleOCRVLImageProcessor.auto_mm_io_worker_num > 1 - - -def test_worker_count_stays_at_the_measured_optimum(): - """Two beat both one and four at 32-way concurrency on an H200, on document - pages and on small images with long outputs alike. Past two, spreading - request arrivals fragments GPU prefill batches faster than the extra overlap - pays for itself.""" - assert PaddleOCRVLImageProcessor.auto_mm_processor_worker_num == 2 - - -def test_io_worker_count_is_this_model_own(): - """Concurrency is the base default now, but the IO fan-out is not. - - Fetching a page is network-bound and cheap to overlap, so this model asks for - more IO workers than the conservative base default. That number has to be - declared here, not inherited. - """ - assert PaddleOCRVLImageProcessor.__dict__["auto_mm_io_worker_num"] > ( - BaseMultimodalProcessor.auto_mm_io_worker_num - ), "the IO fan-out must be declared on PaddleOCRVLImageProcessor itself" - - -def test_prefill_breakable_cuda_graph_is_allowlisted(): - """Breakable CG is the CUDA default but is switched off for every multimodal - arch; PaddleOCR-VL opts back in so its text-only prefill keeps the graph. - - Measured on an H200 (2704-token text prompts): single-stream TTFT 16.1 ms - without the graph, 11.5 ms with it. Image-carrying batches are rejected at - replay and run eager either way, so this is a text/mixed-traffic win only. - """ - assert ( - "PaddleOCRVLForConditionalGeneration" - in multimodal_breakable_cuda_graph_supported_model_archs - ) - - -if __name__ == "__main__": - raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/test/registered/unit/multimodal/test_base_processor_image_decode.py b/test/registered/unit/multimodal/test_base_processor_image_decode.py index f6ca7a22355a..f9f6f97d4598 100644 --- a/test/registered/unit/multimodal/test_base_processor_image_decode.py +++ b/test/registered/unit/multimodal/test_base_processor_image_decode.py @@ -69,12 +69,6 @@ def _is_decoded(img: Image.Image) -> bool: class TestLoadSingleItemImageDecode(CustomTestCase): - def test_plain_open_is_lazy(self): - # Documents why the fix matters: a bare Image.open is not decoded yet, so - # without the fix the decode would land on the caller (main) thread. - lazy = Image.open(io.BytesIO(_png_bytes())) - self.assertFalse(_is_decoded(lazy)) - def test_load_single_item_forces_decode(self): img = _StubProcessor._load_single_item(_png_bytes("RGB"), Modality.IMAGE) self.assertIsInstance(img, Image.Image) diff --git a/test/registered/unit/multimodal/test_processor_clone_isolation.py b/test/registered/unit/multimodal/test_processor_clone_isolation.py index 0ab717dfa6f2..3107f4443cda 100644 --- a/test/registered/unit/multimodal/test_processor_clone_isolation.py +++ b/test/registered/unit/multimodal/test_processor_clone_isolation.py @@ -33,11 +33,6 @@ def _preprocess(self, images, do_resize=None, do_rescale=None): class TestSarashina2PreprocessFilterSurvivesCloning(CustomTestCase): - def test_unfiltered_preprocess_rejects_what_transformers_forwards(self): - """Why the filter exists: the raw method cannot take the full kwarg set.""" - with self.assertRaises(TypeError): - _NarrowImageProcessor()._preprocess(["img"], do_resize=True, do_pad=False) - def test_filter_applies_to_the_patched_processor(self): image_processor = _NarrowImageProcessor() _install_preprocess_kwarg_filter(image_processor) diff --git a/test/registered/unit/npu/attention/test_npu_ascend_backend.py b/test/registered/unit/npu/attention/test_npu_ascend_backend.py index 06189ef8d2a2..c98aa0af314c 100644 --- a/test/registered/unit/npu/attention/test_npu_ascend_backend.py +++ b/test/registered/unit/npu/attention/test_npu_ascend_backend.py @@ -4,7 +4,6 @@ import sys import unittest -from dataclasses import fields, is_dataclass from types import SimpleNamespace from unittest.mock import MagicMock @@ -32,7 +31,6 @@ AscendAttnBackend, AscendAttnMaskBuilder, AscendAttnMultiStepDraftBackend, - ForwardMetadata, _expand_dsa_sparse_indices, _reshape_kv_for_fia_nz, ) @@ -110,58 +108,6 @@ def test_view_relationship(self): self.assertEqual(result.data_ptr(), tensor.data_ptr()) -class TestForwardMetadata(unittest.TestCase): - def test_is_dataclass(self): - self.assertTrue(is_dataclass(ForwardMetadata)) - - def test_all_fields_default_none(self): - metadata = ForwardMetadata() - for f in fields(ForwardMetadata): - self.assertIsNone( - getattr(metadata, f.name), - f"Field '{f.name}' should default to None", - ) - - def test_create_with_values(self): - block_tables = torch.tensor([[1, 2], [3, 4]]) - seq_lens = torch.tensor([10, 20]) - metadata = ForwardMetadata( - block_tables=block_tables, - seq_lens=seq_lens, - seq_lens_cpu_list=[10, 20], - ) - self.assertTrue(torch.equal(metadata.block_tables, block_tables)) - self.assertTrue(torch.equal(metadata.seq_lens, seq_lens)) - self.assertEqual(metadata.seq_lens_cpu_list, [10, 20]) - - def test_partial_assignment(self): - metadata = ForwardMetadata(swa_mask=torch.ones(3, 3)) - self.assertIsNotNone(metadata.swa_mask) - self.assertIsNone(metadata.block_tables) - self.assertIsNone(metadata.seq_lens) - - def test_field_names(self): - names = {f.name for f in fields(ForwardMetadata)} - expected = { - "block_tables", - "block_tables_swa", - "swa_out_cache_loc", - "extend_seq_lens_cpu_int", - "seq_lens_cpu_int", - "seq_lens_cpu_list", - "seq_lens_list_cumsum", - "seq_lens", - "actual_seq_lengths_q", - "actual_seq_lengths_q_pa", - "actual_seq_lengths_q_pa_cpu", - "actual_seq_lengths_kv", - "swa_mask", - "prefix_lens", - "flatten_prefix_block_tables", - } - self.assertEqual(names, expected) - - class TestGenerateMaskFlag(unittest.TestCase): def test_shape(self): mask = AscendAttnMaskBuilder.generate_mask_flag(8) @@ -682,12 +628,6 @@ def test_default_dtype_bfloat16(self): self.assertEqual(result.dtype, torch.bfloat16) -class TestGetCudaGraphSeqLenFillValue(unittest.TestCase): - def test_returns_zero(self): - backend = object.__new__(AscendAttnBackend) - self.assertEqual(backend.get_cuda_graph_seq_len_fill_value(), 0) - - class TestGetVerifyBuffers(unittest.TestCase): def test_no_verify_mask(self): backend = object.__new__(AscendAttnBackend) diff --git a/test/registered/unit/npu/attention/test_npu_ascend_torch_native_backend.py b/test/registered/unit/npu/attention/test_npu_ascend_torch_native_backend.py index 98c1eb5a6f62..10377c93dc35 100644 --- a/test/registered/unit/npu/attention/test_npu_ascend_torch_native_backend.py +++ b/test/registered/unit/npu/attention/test_npu_ascend_torch_native_backend.py @@ -16,18 +16,6 @@ register_npu_ci(est_time=4, suite="stage-a-unit-test-npu") -class TestInit(unittest.TestCase): - def test_construction(self): - backend = AscendTorchNativeAttnBackend() - self.assertIsNotNone(backend) - - -class TestSupportTriton(unittest.TestCase): - def test_returns_false(self): - backend = AscendTorchNativeAttnBackend() - self.assertFalse(backend.support_triton()) - - class TestScaledDotProductAttentionWithSoftcapping(unittest.TestCase): def setUp(self): self.backend = AscendTorchNativeAttnBackend() diff --git a/test/registered/unit/parser/test_reasoning_content_without_parser.py b/test/registered/unit/parser/test_reasoning_content_without_parser.py deleted file mode 100644 index ebb982276a88..000000000000 --- a/test/registered/unit/parser/test_reasoning_content_without_parser.py +++ /dev/null @@ -1,47 +0,0 @@ -import unittest - -from sglang.srt.parser.reasoning_parser import ReasoningParser -from sglang.test.ci.ci_register import register_cpu_ci -from sglang.test.test_utils import CustomTestCase - -register_cpu_ci(est_time=10, suite="base-a-test-cpu") -register_cpu_ci(est_time=6, suite="stage-b-test-cpu-intel") - -# Simulated model output that contains think tags (e.g. from DeepSeek-R1) -THINK_OUTPUT = ( - "\nLet me think about this.\n1 + 3 = 4\n\nThe answer is 4." -) -THINK_OUTPUT_QWEN3 = ( - "\nLet me think about this.\n1 + 3 = 4\n\n\nThe answer is 4." -) - - -class TestReasoningContentWithoutParser(CustomTestCase): - """Test the code path: when no reasoning parser is configured, reasoning - content should never be separated, even if the model output contains - think tags. This mirrors the guard in serving_chat.py: - - if self.reasoning_parser and request.separate_reasoning: - ... - - When reasoning_parser is None the block is skipped entirely. - """ - - def test_with_parser_separates_reasoning(self): - """With a parser, reasoning content is correctly separated.""" - for parser_name, output in [ - ("deepseek-r1", THINK_OUTPUT), - ("qwen3", THINK_OUTPUT_QWEN3), - ]: - with self.subTest(parser=parser_name): - parser = ReasoningParser(parser_name, stream_reasoning=False) - reasoning_text, text = parser.parse_non_stream(output) - - self.assertIsNotNone(reasoning_text) - self.assertGreater(len(reasoning_text), 0) - self.assertNotIn("", reasoning_text) - self.assertIn("The answer is 4.", text) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/registered/unit/spec/test_ngram_mamba_verify_update.py b/test/registered/unit/spec/test_ngram_mamba_verify_update.py index fe062eb25605..616279b8c7db 100644 --- a/test/registered/unit/spec/test_ngram_mamba_verify_update.py +++ b/test/registered/unit/spec/test_ngram_mamba_verify_update.py @@ -10,89 +10,6 @@ register_cpu_ci(est_time=23, suite="base-a-test-cpu") -class TestNgramLastCorrectStepIndices(CustomTestCase): - def _compute_last_correct_step_indices( - self, - accept_indices: torch.Tensor, - num_correct_drafts: torch.Tensor, - draft_token_num: int, - ) -> torch.Tensor: - bs = accept_indices.shape[0] - req_idx = torch.arange(bs, dtype=torch.int64, device=accept_indices.device) - accept_indices_offset = (req_idx * draft_token_num).to(accept_indices.dtype) - last_correct_step_indices = ( - accept_indices[req_idx, num_correct_drafts.to(torch.int64)] - - accept_indices_offset - ) - return last_correct_step_indices - - def test_linear_chain_all_accepted(self): - bs, draft_token_num = 3, 5 - accept_indices = torch.stack( - [ - torch.arange( - i * draft_token_num, - i * draft_token_num + draft_token_num, - dtype=torch.int32, - ) - for i in range(bs) - ] - ) - num_correct_drafts = torch.tensor([4, 4, 4], dtype=torch.int32) - - result = self._compute_last_correct_step_indices( - accept_indices, num_correct_drafts, draft_token_num - ) - expected = torch.tensor([4, 4, 4], dtype=torch.int32) - self.assertTrue(torch.equal(result, expected)) - - def test_linear_chain_partial_accept(self): - bs, draft_token_num = 3, 5 - accept_indices = torch.tensor( - [ - [0, 1, 2, -1, -1], - [5, -1, -1, -1, -1], - [10, 11, 12, 13, 14], - ], - dtype=torch.int32, - ) - num_correct_drafts = torch.tensor([2, 0, 4], dtype=torch.int32) - - result = self._compute_last_correct_step_indices( - accept_indices, num_correct_drafts, draft_token_num - ) - expected = torch.tensor([2, 0, 4], dtype=torch.int32) - self.assertTrue(torch.equal(result, expected)) - - def test_tree_structure_non_sequential(self): - bs, draft_token_num = 2, 6 - accept_indices = torch.tensor( - [ - [0, 2, 5, -1, -1, -1], - [6, 7, 10, -1, -1, -1], - ], - dtype=torch.int32, - ) - num_correct_drafts = torch.tensor([2, 2], dtype=torch.int32) - - result = self._compute_last_correct_step_indices( - accept_indices, num_correct_drafts, draft_token_num - ) - expected = torch.tensor([5, 4], dtype=torch.int32) - self.assertTrue(torch.equal(result, expected)) - - def test_single_request_zero_drafts(self): - bs, draft_token_num = 1, 4 - accept_indices = torch.tensor([[0, -1, -1, -1]], dtype=torch.int32) - num_correct_drafts = torch.tensor([0], dtype=torch.int32) - - result = self._compute_last_correct_step_indices( - accept_indices, num_correct_drafts, draft_token_num - ) - expected = torch.tensor([0], dtype=torch.int32) - self.assertTrue(torch.equal(result, expected)) - - class TestNgramMambaVerifyUpdate(CustomTestCase): def _make_mock_target_worker(self): target_worker = MagicMock() @@ -231,149 +148,6 @@ def test_mamba_verify_update_with_track_indices(self): ) -class TestConvWindowDedupLayout(CustomTestCase): - """KDA stores conv_state as (K-1, channel), unlike GDN; partial-accept - commits must preserve that layout in the overlapping view. - """ - - @staticmethod - def _build_fixed_view(channel_dim, win_len, draft_tokens, window_major, device): - shared_win = draft_tokens + win_len - 1 - L, S = 1, 1 - phys = torch.zeros(L, S, channel_dim, shared_win, device=device) - # Encoding both coordinates makes axis aliasing observable. - for c in range(channel_dim): - for w in range(shared_win): - phys[0, 0, c, w] = c * 1000 + w - if not window_major: - # GDN: view[l, s, step, d, w] = phys[l, s, d, step + w] - view = phys.as_strided( - (L, S, draft_tokens, channel_dim, win_len), - ( - phys.stride(0), - phys.stride(1), - phys.stride(3), - phys.stride(2), - phys.stride(3), - ), - ) - else: - # KDA: view[l, s, step, w, d] = phys[l, s, d, step + w] - view = phys.as_strided( - (L, S, draft_tokens, win_len, channel_dim), - ( - phys.stride(0), - phys.stride(1), - phys.stride(3), - phys.stride(3), - phys.stride(2), - ), - ) - return view, phys - - @staticmethod - def _build_buggy_kda_view(channel_dim, win_len, draft_tokens, device): - """Preserve the former axis swap so the regression test distinguishes - the corrected view from the broken one. - """ - conv_shape = (win_len, channel_dim) - conv_dim, win = conv_shape - shared_win = draft_tokens + win - 1 - L, S = 1, 1 - phys = torch.zeros(L, S, conv_dim, shared_win, device=device) - for c in range(conv_dim): - for w in range(shared_win): - phys[0, 0, c, w] = c * 1000 + w - view = phys.as_strided( - (L, S, draft_tokens, conv_dim, win), - ( - phys.stride(0), - phys.stride(1), - phys.stride(3), - phys.stride(2), - phys.stride(3), - ), - ) - return view - - def test_kda_window_major_sliding_window(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - view, _ = self._build_fixed_view( - channel_dim, win_len, draft_tokens, window_major=True, device="cpu" - ) - for t in range(draft_tokens): - for w in range(win_len): - for d in range(channel_dim): - got = int(view[0, 0, t, w, d].item()) - self.assertEqual( - got, - d * 1000 + (t + w), - msg=f"KDA view alias at step={t} w={w} d={d}", - ) - - def test_kda_channel_axis_independent(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - view, _ = self._build_fixed_view( - channel_dim, win_len, draft_tokens, window_major=True, device="cpu" - ) - for t in range(draft_tokens): - for w in range(win_len): - for d in range(channel_dim): - self.assertEqual(int(view[0, 0, t, w, d].item()) // 1000, d) - - def test_kda_window_shifts_by_one_per_step(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - view, _ = self._build_fixed_view( - channel_dim, win_len, draft_tokens, window_major=True, device="cpu" - ) - fixed_channel = 2 - for t in range(draft_tokens - 1): - a = view[0, 0, t, :, fixed_channel].tolist() - b = view[0, 0, t + 1, :, fixed_channel].tolist() - self.assertEqual(a[1:], b[:-1]) - - def test_gdn_channel_major_unchanged(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - view, _ = self._build_fixed_view( - channel_dim, win_len, draft_tokens, window_major=False, device="cpu" - ) - for t in range(draft_tokens): - for d in range(channel_dim): - for w in range(win_len): - self.assertEqual( - int(view[0, 0, t, d, w].item()), d * 1000 + (t + w) - ) - - def test_partial_accept_commit_reads_correct_window(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - view, _ = self._build_fixed_view( - channel_dim, win_len, draft_tokens, window_major=True, device="cpu" - ) - n = 1 - committed = view[0, 0, n] - for w in range(win_len): - for d in range(channel_dim): - self.assertEqual(int(committed[w, d].item()), d * 1000 + (n + w)) - - def test_buggy_kda_view_aliases_step_onto_channel(self): - channel_dim, win_len, draft_tokens = 5, 3, 4 - buggy = self._build_buggy_kda_view( - channel_dim, win_len, draft_tokens, device="cpu" - ) - self.assertEqual(buggy.shape[3], win_len) - self.assertEqual(buggy.shape[4], channel_dim) - aliased = False - for c in range(buggy.shape[3]): - if buggy[0, 0, 0, c, :].tolist() != buggy[0, 0, 1, c, :].tolist(): - aliased = True - break - self.assertTrue( - aliased, - "expected the buggy KDA view to alias the draft-step axis onto the " - "channel axis", - ) - - class TestMtpVerifyHookSignature(CustomTestCase): """Every ``update_mamba_state_after_mtp_verify`` override must accept the full keyword call the spec workers make, or it raises TypeError at verify time on diff --git a/test/registered/unit/test_ray_driver_reads_the_bags.py b/test/registered/unit/test_ray_driver_reads_the_bags.py deleted file mode 100644 index 3f3fb61ce18c..000000000000 --- a/test/registered/unit/test_ray_driver_reads_the_bags.py +++ /dev/null @@ -1,68 +0,0 @@ -"""The Ray driver sizes its actors from the published configuration. - -`RayEngine` publishes as part of `Engine._launch_subprocesses` and *then* lays -out the actors, so the placement arithmetic reads the `parallel` bag. That is -where a resolution decision lives: a launch that leaves `dp_size` to resolution -has it in the `parallel` bag, and the override case below is what tells the two -apart. - -There is no CI coverage of the Ray path (`test/manual/test_ray_engine.py` boots a -real cluster), so these cases drive the two pure helpers directly against a -published config -- including the override direction, which is what tells a bag -read from a record read. -""" - -import importlib.util -import unittest - -from sglang.srt.runtime_context import get_context, get_parallel -from sglang.test.ci.ci_register import register_cpu_ci -from sglang.test.test_utils import CustomTestCase - -register_cpu_ci(est_time=10, suite="base-a-test-cpu") - -# `sglang.srt.ray.engine` imports `ray` at module scope, and the CPU runner has -# no ray wheel. -_HAS_RAY = importlib.util.find_spec("ray") is not None -_needs_ray = unittest.skipUnless(_HAS_RAY, "ray is not installed") - - -class TestRayDriverReadsTheBags(CustomTestCase): - def _publish(self, **fields): - override = get_context().override_server_args(**fields) - override.install() - self.addCleanup(override.restore) - - @_needs_ray - def test_world_size_multiplies_the_published_sizes(self): - from sglang.srt.ray.engine import _compute_world_size - - self._publish(tp_size=2, pp_size=3, dp_size=4, enable_dp_attention=False) - self.assertEqual(_compute_world_size(), 24) - - @_needs_ray - def test_dp_attention_folds_dp_into_tp(self): - from sglang.srt.ray.engine import _compute_world_size - - self._publish(tp_size=4, pp_size=2, dp_size=4, enable_dp_attention=True) - # DP attention folds DP into TP, so dp_size drops out of the product. - self.assertEqual(_compute_world_size(), 8) - - @_needs_ray - def test_the_world_size_follows_a_post_publish_override(self): - """The direction that separates a bag read from a record read. - - `override` writes the bag and never the record, so a driver still - reading `server_args.tp_size` would keep answering with the old size. - """ - from sglang.srt.ray.engine import _compute_world_size - - self._publish(tp_size=2, pp_size=1, dp_size=1, enable_dp_attention=False) - self.assertEqual(_compute_world_size(), 2) - get_context().override("test.ray_driver", tp_size=8) - self.assertEqual(get_parallel().tp_size, 8) - self.assertEqual(_compute_world_size(), 8) - - -if __name__ == "__main__": - unittest.main()