minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[misc] Remove unit test cases that fail the admission criteria (round 3) (#30713)

Browse Source
This commit is contained in:
Liangsheng Yin
2026-07-09 19:42:51 -07:00
committed by GitHub
parent 7b9b2e4798
commit 23390589f7
19 changed files with 0 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/unit/batch_invariant_ops/test_batch_invariant_ops.py
-15
View File
@@ -158,21 +158,6 @@ class TestBatchInvariantOps(CustomTestCase):
)
self._assert_batch_invariant_results(difflist, dtype, name)
def test_without_batch_invariant_mode(self):
"""
Test that without batch-invariant mode, results may differ.
This test demonstrates the difference batch-invariant mode makes.
"""
M, K, N = 32, 128, 1024
dtype = torch.float32
# Run without batch-invariant mode
with set_batch_invariant_mode(False):
difflist = self._run_multiple_iterations(
iters=5, M=M, K=K, N=N, dtype=dtype
)
print(f"Without batch-invariant mode, we get diffs: {difflist}")
def _test_bmm_batch_invariance(self, B, M, K, N, dtype):
"""
Test that BMM operations produce identical results for:
test/registered/unit/constrained/test_base_grammar_backend.py
-45
View File
@@ -58,14 +58,6 @@ class TestGrammarStats(unittest.TestCase):
class TestBaseGrammarObject(unittest.TestCase):
"""Test BaseGrammarObject base class."""
def test_is_terminated_default(self):
obj = BaseGrammarObject()
self.assertFalse(obj.is_terminated())
def test_maybe_init_reasoning_noop(self):
obj = BaseGrammarObject()
obj.maybe_init_reasoning(True) # Should not raise
class TestInvalidGrammarObject(unittest.TestCase):
"""Test InvalidGrammarObject."""
@@ -88,18 +80,6 @@ class TestBaseGrammarBackend(unittest.TestCase):
def tearDown(self):
self.backend.executor.shutdown(wait=True)
def test_set_and_get_cache(self):
obj = BaseGrammarObject()
key = ("json", '{"type": "object"}')
self.backend.set_cache(key, obj)
self.assertIn(key, self.backend.cache)
self.assertIs(self.backend.cache[key], obj)
def test_reset_clears_cache(self):
self.backend.set_cache(("json", "schema"), BaseGrammarObject())
self.backend.reset()
self.assertEqual(len(self.backend.cache), 0)
def test_cache_hit_returns_copy(self):
"""Cache hit should return a copy of the cached object."""
mock_copy = BaseGrammarObject()
@@ -133,19 +113,6 @@ class TestBaseGrammarBackend(unittest.TestCase):
value = result.result(timeout=5)
self.assertIsInstance(value, InvalidGrammarObject)
def test_all_dispatch_methods_unsupported(self):
"""All dispatch methods on base class return InvalidGrammarObject."""
cases = [
("dispatch_json", ("schema",)),
("dispatch_regex", ("[a-z]+",)),
("dispatch_ebnf", ("root ::= 'hello'",)),
("dispatch_structural_tag", ("{}",)),
]
for method_name, args in cases:
with self.subTest(method=method_name):
result = getattr(self.backend, method_name)(*args)
self.assertIsInstance(result, InvalidGrammarObject)
def test_dispatch_fallback_raises(self):
with self.assertRaises(ValueError):
self.backend.dispatch_fallback("unknown", "value")
@@ -246,11 +213,6 @@ class TestRegisterGrammarBackend(unittest.TestCase):
GRAMMAR_BACKEND_REGISTRY.clear()
GRAMMAR_BACKEND_REGISTRY.update(self._saved)
def test_register_and_use(self):
mock_init = MagicMock(return_value="custom_backend")
register_grammar_backend("my_backend", mock_init)
self.assertIn("my_backend", GRAMMAR_BACKEND_REGISTRY)
def test_overwrite_registration(self):
register_grammar_backend("dup", lambda *a: "first")
register_grammar_backend("dup", lambda *a: "second")
@@ -298,13 +260,6 @@ class TestCreateGrammarBackend(unittest.TestCase):
with self.assertRaises(ValueError):
create_grammar_backend(args, None, 32000)
def test_custom_registered_backend(self):
mock_backend = MagicMock()
register_grammar_backend("test_custom", lambda *a: mock_backend)
args = self._make_server_args("test_custom")
result = create_grammar_backend(args, "tok", 32000, {1, 2})
self.assertIs(result, mock_backend)
def test_custom_backend_receives_args(self):
received = {}
test/registered/unit/disaggregation/test_disaggregation_wire.py
-5
View File
@@ -75,11 +75,6 @@ class TestGroupConcurrentContiguous(unittest.TestCase):
([[10, 11], [20]], [[5, 6], [7]]),
)
def test_both_empty(self):
self.assertEqual(
group_concurrent_contiguous(self._arr([]), self._arr([])), ([], [])
)
def test_empty_src_nonempty_dst(self):
self.assertEqual(
group_concurrent_contiguous(self._arr([]), self._arr([1, 2])), ([], [])
test/registered/unit/entrypoints/openai/test_serving_completions.py
-15
View File
@@ -60,27 +60,12 @@ class ServingCompletionTestCase(unittest.TestCase):
self.fastapi_request = Mock(spec=Request)
# ---------- prompt-handling ----------
def test_single_string_prompt(self):
req = CompletionRequest(
model="x",
prompt="Hello world",
max_tokens=100,
session_id="session-1",
)
internal, _ = self.sc._convert_to_internal_request(req)
self.assertEqual(internal.text, "Hello world")
self.assertEqual(internal.session_id, "session-1")
def test_single_token_ids_prompt(self):
req = CompletionRequest(model="x", prompt=[1, 2, 3, 4], max_tokens=100)
internal, _ = self.sc._convert_to_internal_request(req)
self.assertEqual(internal.input_ids, [1, 2, 3, 4])
# ---------- echo-handling ----------
def test_echo_with_string_prompt_streaming(self):
req = CompletionRequest(model="x", prompt="Hello", max_tokens=1, echo=True)
self.assertEqual(self.sc._get_echo_text(req, 0), "Hello")
def test_echo_with_list_of_strings_streaming(self):
req = CompletionRequest(
model="x", prompt=["A", "B"], max_tokens=1, echo=True, n=1
test/registered/unit/eplb/test_compute_logical_to_rank_dispatch_physical_map.py
-9
View File
@@ -100,15 +100,6 @@ class TestComputeLogicalToRankDispatchPhysicalMap(CustomTestCase):
f"ep_rank={ep_rank} has out-of-range values",
)
def test_no_minus_one_in_output(self):
"""No -1 sentinel values remain in the output (all ranks are assigned)."""
for ep_rank in range(self.EP_SIZE):
result = self._call(ep_rank=ep_rank)
self.assertFalse(
torch.any(result == -1),
f"ep_rank={ep_rank} still has unassigned entries",
)
# ------------------------------------------------------------------ correctness
def test_gpu0_prefers_local_experts(self):
test/registered/unit/function_call/test_parallel_tool_calls.py
-25
View File
@@ -134,31 +134,6 @@ class TestParallelToolCalls(unittest.TestCase):
params2["filename"], "doc2", "Second tool filename should be doc2"
)
def test_simple_parallel_tool_calls(self):
"""
Test a simpler case of two parallel tool calls with array parameters.
This is a minimal test case that still tests the core functionality.
"""
chunks = [
"[\n",
' {"name": "search_docs", "parameters": {"title": ["a"]}},',
"\n",
' {"name": "search_docs", "parameters": {"title": ["b"]}}',
"]",
]
tool_calls = []
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
self._accumulate_tool_calls(tool_calls, result)
# Should parse both tools successfully
self.assertEqual(len(tool_calls), 2, "Should parse 2 tool calls")
self.assertEqual(tool_calls[0]["name"], "search_docs")
self.assertEqual(tool_calls[1]["name"], "search_docs")
if __name__ == "__main__":
unittest.main()
test/registered/unit/hardware_backend/mlx/test_mlx_pool_dtype.py
-12
View File
@@ -104,18 +104,6 @@ class TestPoolDtypeInference(CustomTestCase):
_, _, dtype = _runner_for(model)._get_attn_config()
self.assertEqual(dtype, mx.float32)
def test_pool_bytes_per_slot_halves_for_bf16_quantized_model(self):
# The practical effect: bytes/slot uses dtype.size, so bf16 halves
# the fp32 fallback and the auto-sized pool fits ~2x the tokens.
model = _tiny_qwen2_model()
model.set_dtype(mx.bfloat16)
nn.quantize(model, group_size=64, bits=4)
n_kv_heads, head_dim, dtype = _runner_for(model)._get_attn_config()
num_layers = 2
bytes_per_slot = 2 * num_layers * n_kv_heads * head_dim * dtype.size
fp32_bytes_per_slot = 2 * num_layers * n_kv_heads * head_dim * 4
self.assertEqual(bytes_per_slot * 2, fp32_bytes_per_slot)
if __name__ == "__main__":
unittest.main()
test/registered/unit/hardware_backend/mlx/test_runner_init_contract.py
-16
View File
@@ -79,22 +79,6 @@ class TestMlxRunnerInitContract(unittest.TestCase):
"longer passes (regression of #28660)."
)
def test_stub_initialize_requires_no_extra_args(self):
# Same guard from the other side: no parameter beyond ``self`` may be
# REQUIRED. A defaulted parameter would still bind, but a required one
# (the pre-#28660 ``pre_model_load_memory``) is the desync we forbid.
required = _required_params_beyond_self(MlxModelRunnerStub.initialize)
self.assertEqual(
required,
[],
msg=(
"MlxModelRunnerStub.initialize requires parameter(s) "
f"{required} that the base never passes (it calls "
"self.initialize()). This re-introduces the #28660 desync; base "
"initialize(self) is parameterless since #23862."
),
)
if __name__ == "__main__":
unittest.main()
test/registered/unit/hardware_backend/mlx/test_tp_worker_routing.py
-4
View File
@@ -200,10 +200,6 @@ class TestMlxExtendRouting(unittest.TestCase):
runner = self._run_sync([_FakeReq("r1")], [1], {"r1"}, None, ForwardMode.EXTEND)
self.assertEqual(runner.ops_for("r1"), ["extend"])
def test_sync_multi_token_continuation_routes_to_extend(self):
runner = self._run_sync([_FakeReq("r1")], [4], {"r1"}, None, ForwardMode.EXTEND)
self.assertEqual(runner.ops_for("r1"), ["extend"])
def test_sync_genuine_mixed_decode_routes_to_decode(self):
p, d = _FakeReq("p1"), _FakeReq("d1")
runner = self._run_sync([p, d], [4, 1], {"d1"}, [d], ForwardMode.MIXED)
test/registered/unit/layers/quantization/test_quark_utils.py
-4
View File
@@ -52,10 +52,6 @@ class TestE8M0ToF32(CustomTestCase):
# ---- Guardrails: pass on both buggy and fixed code ----------------------
def test_zero_exponent_is_one(self):
x = torch.tensor([127], dtype=torch.uint8)
self.assertEqual(e8m0_to_f32(x).item(), 1.0)
def test_shape_preserved(self):
x = torch.zeros((3, 4, 5), dtype=torch.uint8)
self.assertEqual(tuple(e8m0_to_f32(x).shape), (3, 4, 5))
test/registered/unit/layers/test_conv_layer.py
-4
View File
@@ -62,10 +62,6 @@ class TestConv2dLayer(unittest.TestCase):
layer = Conv2dLayer(4, 8, kernel_size=2, stride=2, groups=2)
self.assertFalse(layer.enable_linear)
def test_default_disables_linear(self):
layer = Conv2dLayer(3, 768, kernel_size=14, stride=14)
self.assertFalse(layer.enable_linear)
def test_dilation_disables_linear(self):
layer = Conv2dLayer(3, 64, kernel_size=3, stride=3, dilation=2)
self.assertFalse(layer.enable_linear)
test/registered/unit/layers/test_mamba_state_scatter_triton.py
-142
View File
@@ -3,7 +3,6 @@ from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
register_cuda_ci(est_time=7, stage="base-b", runner_config="1-gpu-small")
register_amd_ci(est_time=7, suite="stage-b-test-1-gpu-small-amd-mi35x")
import os
import unittest
import torch
@@ -19,19 +18,6 @@ except Exception as e: # pragma: no cover
_FUSED_IMPORT_ERROR = e
def _dtype_from_str(name: str) -> torch.dtype:
mapping = {
"bfloat16": torch.bfloat16,
"float16": torch.float16,
"float32": torch.float32,
}
if name not in mapping:
raise ValueError(
f"Unsupported dtype string {name!r}. Supported: {sorted(mapping.keys())}"
)
return mapping[name]
def _ref_scatter(dst, src, dst_indices, src_indices, step_indices):
"""Reference implementation using PyTorch advanced indexing."""
# dst: [L, C, E]
@@ -146,22 +132,6 @@ def _fused_update_like(
)
def _time_cuda_ms(fn, iters=50, warmup=10):
"""Measure average CUDA time (ms) using CUDA events."""
for _ in range(warmup):
fn()
torch.cuda.synchronize()
start = torch.cuda.Event(enable_timing=True)
end = torch.cuda.Event(enable_timing=True)
start.record()
for _ in range(iters):
fn()
end.record()
torch.cuda.synchronize()
return start.elapsed_time(end) / iters
class TestMambaStateScatterCorrectness(unittest.TestCase):
@unittest.skipUnless(torch.cuda.is_available(), "CUDA is required for this test.")
def test_fused_matches_reference(self):
@@ -243,117 +213,5 @@ class TestMambaStateScatterCorrectness(unittest.TestCase):
torch.testing.assert_close(conv_fused, conv_ref)
class TestMambaStateScatterPerf(unittest.TestCase):
@unittest.skipUnless(torch.cuda.is_available(), "CUDA is required for this test.")
def test_perf_report_old_vs_fused(self):
"""Optional microbenchmark comparing baseline vs fused kernel.
Enable with: SGLANG_RUN_MAMBA_SCATTER_PERF_TEST=1
"""
if os.environ.get("SGLANG_RUN_MAMBA_SCATTER_PERF_TEST", "0") != "1":
self.skipTest("Set SGLANG_RUN_MAMBA_SCATTER_PERF_TEST=1 to run perf test.")
if fused_mamba_state_scatter_with_mask is None:
self.skipTest(
f"fused_mamba_state_scatter_with_mask import failed: {_FUSED_IMPORT_ERROR}"
)
torch.manual_seed(0)
device = torch.device("cuda")
# Parameterize sizes via env vars so we can match a real model more closely.
L = int(os.environ.get("SGLANG_MAMBA_SCATTER_LAYERS", "32"))
B = int(os.environ.get("SGLANG_MAMBA_SCATTER_BATCH", "48"))
C = int(os.environ.get("SGLANG_MAMBA_SCATTER_CACHE", "49"))
D = int(os.environ.get("SGLANG_MAMBA_SCATTER_DRAFT_TOKENS", "5"))
ssm_elems = int(os.environ.get("SGLANG_MAMBA_SCATTER_SSM_ELEMS", "4096"))
conv_elems = int(os.environ.get("SGLANG_MAMBA_SCATTER_CONV_ELEMS", "512"))
invalid_ratio = float(
os.environ.get("SGLANG_MAMBA_SCATTER_INVALID_RATIO", "0.0")
)
track_ratio = float(os.environ.get("SGLANG_MAMBA_SCATTER_TRACK_RATIO", "0.0"))
ssm_dtype = _dtype_from_str(
os.environ.get("SGLANG_MAMBA_SCATTER_SSM_DTYPE", "bfloat16")
)
conv_dtype = _dtype_from_str(
os.environ.get("SGLANG_MAMBA_SCATTER_CONV_DTYPE", "bfloat16")
)
# Use zeros for dst so each iteration overwrites the same memory.
ssm_states = torch.zeros((L, C, ssm_elems), device=device, dtype=ssm_dtype)
conv_states = torch.zeros((L, C, conv_elems), device=device, dtype=conv_dtype)
intermediate_ssm = torch.randn(
(L, B, D, ssm_elems), device=device, dtype=ssm_dtype
)
intermediate_conv = torch.randn(
(L, B, D, conv_elems), device=device, dtype=conv_dtype
)
state_indices_tensor = torch.randperm(C, device=device, dtype=torch.int64)[
:B
].to(torch.int32)
step_indices_raw = torch.randint(0, D, (B,), device=device, dtype=torch.int64)
if invalid_ratio > 0:
invalid = torch.rand((B,), device=device) < invalid_ratio
step_indices_raw[invalid] = -1
mamba_track_indices = None
mamba_steps_to_track = None
if track_ratio > 0:
mamba_track_indices = torch.randperm(C, device=device, dtype=torch.int64)[
:B
]
mamba_steps_to_track = torch.randint(
0, D, (B,), device=device, dtype=torch.int64
)
track_invalid = torch.rand((B,), device=device) >= track_ratio
mamba_steps_to_track[track_invalid] = -1
def ref_fn():
_ref_update_like(
ssm_states,
intermediate_ssm,
conv_states,
intermediate_conv,
state_indices_tensor=state_indices_tensor,
step_indices_raw=step_indices_raw,
mamba_track_indices=mamba_track_indices,
mamba_steps_to_track=mamba_steps_to_track,
)
def fused_fn():
_fused_update_like(
ssm_states,
intermediate_ssm,
conv_states,
intermediate_conv,
state_indices_tensor=state_indices_tensor,
step_indices_raw=step_indices_raw,
mamba_track_indices=mamba_track_indices,
mamba_steps_to_track=mamba_steps_to_track,
)
# Warm up JIT compilation for triton kernels (and caches for torch indexing)
ref_fn()
fused_fn()
torch.cuda.synchronize()
ref_ms = _time_cuda_ms(ref_fn)
fused_ms = _time_cuda_ms(fused_fn)
num_valid = int((step_indices_raw >= 0).sum().item())
ratio = fused_ms / ref_ms if ref_ms > 0 else float("inf")
speedup = ref_ms / fused_ms if fused_ms > 0 else float("inf")
# Print a concise report
print(
"\n[MambaStateScatterPerf]\n"
f" shapes: L={L} B={B} C={C} D={D} ssm_elems={ssm_elems} conv_elems={conv_elems}\n"
f" dtypes: ssm={ssm_dtype} conv={conv_dtype}\n"
f" valid: {num_valid}/{B} invalid_ratio={invalid_ratio} track_ratio={track_ratio}\n"
f" ref_total_ms (baseline): {ref_ms:.4f}\n"
f" fused_total_ms: {fused_ms:.4f} (ratio={ratio:.3f}x, speedup={speedup:.2f}x)\n"
)
if __name__ == "__main__": # pragma: no cover
unittest.main()
test/registered/unit/layers/test_pooler_score_and_pool.py
-11
View File
@@ -100,17 +100,6 @@ class TestScoreAndPool(CustomTestCase):
self.assertEqual(out.embeddings[0].shape, (2, self.num_labels))
self.assertEqual(out.embeddings[1].shape, (1, self.num_labels))
def test_no_delimiter_indices_falls_back(self):
"""multi_item_delimiter_indices=None -> single-item fallback."""
input_ids = torch.tensor([0, 1, 2, 3, 4, 5, 6, 7])
hidden = torch.randn(8, self.hidden_dim)
fb = _make_forward_batch(extend_seq_lens=[5, 3])
out = score_and_pool(self.score_head, self.pooler, hidden, fb, input_ids)
self.assertIsInstance(out.embeddings, torch.Tensor)
self.assertEqual(out.embeddings.shape, (2, self.num_labels))
def test_mis_extracts_positions_before_delimiter(self):
"""Verify MIS picks hidden states at index (delimiter_position - 1)."""
# Delimiters at indices 2 and 5 -> extract hidden at indices 1 and 4
test/registered/unit/managers/test_profile_merger_http_api.py
-105
View File
@@ -1,4 +1,3 @@
import json
import unittest
from sglang.srt.managers.io_struct import ProfileReq
@@ -15,35 +14,6 @@ register_cpu_ci(est_time=8, suite="base-c-test-cpu")
class TestProfileMergerHTTPAPI(CustomTestCase):
def test_profile_req_merge_profiles_json_serialization(self):
# Test with merge_profiles=True
req = ProfileReq(
output_dir="/tmp/test",
num_steps=5,
activities=["CPU", "GPU"],
profile_by_stage=True,
merge_profiles=True,
)
# Convert to dict (as would happen in HTTP request)
req_dict = {
"output_dir": req.output_dir,
"num_steps": req.num_steps,
"activities": req.activities,
"profile_by_stage": req.profile_by_stage,
"merge_profiles": req.merge_profiles,
}
# Test JSON serialization
json_str = json.dumps(req_dict)
parsed_data = json.loads(json_str)
self.assertTrue(parsed_data["merge_profiles"])
self.assertEqual(parsed_data["output_dir"], "/tmp/test")
self.assertEqual(parsed_data["num_steps"], 5)
self.assertEqual(parsed_data["activities"], ["CPU", "GPU"])
self.assertTrue(parsed_data["profile_by_stage"])
def test_profile_req_merge_profiles_json_deserialization(self):
# Test JSON data as would come from HTTP request
json_data = {
@@ -76,26 +46,6 @@ class TestProfileMergerHTTPAPI(CustomTestCase):
req = ProfileReq(**json_data)
self.assertFalse(req.merge_profiles)
def test_http_api_parameter_flow(self):
# Simulate HTTP request data
request_data = {
"output_dir": "/tmp/test",
"num_steps": 5,
"activities": ["CPU", "GPU"],
"profile_by_stage": True,
"merge_profiles": True,
}
# Create ProfileReq as HTTP server would
obj = ProfileReq(**request_data)
# Verify the parameter is set correctly
self.assertTrue(obj.merge_profiles)
self.assertEqual(obj.output_dir, "/tmp/test")
self.assertEqual(obj.num_steps, 5)
self.assertEqual(obj.activities, ["CPU", "GPU"])
self.assertTrue(obj.profile_by_stage)
def test_http_api_parameter_validation(self):
# Test with True
json_data = {"merge_profiles": True}
@@ -112,61 +62,6 @@ class TestProfileMergerHTTPAPI(CustomTestCase):
req = ProfileReq(**json_data)
self.assertEqual(req.merge_profiles, "true") # String, not boolean
def test_http_api_backward_compatibility(self):
# Test minimal request (no merge_profiles)
json_data = {}
req = ProfileReq(**json_data)
self.assertFalse(req.merge_profiles) # Should default to False
# Test with other parameters but no merge_profiles
json_data = {
"output_dir": "/tmp/test",
"num_steps": 5,
"activities": ["CPU", "GPU"],
}
req = ProfileReq(**json_data)
self.assertFalse(req.merge_profiles) # Should default to False
def test_http_api_parameter_combinations(self):
test_cases = [
{
"name": "minimal with merge_profiles",
"data": {"merge_profiles": True},
"expected_merge": True,
},
{
"name": "full parameters with merge_profiles=True",
"data": {
"output_dir": "/tmp/test",
"num_steps": 10,
"activities": ["CPU", "GPU", "MEM"],
"profile_by_stage": True,
"with_stack": True,
"record_shapes": True,
"merge_profiles": True,
},
"expected_merge": True,
},
{
"name": "full parameters with merge_profiles=False",
"data": {
"output_dir": "/tmp/test",
"num_steps": 10,
"activities": ["CPU", "GPU", "MEM"],
"profile_by_stage": False,
"with_stack": False,
"record_shapes": False,
"merge_profiles": False,
},
"expected_merge": False,
},
]
for test_case in test_cases:
with self.subTest(test_case["name"]):
req = ProfileReq(**test_case["data"])
self.assertEqual(req.merge_profiles, test_case["expected_merge"])
if __name__ == "__main__":
unittest.main()
test/registered/unit/managers/test_vocab_boundary_finish.py
-8
View File
@@ -45,14 +45,6 @@ class TestVocabBoundaryFinish(CustomTestCase):
# The offending slot is rewritten to the eos token.
self.assertEqual(req.output_ids[1], 2)
def test_token_above_vocab_size_is_out_of_bounds(self):
# A wildly large garbage id (typical of NaN sampling) is caught.
req = _make_req(
output_ids=[5, VOCAB_SIZE + 12345], eos_token_ids={2}, stop_token_ids=set()
)
self.assertTrue(req._check_vocab_boundary_finish([5, VOCAB_SIZE + 12345]))
self.assertEqual(req.finished_len, 2)
def test_negative_token_is_out_of_bounds(self):
# Negative ids also indicate corrupted sampling output.
req = _make_req(output_ids=[5, -1], eos_token_ids={2}, stop_token_ids=set())
test/registered/unit/observability/test_metrics_utils.py
-11
View File
@@ -128,17 +128,6 @@ class TestMetricsUtils(unittest.TestCase):
for value in result:
self.assertIsInstance(value, float)
def test_integration_tse_through_generate_buckets(self):
"""Test integration of TSE buckets through generate_buckets function."""
default_buckets = [1.0, 10.0, 100.0]
# Generate buckets using both methods
direct_result = two_sides_exponential_buckets(50.0, 1.5, 6)
indirect_result = generate_buckets(["tse", "50.0", "1.5", "6"], default_buckets)
# Results should be identical
self.assertEqual(direct_result, indirect_result)
if __name__ == "__main__":
unittest.main()
test/registered/unit/parser/test_reasoning_content_without_parser.py
-34
View File
@@ -27,21 +27,6 @@ class TestReasoningContentWithoutParser(CustomTestCase):
When reasoning_parser is None the block is skipped entirely.
"""
def test_no_parser_text_passthrough(self):
"""Without a parser, raw text with <think> tags passes through as-is."""
reasoning_parser = None
# Simulate serving_chat.py logic
reasoning_text = None
text = THINK_OUTPUT
if reasoning_parser:
parser = ReasoningParser(reasoning_parser)
reasoning_text, text = parser.parse_non_stream(text)
self.assertIsNone(reasoning_text)
self.assertIn("<think>", text)
self.assertIn("The answer is 4.", text)
def test_with_parser_separates_reasoning(self):
"""With a parser, reasoning content is correctly separated."""
for parser_name, output in [
@@ -57,25 +42,6 @@ class TestReasoningContentWithoutParser(CustomTestCase):
self.assertNotIn("<think>", reasoning_text)
self.assertIn("The answer is 4.", text)
def test_no_parser_streaming_passthrough(self):
"""Without a parser, streaming chunks pass through without reasoning separation."""
reasoning_parser = None
# Simulate serving_chat.py streaming logic
chunks = ["<think>\nLet me", " think.\n</think>\nThe answer", " is 4."]
all_text = ""
reasoning_text_seen = False
for chunk in chunks:
delta = chunk
if reasoning_parser:
# This block would separate reasoning in streaming
reasoning_text_seen = True
all_text += delta
self.assertFalse(reasoning_text_seen)
self.assertIn("<think>", all_text)
if __name__ == "__main__":
unittest.main()
test/registered/unit/test_runai_utils.py
-4
View File
@@ -1,7 +1,6 @@
import unittest
from pathlib import Path
from sglang.srt.configs.load_config import LoadFormat
from sglang.srt.utils.runai_utils import ObjectStorageModel, is_runai_obj_uri
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
@@ -50,9 +49,6 @@ class TestRunaiUtils(CustomTestCase):
path = ObjectStorageModel.get_path("s3://bucket/model/")
self.assertIn("model_streamer", path)
def test_load_format_enum(self):
self.assertEqual(LoadFormat.RUNAI_STREAMER.value, "runai_streamer")
if __name__ == "__main__":
unittest.main()
test/registered/unit/tokenizer/test_tiktoken_tokenizer.py
-31
View File
@@ -36,11 +36,6 @@ class TestConstants(CustomTestCase):
self.assertEqual(CONTROL_TOKEN_TEXTS[0], "<|control1|>")
self.assertEqual(CONTROL_TOKEN_TEXTS[-1], "<|control704|>")
def test_special_token_values(self):
self.assertEqual(PAD, "<|pad|>")
self.assertEqual(EOS, "<|eos|>")
self.assertEqual(SEP, "<|separator|>")
def test_default_special_tokens_contains_all(self):
self.assertIn(PAD, DEFAULT_SPECIAL_TOKENS)
self.assertIn(EOS, DEFAULT_SPECIAL_TOKENS)
@@ -62,14 +57,6 @@ class TestTiktokenProcessor(CustomTestCase):
self.addCleanup(tokenizer_patcher.stop)
self.processor = TiktokenProcessor(name="dummy")
def test_image_processor_returns_dict(self):
result = self.processor.image_processor("fake_image")
self.assertIsInstance(result, dict)
def test_image_processor_has_pixel_values_key(self):
result = self.processor.image_processor("fake_image")
self.assertIn("pixel_values", result)
def test_image_processor_wraps_image_in_list(self):
image = "fake_image_data"
result = self.processor.image_processor(image)
@@ -95,24 +82,6 @@ class TestTiktokenTokenizer(CustomTestCase):
"{% if add_generation_prompt %}assistant:{% endif %}"
)
def test_encode_delegates_to_tokenizer(self):
self.mock_tokenizer.encode.return_value = [1, 2, 3]
result = self.tok.encode("hello")
self.mock_tokenizer.encode.assert_called_once_with("hello")
self.assertEqual(result, [1, 2, 3])
def test_decode_delegates_to_tokenizer(self):
self.mock_tokenizer.decode.return_value = "hello"
result = self.tok.decode([1, 2, 3])
self.mock_tokenizer.decode.assert_called_once_with([1, 2, 3])
self.assertEqual(result, "hello")
def test_batch_decode_list_of_lists(self):
self.mock_tokenizer.decode_batch.return_value = ["hello", "world"]
result = self.tok.batch_decode([[1, 2], [3, 4]])
self.mock_tokenizer.decode_batch.assert_called_once_with([[1, 2], [3, 4]])
self.assertEqual(result, ["hello", "world"])
def test_batch_decode_flat_list_wraps_each(self):
self.mock_tokenizer.decode_batch.return_value = ["a", "b"]
self.tok.batch_decode([1, 2])