test: recover the config-namespace-migration deferrals (#33171)

The module-skipped tests injected config by faking get_server_args (a
SimpleNamespace stand-in patched onto the module) or by writing fields
onto a ServerArgs instance post-publish — both invisible to the namespace
accessors the production code now reads. Re-enable them by publishing the
config they need (get_context().override_server_args seeding, scoped per
test), asserting bag state where the old assertions checked instance
write-through (declare_load_time_override is bag-only), and extending the
per-runner stubs the code genuinely reads (kv_cache_dtype_str,
max_total_tokens, context_len).

The unified-radix-cache file (which grew a large hicache/insert-walk suite
while skipped) is recovered in the same change:

- test_cache_finished_req_strips_thinking (19 parametrized classes) wrote
  strip_thinking_cache onto the ServerArgs instance; the cache reads
  get_serving().strip_thinking_cache — use the serving bag's scoped
  override.
- test_shallower_crossing_backs_up_above_backuped_middle staged its
  broken-backup-continuity setup through insert_host, which now
  deliberately drops refills below an un-backed-up node under
  write-through (host_insert_dropped). Build the same tree state through
  an explicit backup + device eviction.

Every config-namespace-migration deferral is recovered, so the deferral
ratchet (test_migration_deferral_ratchet.py) has done its job and is
retired.
This commit is contained in:
Cheng Wan
2026-08-01 08:59:27 -07:00
committed by GitHub
parent 47d8b5b749
commit 9b44695713
16 changed files with 127 additions and 386 deletions
+6 -22
View File
@@ -7,11 +7,7 @@ import torch.nn as nn
import torch.nn.functional as F import torch.nn.functional as F
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.runtime_context import get_server_args from sglang.srt.runtime_context import get_context
from sglang.srt.server_args import (
ServerArgs,
set_global_server_args_for_scheduler,
)
from sglang.srt.utils import get_device from sglang.srt.utils import get_device
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
@@ -19,21 +15,6 @@ register_cuda_ci(est_time=9, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=15, suite="stage-b-test-1-gpu-small-amd") register_amd_ci(est_time=15, suite="stage-b-test-1-gpu-small-amd")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class LMHeadStub(nn.Module): class LMHeadStub(nn.Module):
def __init__(self, vocab, hidden, dtype, device=get_device()): def __init__(self, vocab, hidden, dtype, device=get_device()):
super().__init__() super().__init__()
@@ -58,8 +39,11 @@ class TestLMHeadFP32(unittest.TestCase):
raise unittest.SkipTest("needs CUDA GPU or XPU") raise unittest.SkipTest("needs CUDA GPU or XPU")
def _make_logprocessor(self, vocab_size, enable_fp32): def _make_logprocessor(self, vocab_size, enable_fp32):
set_global_server_args_for_scheduler(ServerArgs(model_path="dummy")) # LogitsProcessor reads get_exec().features.enable_fp32_lm_head
get_server_args().enable_fp32_lm_head = enable_fp32 # from the published config.
override = get_context().override_server_args(enable_fp32_lm_head=enable_fp32)
override.install()
self.addCleanup(override.restore)
cfg = SimpleNamespace(vocab_size=vocab_size, final_logit_softcapping=None) cfg = SimpleNamespace(vocab_size=vocab_size, final_logit_softcapping=None)
return LogitsProcessor(cfg, skip_all_gather=True, logit_scale=None) return LogitsProcessor(cfg, skip_all_gather=True, logit_scale=None)
@@ -4,7 +4,7 @@
``ForwardBatch.num_token_non_padded`` is a scalar tensor on the model device ``ForwardBatch.num_token_non_padded`` is a scalar tensor on the model device
(see ``ForwardBatch.compute``, which does ``.to(device, ...)``). The eager TBO (see ``ForwardBatch.compute``, which does ``.to(device, ...)``). The eager TBO
split path already honors this -- ``compute_tbo_children_num_token_non_padded_raw`` split path already honors this -- ``compute_tbo_children_num_token_non_padded_raw``
moves the tensor to ``get_server_args().device`` -- but moves the tensor to ``get_device().device`` -- but
``TboCudaGraphRunnerPlugin`` preallocated its persistent buffer with a bare ``TboCudaGraphRunnerPlugin`` preallocated its persistent buffer with a bare
``torch.zeros((2,), dtype=torch.int32)``, leaving it on CPU. ``torch.zeros((2,), dtype=torch.int32)``, leaving it on CPU.
@@ -18,43 +18,26 @@ CPU-only: a ``meta`` device makes the buffer placement observable without a GPU.
""" """
import unittest import unittest
from types import SimpleNamespace
from unittest.mock import patch
import torch import torch
import sglang.srt.batch_overlap.two_batch_overlap as tbo
from sglang.srt.batch_overlap.two_batch_overlap import ( from sglang.srt.batch_overlap.two_batch_overlap import (
TboCudaGraphRunnerPlugin, TboCudaGraphRunnerPlugin,
TboForwardBatchPreparer, TboForwardBatchPreparer,
) )
from sglang.srt.runtime_context import get_context, get_device
from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase from sglang.test.test_utils import CustomTestCase
register_cpu_ci(est_time=5, suite="base-a-test-cpu") register_cpu_ci(est_time=5, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class TestTboCudaGraphNumTokenDevice(CustomTestCase): class TestTboCudaGraphNumTokenDevice(CustomTestCase):
def test_plugin_buffer_on_model_device(self): def test_plugin_buffer_on_model_device(self):
# Use 'meta' so the configured device differs from the implicit CPU # Use 'meta' so the configured device differs from the implicit CPU
# default; a bare torch.zeros() would leave the buffer on CPU and fail. # default; a bare torch.zeros() would leave the buffer on CPU and fail.
fake_args = SimpleNamespace(device="meta") with get_context().override_server_args():
with patch.object(tbo, "get_server_args", lambda: fake_args): with get_device().override(device="meta"):
plugin = TboCudaGraphRunnerPlugin() plugin = TboCudaGraphRunnerPlugin()
buf = plugin._tbo_children_num_token_non_padded buf = plugin._tbo_children_num_token_non_padded
@@ -65,13 +48,11 @@ class TestTboCudaGraphNumTokenDevice(CustomTestCase):
def test_graph_and_eager_paths_agree_on_device(self): def test_graph_and_eager_paths_agree_on_device(self):
# Both the preallocated cuda-graph buffer and the eager split tensor must # Both the preallocated cuda-graph buffer and the eager split tensor must
# land on the same (model) device, matching ForwardBatch's contract. # land on the same (model) device, matching ForwardBatch's contract.
fake_args = SimpleNamespace(device="meta") with get_context().override_server_args():
with patch.object(tbo, "get_server_args", lambda: fake_args): with get_device().override(device="meta"):
eager = ( eager = TboForwardBatchPreparer.compute_tbo_children_num_token_non_padded_raw(
TboForwardBatchPreparer.compute_tbo_children_num_token_non_padded_raw(
tbo_split_token_index=3, num_token_non_padded=8 tbo_split_token_index=3, num_token_non_padded=8
) )
)
plugin = TboCudaGraphRunnerPlugin() plugin = TboCudaGraphRunnerPlugin()
self.assertEqual( self.assertEqual(
@@ -82,13 +63,11 @@ class TestTboCudaGraphNumTokenDevice(CustomTestCase):
def test_eager_split_values(self): def test_eager_split_values(self):
# value_a = min(split, n); value_b = max(0, n - split). Computed on CPU # value_a = min(split, n); value_b = max(0, n - split). Computed on CPU
# so the values are materializable. # so the values are materializable.
fake_args = SimpleNamespace(device="cpu") with get_context().override_server_args():
with patch.object(tbo, "get_server_args", lambda: fake_args): with get_device().override(device="cpu"):
eager = ( eager = TboForwardBatchPreparer.compute_tbo_children_num_token_non_padded_raw(
TboForwardBatchPreparer.compute_tbo_children_num_token_non_padded_raw(
tbo_split_token_index=3, num_token_non_padded=8 tbo_split_token_index=3, num_token_non_padded=8
) )
)
self.assertEqual(eager.dtype, torch.int32) self.assertEqual(eager.dtype, torch.int32)
self.assertEqual(eager.tolist(), [3, 5]) self.assertEqual(eager.tolist(), [3, 5])
@@ -6,36 +6,18 @@ crashed TBO cuda-graph capture until reset. CPU-only.
""" """
import unittest import unittest
from types import SimpleNamespace
from unittest.mock import patch
import torch import torch
import sglang.srt.batch_overlap.two_batch_overlap as tbo
from sglang.srt.batch_overlap.two_batch_overlap import TboForwardBatchPreparer from sglang.srt.batch_overlap.two_batch_overlap import TboForwardBatchPreparer
from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode
from sglang.srt.runtime_context import get_parallel from sglang.srt.runtime_context import get_context, get_parallel
from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase from sglang.test.test_utils import CustomTestCase
register_cpu_ci(est_time=5, suite="base-a-test-cpu") register_cpu_ci(est_time=5, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
def _make_target_verify_batch(bs: int) -> ForwardBatch: def _make_target_verify_batch(bs: int) -> ForwardBatch:
return ForwardBatch( return ForwardBatch(
forward_mode=ForwardMode.TARGET_VERIFY, forward_mode=ForwardMode.TARGET_VERIFY,
@@ -52,10 +34,11 @@ def _make_target_verify_batch(bs: int) -> ForwardBatch:
def _filter(batch: ForwardBatch, *, lo: int, hi: int) -> ForwardBatch: def _filter(batch: ForwardBatch, *, lo: int, hi: int) -> ForwardBatch:
fake_args = SimpleNamespace(moe_dense_tp_size=None, attention_backend="fa3") # filter_batch reads attention_backend (get_server_args) and
with get_parallel().override(attn_tp_size=1), patch.object( # moe_dense_tp_size (get_parallel) from the published config.
tbo, "get_server_args", lambda: fake_args with get_context().override_server_args(
): attention_backend="fa3", moe_dense_tp_size=None
), get_parallel().override(attn_tp_size=1):
return TboForwardBatchPreparer.filter_batch( return TboForwardBatchPreparer.filter_batch(
batch, batch,
start_token_index=lo, start_token_index=lo,
@@ -30,20 +30,6 @@ from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(2.0, "base-a-test-cpu") register_cpu_ci(2.0, "base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
register_cpu_ci(est_time=7, suite="base-c-test-cpu") register_cpu_ci(est_time=7, suite="base-c-test-cpu")
@@ -5,30 +5,26 @@ from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=5, suite="base-a-test-cpu") register_cpu_ci(est_time=5, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
import unittest import unittest
from unittest.mock import MagicMock, call, patch from unittest.mock import MagicMock, call, patch
from sglang.srt.runtime_context import get_context
from sglang.test.test_utils import CustomTestCase from sglang.test.test_utils import CustomTestCase
class TestRegisterToBootstrap(CustomTestCase): class TestRegisterToBootstrap(CustomTestCase):
"""Tests for CommonKVManager.register_to_bootstrap retry/backoff behavior.""" """Tests for CommonKVManager.register_to_bootstrap retry/backoff behavior."""
def setUp(self):
# register_to_bootstrap reads get_parallel().load_balance_method /
# .enable_dsa_cache_layer_split and get_serving().port from the
# published config.
override = get_context().override_server_args(
load_balance_method="follow_bootstrap_room", port=30000
)
override.install()
self.addCleanup(override.restore)
@patch("sglang.srt.disaggregation.common.conn.time") @patch("sglang.srt.disaggregation.common.conn.time")
@patch("sglang.srt.disaggregation.common.conn.requests.put") @patch("sglang.srt.disaggregation.common.conn.requests.put")
def test_succeeds_on_first_attempt(self, mock_put, mock_time): def test_succeeds_on_first_attempt(self, mock_put, mock_time):
@@ -282,11 +278,8 @@ class TestRegisterToBootstrap(CustomTestCase):
mgr.kv_args = MagicMock() mgr.kv_args = MagicMock()
mgr.kv_args.page_size = 16 mgr.kv_args.page_size = 16
# Resolved per-runner value threaded through KVArgs (the payload field).
mgr.server_args = MagicMock() mgr.kv_cache_dtype_str = "auto"
mgr.server_args.kv_cache_dtype = "auto"
mgr.server_args.load_balance_method = "follow_bootstrap_room"
mgr.server_args.port = 30000
return mgr return mgr
@@ -21,25 +21,12 @@ import unittest
from types import SimpleNamespace from types import SimpleNamespace
from unittest import mock from unittest import mock
from sglang.srt.runtime_context import get_context
from sglang.test.ci.ci_register import register_cpu_ci, register_mlx_ci from sglang.test.ci.ci_register import register_cpu_ci, register_mlx_ci
from sglang.test.test_utils import CustomTestCase from sglang.test.test_utils import CustomTestCase
register_cpu_ci(est_time=1, suite="base-a-test-cpu") register_cpu_ci(est_time=1, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
register_mlx_ci(est_time=1, suite="stage-a-unit-test-mlx") register_mlx_ci(est_time=1, suite="stage-a-unit-test-mlx")
@@ -69,9 +56,19 @@ def _arch(hybrid):
) )
def _published(stub):
"""Publish the config the resolver reads (get_schedule() /
get_memory())."""
return get_context().override_server_args(
max_running_requests=stub._max_running_requests,
max_mamba_cache_size=stub._max_mamba_cache_size,
disable_radix_cache=stub._disable_radix_cache,
)
def _resolve(stub, hybrid=False): def _resolve(stub, hybrid=False):
"""Run the resolver with the model architecture patched (see _arch).""" """Run the resolver with the model architecture patched (see _arch)."""
with _arch(hybrid): with _arch(hybrid), _published(stub):
return stub._resolve_max_running_requests() return stub._resolve_max_running_requests()
@@ -85,13 +82,11 @@ def _stub(
"""A stub carrying only what _resolve_max_running_requests reads.""" """A stub carrying only what _resolve_max_running_requests reads."""
stub = MlxModelRunnerStub.__new__(MlxModelRunnerStub) stub = MlxModelRunnerStub.__new__(MlxModelRunnerStub)
stub.model_config = SimpleNamespace() # only handed to the patched _arch fn stub.model_config = SimpleNamespace() # only handed to the patched _arch fn
stub.server_args = SimpleNamespace( stub._max_running_requests = max_running_requests
max_running_requests=max_running_requests, stub._max_mamba_cache_size = max_mamba_cache_size
max_mamba_cache_size=max_mamba_cache_size, stub._disable_radix_cache = disable_radix_cache
disable_radix_cache=disable_radix_cache,
)
stub.max_total_num_tokens = max_total_num_tokens stub.max_total_num_tokens = max_total_num_tokens
stub.dp_size = dp_size stub.ps = SimpleNamespace(attn_dp_size=dp_size)
return stub return stub
@@ -101,14 +96,14 @@ def _hybrid_stub_for_initialize(
"""A stub carrying what the real initialize() reads (hybrid path).""" """A stub carrying what the real initialize() reads (hybrid path)."""
stub = MlxModelRunnerStub.__new__(MlxModelRunnerStub) stub = MlxModelRunnerStub.__new__(MlxModelRunnerStub)
stub._mlx_pool_size = pool stub._mlx_pool_size = pool
stub.dp_size = 1 stub.ps = SimpleNamespace(attn_dp_size=1)
stub.device = "cpu" # read by init_ngram_embedding_manager stub.device = "cpu" # read by init_ngram_embedding_manager
stub.server_args = SimpleNamespace( # Evaluated as a call argument in init_ngram_embedding_manager before
enable_memory_saver=False, # the use_ngram_embedding short-circuit; never read.
max_running_requests=max_running_requests, stub.server_args = SimpleNamespace()
max_mamba_cache_size=max_mamba_cache_size, stub._max_running_requests = max_running_requests
disable_radix_cache=disable_radix_cache, stub._max_mamba_cache_size = max_mamba_cache_size
) stub._disable_radix_cache = disable_radix_cache
stub.model_config = SimpleNamespace( stub.model_config = SimpleNamespace(
is_hybrid_swa=False, is_hybrid_swa=False,
sliding_window_size=None, sliding_window_size=None,
@@ -220,7 +215,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
stub = _hybrid_stub_for_initialize( stub = _hybrid_stub_for_initialize(
max_running_requests=8, max_mamba_cache_size=2 * RATIO max_running_requests=8, max_mamba_cache_size=2 * RATIO
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
self.assertEqual(stub.max_running_requests, 2) self.assertEqual(stub.max_running_requests, 2)
pool = stub.req_to_token_pool pool = stub.req_to_token_pool
@@ -233,7 +228,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
stub = _hybrid_stub_for_initialize( stub = _hybrid_stub_for_initialize(
max_running_requests=4, max_mamba_cache_size=2 max_running_requests=4, max_mamba_cache_size=2
) )
with _arch(hybrid=True), self.assertRaisesRegex( with _arch(hybrid=True), _published(stub), self.assertRaisesRegex(
RuntimeError, "max_mamba_cache_size" RuntimeError, "max_mamba_cache_size"
): ):
stub.initialize() stub.initialize()
@@ -247,7 +242,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
max_mamba_cache_size=8, max_mamba_cache_size=8,
disable_radix_cache=True, disable_radix_cache=True,
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
self.assertEqual(stub.max_running_requests, 8) self.assertEqual(stub.max_running_requests, 8)
pool = stub.req_to_token_pool pool = stub.req_to_token_pool
@@ -262,7 +257,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
max_mamba_cache_size=None, max_mamba_cache_size=None,
disable_radix_cache=True, disable_radix_cache=True,
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
self.assertEqual(stub.max_running_requests, 3) self.assertEqual(stub.max_running_requests, 3)
self.assertEqual(stub.req_to_token_pool.auxiliary_state_pool.size, 3) self.assertEqual(stub.req_to_token_pool.auxiliary_state_pool.size, 3)
@@ -282,7 +277,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
max_mamba_cache_size=2, max_mamba_cache_size=2,
disable_radix_cache=True, disable_radix_cache=True,
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
pool = stub.req_to_token_pool pool = stub.req_to_token_pool
aux_capacity = pool.auxiliary_state_pool.available_size() aux_capacity = pool.auxiliary_state_pool.available_size()
@@ -305,7 +300,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
max_mamba_cache_size=2 * RATIO, max_mamba_cache_size=2 * RATIO,
disable_radix_cache=False, disable_radix_cache=False,
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
pool = stub.req_to_token_pool pool = stub.req_to_token_pool
free_before = pool.auxiliary_state_pool.available_size() free_before = pool.auxiliary_state_pool.available_size()
@@ -322,7 +317,7 @@ class TestMlxHybridInitializeAllocation(CustomTestCase):
stub = _hybrid_stub_for_initialize( stub = _hybrid_stub_for_initialize(
max_running_requests=2, max_mamba_cache_size=None max_running_requests=2, max_mamba_cache_size=None
) )
with _arch(hybrid=True): with _arch(hybrid=True), _published(stub):
stub.initialize() stub.initialize()
self.assertEqual(stub.max_running_requests, 2) self.assertEqual(stub.max_running_requests, 2)
self.assertEqual(stub.req_to_token_pool.auxiliary_state_pool.size, 2 * RATIO) self.assertEqual(stub.req_to_token_pool.auxiliary_state_pool.size, 2 * RATIO)
@@ -1,28 +1,13 @@
import json import json
from pathlib import Path from pathlib import Path
from types import SimpleNamespace
from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=3, suite="base-a-test-cpu") register_cpu_ci(est_time=3, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
from sglang.srt.layers.moe.moe_runner.triton_utils import fused_moe_triton_config from sglang.srt.layers.moe.moe_runner.triton_utils import fused_moe_triton_config
from sglang.srt.runtime_context import get_context
def test_h200_bf16_config_is_available_for_current_triton_runtime(): def test_h200_bf16_config_is_available_for_current_triton_runtime():
@@ -47,11 +32,6 @@ def test_down_moe_reuses_tuned_up_config_when_separate_config_is_absent(
monkeypatch.setenv("SGLANG_MOE_CONFIG_DIR", str(tmp_path)) monkeypatch.setenv("SGLANG_MOE_CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(fused_moe_triton_config.triton, "__version__", "3.6.0") monkeypatch.setattr(fused_moe_triton_config.triton, "__version__", "3.6.0")
monkeypatch.setattr(
fused_moe_triton_config,
"get_server_args",
lambda: SimpleNamespace(enable_deterministic_inference=False),
)
monkeypatch.setattr( monkeypatch.setattr(
fused_moe_triton_config, fused_moe_triton_config,
"get_config_file_name", "get_config_file_name",
@@ -60,6 +40,8 @@ def test_down_moe_reuses_tuned_up_config_when_separate_config_is_absent(
fused_moe_triton_config.get_moe_configs.cache_clear() fused_moe_triton_config.get_moe_configs.cache_clear()
try: try:
# get_moe_configs reads get_exec().deterministic.
with get_context().override_server_args(enable_deterministic_inference=False):
assert fused_moe_triton_config.get_moe_configs( assert fused_moe_triton_config.get_moe_configs(
32, 768, None, down_moe=True 32, 768, None, down_moe=True
) == {128: {"BLOCK_SIZE_M": 64}} ) == {128: {"BLOCK_SIZE_M": 64}}
@@ -11,21 +11,6 @@ register_cuda_ci(est_time=9, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=1, suite="stage-b-test-1-gpu-small-amd") register_amd_ci(est_time=1, suite="stage-b-test-1-gpu-small-amd")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class TestMmProcessConfigValidation(unittest.TestCase): class TestMmProcessConfigValidation(unittest.TestCase):
"""Server-args validation for mm_process_config.""" """Server-args validation for mm_process_config."""
@@ -15,27 +15,13 @@ from sglang.srt.disaggregation.decode import ( # noqa: E402
from sglang.srt.disaggregation.utils import DisaggregationMode # noqa: E402 from sglang.srt.disaggregation.utils import DisaggregationMode # noqa: E402
from sglang.srt.managers.schedule_batch import FINISH_ABORT, Req # noqa: E402 from sglang.srt.managers.schedule_batch import FINISH_ABORT, Req # noqa: E402
from sglang.srt.managers.scheduler import Scheduler # noqa: E402 from sglang.srt.managers.scheduler import Scheduler # noqa: E402
from sglang.srt.runtime_context import get_context # noqa: E402
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
register_cuda_ci(est_time=5, stage="base-b", runner_config="1-gpu-small") register_cuda_ci(est_time=5, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=5, suite="stage-b-test-1-gpu-small-amd") register_amd_ci(est_time=5, suite="stage-b-test-1-gpu-small-amd")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class TestDisaggregationPriorityQueueing(unittest.TestCase): class TestDisaggregationPriorityQueueing(unittest.TestCase):
def _new_scheduler(self, disaggregation_mode: DisaggregationMode) -> Scheduler: def _new_scheduler(self, disaggregation_mode: DisaggregationMode) -> Scheduler:
scheduler = Scheduler.__new__(Scheduler) scheduler = Scheduler.__new__(Scheduler)
@@ -442,9 +428,8 @@ class TestDecodePrebuiltPriority(unittest.TestCase):
scheduler.enable_overlap = False scheduler.enable_overlap = False
scheduler.spec_algorithm = MagicMock() scheduler.spec_algorithm = MagicMock()
scheduler.max_running_requests = 1 scheduler.max_running_requests = 1
scheduler.server_args = SimpleNamespace( # Passed whole into the (mocked) batch's process_prebuilt; never read.
disaggregation_decode_enable_radix_cache=False scheduler.server_args = SimpleNamespace()
)
scheduler.future_map = MagicMock() scheduler.future_map = MagicMock()
scheduler.policy = MagicMock() scheduler.policy = MagicMock()
scheduler.policy.calc_priority.side_effect = lambda waiting_queue, _: ( scheduler.policy.calc_priority.side_effect = lambda waiting_queue, _: (
@@ -452,10 +437,14 @@ class TestDecodePrebuiltPriority(unittest.TestCase):
) )
new_batch = MagicMock() new_batch = MagicMock()
# get_new_prebuilt_batch reads the published disagg config
# (disaggregation_decode_enable_radix_cache).
with patch( with patch(
"sglang.srt.disaggregation.decode.ScheduleBatch.init_new", "sglang.srt.disaggregation.decode.ScheduleBatch.init_new",
return_value=new_batch, return_value=new_batch,
) as init_new: ) as init_new, get_context().override_server_args(
disaggregation_decode_enable_radix_cache=False
):
ret = SchedulerDisaggregationDecodeMixin.get_new_prebuilt_batch( ret = SchedulerDisaggregationDecodeMixin.get_new_prebuilt_batch(
scheduler, scheduler.running_batch scheduler, scheduler.running_batch
) )
@@ -30,21 +30,6 @@ from sglang.srt.observability.req_time_stats import APIServerReqTimeStats
register_cpu_ci(est_time=15, suite="base-a-test-cpu") register_cpu_ci(est_time=15, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
_NOT_FINISHED = object() # Sentinel: request has not finished yet _NOT_FINISHED = object() # Sentinel: request has not finished yet
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -15,21 +15,6 @@ from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=5, suite="base-a-test-cpu") register_cpu_ci(est_time=5, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class _FakeAllocator: class _FakeAllocator:
def __init__(self, base=1000, page_size=1): def __init__(self, base=1000, page_size=1):
self.base = base self.base = base
@@ -81,7 +81,7 @@ from sglang.srt.mem_cache.unified_radix_cache import (
_OngoingPrefetch, _OngoingPrefetch,
_OngoingWriteThrough, _OngoingWriteThrough,
) )
from sglang.srt.runtime_context import get_server_args from sglang.srt.runtime_context import get_server_args, get_serving
from sglang.srt.sampling.sampling_params import SamplingParams from sglang.srt.sampling.sampling_params import SamplingParams
from sglang.srt.server_args import ( from sglang.srt.server_args import (
ServerArgs, ServerArgs,
@@ -95,21 +95,6 @@ register_cuda_ci(est_time=16, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=16, suite="stage-b-test-1-gpu-small-amd") register_amd_ci(est_time=16, suite="stage-b-test-1-gpu-small-amd")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
@dataclass(frozen=True) @dataclass(frozen=True)
class CacheConfig: class CacheConfig:
# Tree # Tree
@@ -947,15 +932,13 @@ class UnifiedRadixCacheSuite:
req.mamba_last_track_seqlen = kv_len req.mamba_last_track_seqlen = kv_len
req.reasoning_tokens = 1 req.reasoning_tokens = 1
get_server_args().strip_thinking_cache = True # cache_finished_req reads get_serving().strip_thinking_cache
try: with get_serving().override(strip_thinking_cache=True):
avail_before = allocator.available_size() avail_before = allocator.available_size()
cache.cache_finished_req( cache.cache_finished_req(
req, is_insert=True, kv_len_to_handle=req.effective_kv_committed_len() req, is_insert=True, kv_len_to_handle=req.effective_kv_committed_len()
) )
start_p, end_p = req.effective_kv_committed_len(), req.kv.kv_allocated_len start_p, end_p = req.effective_kv_committed_len(), req.kv.kv_allocated_len
finally:
get_server_args().strip_thinking_cache = False
if ps > 1: if ps > 1:
start_p = ((start_p + ps - 1) // ps) * ps start_p = ((start_p + ps - 1) // ps) * ps
if start_p < end_p: if start_p < end_p:
@@ -5740,19 +5723,17 @@ class TestResumableInsertWalk(_InsertWalkSuite):
self._insert(cache, allocator, req_to_token_pool, [1, 2, 3, 4]) self._insert(cache, allocator, req_to_token_pool, [1, 2, 3, 4])
top = next(iter(cache.root_node.children.values())) top = next(iter(cache.root_node.children.values()))
# A storage-prefetch completion host-inserts a backuped node below the # Break backup continuity: a backuped (then device-evicted) middle
# still-unbacked top, legitimately breaking backup continuity. # below the still-unbacked top. insert_host refills below an
host_indices = cache.cache_controller.mem_pool_host.alloc(8) # un-backed-up node are dropped under write-through
host_result = cache.tree_core.insert_host( # (host_insert_dropped), so the state is built through an explicit
cache.root_node.id, # backup + device eviction.
RadixKey(array("q", list(range(1, 9)))), self._insert(cache, allocator, req_to_token_pool, list(range(1, 9)))
host_indices,
[f"h{i}" for i in range(8)],
)
cache.cache_controller.mem_pool_host.free(
host_indices[: host_result.prefix_len]
)
middle = next(iter(top.children.values())) middle = next(iter(top.children.values()))
self.assertGreater(_write_backup(cache, middle, write_back=True), 0)
cache.writing_check(write_back=True)
cache.evict(EvictParams(num_tokens=4))
self.assertTrue(middle.evicted)
self.assertTrue(middle.backuped) self.assertTrue(middle.backuped)
self.assertFalse(top.backuped) self.assertFalse(top.backuped)
@@ -17,21 +17,6 @@ from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=10, suite="base-a-test-cpu") register_cpu_ci(est_time=10, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
@contextlib.contextmanager @contextlib.contextmanager
def mock_cpu_env(kv_size=2, tp_size=1, swa_eviction_interval=4): def mock_cpu_env(kv_size=2, tp_size=1, swa_eviction_interval=4):
"""Mock GPU-dependent functions for CPU-only testing. """Mock GPU-dependent functions for CPU-only testing.
@@ -116,10 +101,12 @@ def _make_model_runner(
mc.hf_config = SimpleNamespace(architectures=["LlamaForCausalLM"]) mc.hf_config = SimpleNamespace(architectures=["LlamaForCausalLM"])
mc.hf_config.get_text_config = lambda: mc.hf_config mc.hf_config.get_text_config = lambda: mc.hf_config
mc.linear_attn_registry_result = None mc.linear_attn_registry_result = None
mc.context_len = 8192
mr.model_config = mc mr.model_config = mc
mr.kv_cache_dtype = "fake_bf16" mr.kv_cache_dtype = "fake_bf16"
sa = SimpleNamespace() sa = SimpleNamespace()
sa.max_total_tokens = None
sa.swa_full_tokens_ratio = swa_full_tokens_ratio sa.swa_full_tokens_ratio = swa_full_tokens_ratio
sa.page_size = page_size sa.page_size = page_size
sa.disable_radix_cache = disable_radix_cache sa.disable_radix_cache = disable_radix_cache
@@ -2,40 +2,16 @@ import unittest
from types import SimpleNamespace from types import SimpleNamespace
from sglang.srt.models.deepseek_v4 import DeepseekV4ForCausalLM from sglang.srt.models.deepseek_v4 import DeepseekV4ForCausalLM
from sglang.srt.runtime_context import get_context, reset_context from sglang.srt.runtime_context import get_context, get_exec
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=4, suite="base-a-test-cpu") register_cpu_ci(est_time=4, suite="base-a-test-cpu")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
class TestDeepseekV4SharedExpertFusionPolicy(unittest.TestCase): class TestDeepseekV4SharedExpertFusionPolicy(unittest.TestCase):
"""The disable decision is a load-time resolution: it writes through to """The disable decision is a load-time resolution: it lands on the
the published config via declare_load_time_override.""" published config bag via declare_load_time_override (bag-only; the
ServerArgs instance stays pristine)."""
def setUp(self):
self._saved_server_args = get_context()._server_args
def tearDown(self):
if self._saved_server_args is None:
reset_context()
else:
get_context().set_server_args(self._saved_server_args)
def _make_model(self, n_shared_experts=1): def _make_model(self, n_shared_experts=1):
return SimpleNamespace( return SimpleNamespace(
@@ -43,29 +19,30 @@ class TestDeepseekV4SharedExpertFusionPolicy(unittest.TestCase):
) )
def _publish(self, enforce): def _publish(self, enforce):
server_args = ServerArgs(model_path="dummy") override = get_context().override_server_args(
server_args.enforce_shared_experts_fusion = enforce enforce_shared_experts_fusion=enforce
get_context().set_server_args(server_args) )
return server_args override.install()
self.addCleanup(override.restore)
def test_disables_shared_fusion_without_enforce(self): def test_disables_shared_fusion_without_enforce(self):
server_args = self._publish(enforce=False) self._publish(enforce=False)
model = self._make_model() model = self._make_model()
DeepseekV4ForCausalLM.determine_num_fused_shared_experts(model) DeepseekV4ForCausalLM.determine_num_fused_shared_experts(model)
self.assertEqual(model.num_fused_shared_experts, 0) self.assertEqual(model.num_fused_shared_experts, 0)
# post-init declaration writes through to the published config # post-init declaration lands on the published config bag
self.assertTrue(server_args.disable_shared_experts_fusion) self.assertTrue(get_exec().moe.disable_shared_experts_fusion)
def test_enables_shared_fusion_when_enforced(self): def test_enables_shared_fusion_when_enforced(self):
server_args = self._publish(enforce=True) self._publish(enforce=True)
model = self._make_model() model = self._make_model()
DeepseekV4ForCausalLM.determine_num_fused_shared_experts(model) DeepseekV4ForCausalLM.determine_num_fused_shared_experts(model)
self.assertEqual(model.num_fused_shared_experts, 1) self.assertEqual(model.num_fused_shared_experts, 1)
self.assertFalse(server_args.disable_shared_experts_fusion) self.assertFalse(get_exec().moe.disable_shared_experts_fusion)
if __name__ == "__main__": if __name__ == "__main__":
@@ -13,6 +13,7 @@ from unittest.mock import MagicMock, patch
import torch import torch
from sglang.srt.model_executor.forward_batch_info import ForwardMode from sglang.srt.model_executor.forward_batch_info import ForwardMode
from sglang.srt.runtime_context import get_context
from sglang.srt.speculative.adaptive_runtime_state import SpecRuntimeState from sglang.srt.speculative.adaptive_runtime_state import SpecRuntimeState
from sglang.srt.speculative.eagle_utils import organize_draft_results from sglang.srt.speculative.eagle_utils import organize_draft_results
from sglang.srt.speculative.eagle_worker_v2 import EagleDraftWorker, EAGLEWorkerV2 from sglang.srt.speculative.eagle_worker_v2 import EagleDraftWorker, EAGLEWorkerV2
@@ -26,20 +27,6 @@ from sglang.test.test_utils import CustomTestCase
register_cuda_ci(est_time=20, stage="base-b", runner_config="1-gpu-small") register_cuda_ci(est_time=20, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=20, stage="stage-b", runner_config="1-gpu-small-amd") register_amd_ci(est_time=20, stage="stage-b", runner_config="1-gpu-small-amd")
import pytest as _pytest_defer
_DEFER_REASON = (
"Temporarily skipped during the ServerArgs config-namespace migration; "
"re-enabled once the runtime-config accessor API stabilizes."
)
pytestmark = _pytest_defer.mark.skip(reason=_DEFER_REASON)
def setUpModule():
import unittest
raise unittest.SkipTest(_DEFER_REASON)
register_cpu_ci(est_time=20, suite="base-a-test-cpu") register_cpu_ci(est_time=20, suite="base-a-test-cpu")
@@ -109,6 +96,15 @@ def _make_backend_factory(decode_backend, draft_extend_backend, captured_kwargs=
class TestEagleWorkerV2Topk1FastPath(CustomTestCase): class TestEagleWorkerV2Topk1FastPath(CustomTestCase):
def setUp(self):
# _rebuild_topk1_chain_buffers sizes its preallocation from the
# published config: get_exec().graph.cuda_graph_config stays None on
# the dummy-boundary publish (no resolution), so
# get_schedule().max_running_requests alone sizes the buffers.
override = get_context().override_server_args(max_running_requests=8)
override.install()
self.addCleanup(override.restore)
def test_fast_path_matches_slow_path(self): def test_fast_path_matches_slow_path(self):
bs = 3 bs = 3
for num_steps in (1, 2, 3, 4): for num_steps in (1, 2, 3, 4):
@@ -145,6 +141,13 @@ class TestEagleWorkerV2Topk1FastPath(CustomTestCase):
class TestEagleWorkerV2BackendFallback(CustomTestCase): class TestEagleWorkerV2BackendFallback(CustomTestCase):
def setUp(self):
# The adaptive state-machine paths write live spec switches through
# get_context().override, which needs a published config.
override = get_context().override_server_args()
override.install()
self.addCleanup(override.restore)
def test_missing_seed_cuda_graph_fallback(self): def test_missing_seed_cuda_graph_fallback(self):
graph_result = ( graph_result = (
[], [],
@@ -1,53 +0,0 @@
"""Ratchet guard: config-namespace-migration test deferrals may only decrease.
A set of unit tests was module-skipped ("Temporarily skipped during the
ServerArgs config-namespace migration") because their fixtures inject config
in ways the namespace accessors cannot see; they are recovered together with
the reader migration. Nothing else enforces that list: without this pin a new
skip can ride in unnoticed, and the already-skipped files keep growing test
code nobody has ever run. The count is exact: un-skipping a file must lower
the baseline to lock in the recovery, and no new deferral may appear.
"""
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
import unittest
from pathlib import Path
from sglang.test.test_utils import CustomTestCase
# test/registered/unit/<this file> -> test/
_TEST_ROOT = Path(__file__).resolve().parents[2]
_MARKER = "config-namespace migration"
_BASELINE = 15
class TestMigrationDeferralRatchet(CustomTestCase):
def test_deferred_test_files_match_the_baseline(self):
deferred = sorted(
p.relative_to(_TEST_ROOT).as_posix()
for p in _TEST_ROOT.rglob("*.py")
if p.name != Path(__file__).name and _MARKER in p.read_text(errors="ignore")
)
count = len(deferred)
if count > _BASELINE:
self.fail(
f"deferred (module-skipped) migration tests grew: {count} > "
f"baseline {_BASELINE}: {deferred}. Do not add new deferrals — "
"seed the context (override_server_args / publish a real "
"ServerArgs) instead of skipping the module."
)
if count < _BASELINE:
self.fail(
f"deferred migration tests shrank: {count} < baseline "
f"{_BASELINE}. Lower the baseline in this file to lock in the "
"recovery."
)
if __name__ == "__main__":
unittest.main()