diff --git a/.github/workflows/rerun-test.yml b/.github/workflows/rerun-test.yml
index e5c8c98c5..658946b8c 100644
--- a/.github/workflows/rerun-test.yml
+++ b/.github/workflows/rerun-test.yml
@@ -138,6 +138,14 @@ jobs:
- name: Install uv
uses: astral-sh/setup-uv@v5
+ # Needed by setuptools-rust to build the bundled native gRPC extension
+ # (rust/sglang-grpc) when installing the main `sglang` wheel from source.
+ - name: Install protoc
+ run: sudo bash scripts/ci/utils/install_protoc.sh
+
+ - name: Install Rust toolchain
+ run: bash scripts/ci/utils/install_rustup.sh
+
- name: Install dependencies
timeout-minutes: 20
env:
diff --git a/python/sglang/jit_kernel/tests/test_dependency.py b/python/sglang/jit_kernel/tests/test_dependency.py
deleted file mode 100644
index 9fc981956..000000000
--- a/python/sglang/jit_kernel/tests/test_dependency.py
+++ /dev/null
@@ -1,13 +0,0 @@
-import pytest
-
-from sglang.jit_kernel.utils import _REGISTERED_DEPENDENCIES
-from sglang.test.ci.ci_register import register_cuda_ci
-
-register_cuda_ci(est_time=30, suite="stage-b-kernel-unit-1-gpu-large")
-register_cuda_ci(est_time=30, suite="nightly-kernel-1-gpu", nightly=True)
-
-
-@pytest.mark.parametrize("name", _REGISTERED_DEPENDENCIES.keys())
-def test_availability(name: str) -> None:
- # NOTE: the path resolution should not fail
- _REGISTERED_DEPENDENCIES[name]()
diff --git a/python/sglang/test/ci/ci_register.py b/python/sglang/test/ci/ci_register.py
index 385373510..014c408dd 100644
--- a/python/sglang/test/ci/ci_register.py
+++ b/python/sglang/test/ci/ci_register.py
@@ -2,7 +2,7 @@ import ast
import warnings
from dataclasses import dataclass
from enum import Enum, auto
-from typing import List, Optional
+from typing import List, Optional, Tuple
__all__ = [
"HWBackend",
@@ -87,6 +87,7 @@ class RegistryVisitor(ast.NodeVisitor):
def __init__(self, filename: str):
self.filename = filename
self.registries: list[CIRegistry] = []
+ self.has_main_entry: bool = False
def _constant_value(self, node: ast.AST) -> object:
if isinstance(node, ast.Constant):
@@ -180,25 +181,53 @@ class RegistryVisitor(ast.NodeVisitor):
disabled=disabled,
)
+ @staticmethod
+ def _is_main_block_with_call(stmt: ast.If) -> bool:
+ """True iff `stmt` is `if __name__ == "__main__":` with a body that
+ contains at least one call (i.e. actually runs something, not just
+ `pass`). This is what makes `python3 file.py` execute tests."""
+ test = stmt.test
+ if not isinstance(test, ast.Compare):
+ return False
+ if not (isinstance(test.left, ast.Name) and test.left.id == "__name__"):
+ return False
+ if len(test.ops) != 1 or not isinstance(test.ops[0], ast.Eq):
+ return False
+ if len(test.comparators) != 1:
+ return False
+ rhs = test.comparators[0]
+ if not (isinstance(rhs, ast.Constant) and rhs.value == "__main__"):
+ return False
+ for child in ast.walk(ast.Module(body=stmt.body, type_ignores=[])):
+ if isinstance(child, ast.Call):
+ return True
+ return False
+
def visit_Module(self, node):
for stmt in node.body:
- if not isinstance(stmt, ast.Expr) or not isinstance(stmt.value, ast.Call):
- continue
-
- cr = self._collect_ci_registry(stmt.value)
- if cr is not None:
- self.registries.append(cr)
+ if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Call):
+ cr = self._collect_ci_registry(stmt.value)
+ if cr is not None:
+ self.registries.append(cr)
+ elif isinstance(stmt, ast.If) and self._is_main_block_with_call(stmt):
+ self.has_main_entry = True
self.generic_visit(node)
-def ut_parse_one_file(filename: str) -> List[CIRegistry]:
+def ut_parse_one_file(filename: str) -> Tuple[List[CIRegistry], bool]:
+ """Parse a test file and return (registries, has_main_entry).
+
+ `has_main_entry` is True iff the file has `if __name__ == "__main__":`
+ with a call in its body -- required for `python3 file.py` to actually
+ run tests (the CI runner's invocation pattern).
+ """
with open(filename, "r") as f:
file_content = f.read()
tree = ast.parse(file_content, filename=filename)
visitor = RegistryVisitor(filename=filename)
visitor.visit(tree)
- return visitor.registries
+ return visitor.registries, visitor.has_main_entry
def auto_partition(files: List[CIRegistry], rank: int, size: int) -> List[CIRegistry]:
@@ -230,7 +259,7 @@ def auto_partition(files: List[CIRegistry], rank: int, size: int) -> List[CIRegi
def collect_tests(files: list[str], sanity_check: bool = True) -> List[CIRegistry]:
ci_tests = []
for file in files:
- registries = ut_parse_one_file(file)
+ registries, has_main_entry = ut_parse_one_file(file)
if len(registries) == 0:
msg = f"No CI registry found in {file}"
if sanity_check:
@@ -239,6 +268,20 @@ def collect_tests(files: list[str], sanity_check: bool = True) -> List[CIRegistr
warnings.warn(msg)
continue
+ # Every file with at least one enabled registry must have an
+ # executable `if __name__ == "__main__":` block; otherwise
+ # `python3 file.py -f` (how run_unittest_files invokes tests)
+ # silently exits and the file shows green without running.
+ has_enabled = any(r.disabled is None for r in registries)
+ if sanity_check and has_enabled and not has_main_entry:
+ raise ValueError(
+ f'{file}: missing `if __name__ == "__main__":` entry. '
+ f"Pytest-style tests in this file will silently skip under "
+ f"`python3 file.py -f`. Add `unittest.main()` (for "
+ f"unittest.TestCase) or `sys.exit(pytest.main([__file__, "
+ f'"-v"]))` (for pytest-style).'
+ )
+
ci_tests.extend(registries)
return ci_tests
diff --git a/scripts/ci/check_registered_tests.py b/scripts/ci/check_registered_tests.py
index 28459a4b8..8f4a910ed 100755
--- a/scripts/ci/check_registered_tests.py
+++ b/scripts/ci/check_registered_tests.py
@@ -34,7 +34,7 @@ def main() -> int:
errors = []
for f in files:
try:
- registries = ci_register.ut_parse_one_file(f)
+ registries, _has_main_entry = ci_register.ut_parse_one_file(f)
if len(registries) == 0:
errors.append(f)
except Exception:
diff --git a/scripts/ci/utils/ci_coverage_report.py b/scripts/ci/utils/ci_coverage_report.py
index 1dc6708d6..717b801fa 100755
--- a/scripts/ci/utils/ci_coverage_report.py
+++ b/scripts/ci/utils/ci_coverage_report.py
@@ -35,7 +35,7 @@ def collect_all_tests(registered_dir: str) -> list[CIRegistry]:
for file in sorted(files):
try:
- registries = ut_parse_one_file(file)
+ registries, _ = ut_parse_one_file(file)
all_tests.extend(registries)
except Exception as e:
print(f"Warning: Failed to parse {file}: {e}", file=sys.stderr)
diff --git a/test/registered/amd/test_zimage_turbo.py b/test/registered/amd/test_zimage_turbo.py
deleted file mode 100644
index 3e5d8c4ea..000000000
--- a/test/registered/amd/test_zimage_turbo.py
+++ /dev/null
@@ -1,150 +0,0 @@
-"""AMD nightly test for Z-Image-Turbo diffusion model (text-to-image)."""
-
-import io
-import logging
-import os
-
-import pytest
-
-from sglang.multimodal_gen.test.server.test_server_common import ( # noqa: F401
- DiffusionServerBase,
- diffusion_server,
-)
-from sglang.multimodal_gen.test.server.test_server_utils import (
- ServerContext,
- get_generate_fn,
-)
-from sglang.multimodal_gen.test.server.testcase_configs import (
- DiffusionSamplingParams,
- DiffusionServerArgs,
- DiffusionTestCase,
-)
-from sglang.test.ci.ci_register import register_amd_ci
-
-logger = logging.getLogger(__name__)
-
-register_amd_ci(est_time=1800, suite="nightly-amd-1-gpu-zimage-turbo", nightly=True)
-
-AMD_ZIMAGE_CASES = [
- DiffusionTestCase(
- "zimage_image_t2i",
- DiffusionServerArgs(model_path="Tongyi-MAI/Z-Image-Turbo", modality="image"),
- DiffusionSamplingParams(
- prompt="Doraemon is eating dorayaki",
- output_size="1024x1024",
- ),
- ),
-]
-
-CLIP_SCORE_THRESHOLD = 0.20
-
-
-ARTIFACT_DIR = os.environ.get(
- "SGLANG_DIFFUSION_ARTIFACT_DIR", "/tmp/diffusion-artifacts"
-)
-
-
-def _save_image_and_write_summary(
- case_id: str, prompt: str, image_bytes: bytes, clip_score: float | None = None
-):
- """Save generated image to artifact dir and write summary."""
- ext = "jpg" if image_bytes[:2] == b"\xff\xd8" else "png"
- os.makedirs(ARTIFACT_DIR, exist_ok=True)
- img_path = os.path.join(ARTIFACT_DIR, f"{case_id}.{ext}")
- with open(img_path, "wb") as f:
- f.write(image_bytes)
- logger.info("Saved image artifact: %s (%d bytes)", img_path, len(image_bytes))
-
- summary_file = os.environ.get("GITHUB_STEP_SUMMARY")
- if not summary_file:
- return
-
- clip_line = ""
- if clip_score is not None:
- status = "PASS" if clip_score >= CLIP_SCORE_THRESHOLD else "FAIL"
- clip_line = f"| CLIP Score | {clip_score:.4f} ({status}, threshold: {CLIP_SCORE_THRESHOLD}) |\n"
-
- md = (
- f"### Z-Image-Turbo — `{case_id}`\n\n"
- f"| | |\n|---|---|\n"
- f"| Prompt | {prompt} |\n"
- f"| Size | {len(image_bytes):,} bytes |\n"
- f"{clip_line}"
- f"| Artifact | `{case_id}.{ext}` (download from Artifacts section above) |\n\n"
- )
-
- with open(summary_file, "a") as f:
- f.write(md)
-
-
-def _compute_clip_score(image_bytes: bytes, prompt: str) -> float | None:
- """Compute CLIP cosine similarity between the image and prompt."""
- try:
- import torch
- from PIL import Image
- from transformers import CLIPModel, CLIPProcessor
-
- model_name = "openai/clip-vit-base-patch32"
- processor = CLIPProcessor.from_pretrained(model_name)
- model = CLIPModel.from_pretrained(model_name)
- model.eval()
-
- image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
- inputs = processor(text=[prompt], images=image, return_tensors="pt")
-
- with torch.no_grad():
- outputs = model(**inputs)
- score = outputs.logits_per_image.item() / 100.0
-
- logger.info("CLIP score for '%s': %.4f", prompt, score)
- return score
- except Exception as e:
- logger.warning("CLIP score computation failed: %s", e)
- return None
-
-
-class TestZImageTurboAMD(DiffusionServerBase):
- """AMD nightly test for Z-Image-Turbo text-to-image generation."""
-
- @classmethod
- def teardown_class(cls):
- try:
- super().teardown_class()
- except AttributeError:
- pass
-
- @pytest.fixture(params=AMD_ZIMAGE_CASES, ids=lambda c: c.id)
- def case(self, request) -> DiffusionTestCase:
- return request.param
-
- def test_diffusion_generation(
- self,
- case: DiffusionTestCase,
- diffusion_server: ServerContext,
- ):
- generate_fn = get_generate_fn(
- model_path=case.server_args.model_path,
- modality=case.server_args.modality,
- sampling_params=case.sampling_params,
- )
-
- perf_record, content = self.run_and_collect(
- diffusion_server, case.id, generate_fn
- )
-
- self._validate_and_record(case, perf_record)
- self._test_v1_models_endpoint(diffusion_server, case)
-
- prompt = case.sampling_params.prompt or ""
- clip_score = _compute_clip_score(content, prompt)
-
- if clip_score is not None:
- logger.info(
- "CLIP score: %.4f (threshold: %.2f)", clip_score, CLIP_SCORE_THRESHOLD
- )
- assert clip_score >= CLIP_SCORE_THRESHOLD, (
- f"CLIP score {clip_score:.4f} below threshold {CLIP_SCORE_THRESHOLD} "
- f"for prompt '{prompt}'"
- )
-
- _save_image_and_write_summary(case.id, prompt, content, clip_score)
diff --git a/test/registered/debug_utils/comparator/test_preset.py b/test/registered/debug_utils/comparator/test_preset.py
index e6b99df65..4ccb84647 100644
--- a/test/registered/debug_utils/comparator/test_preset.py
+++ b/test/registered/debug_utils/comparator/test_preset.py
@@ -42,3 +42,9 @@ class TestExpandPreset:
"""Unknown preset name raises ValueError."""
with pytest.raises(ValueError, match="Unknown value for --preset"):
expand_preset(["--preset", "nonexistent"], presets=PRESETS)
+
+
+if __name__ == "__main__":
+ import sys
+
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/debug_utils/source_patcher/test_code_patcher.py b/test/registered/debug_utils/source_patcher/test_code_patcher.py
index 7d791788d..2106ad6a1 100644
--- a/test/registered/debug_utils/source_patcher/test_code_patcher.py
+++ b/test/registered/debug_utils/source_patcher/test_code_patcher.py
@@ -254,3 +254,9 @@ class TestCodePatcher:
raise RuntimeError("test error")
assert obj.greet("world") == "hello world"
+
+
+if __name__ == "__main__":
+ import sys
+
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/debug_utils/source_patcher/test_dumper_integration.py b/test/registered/debug_utils/source_patcher/test_dumper_integration.py
index af9d8d077..2b8be98bf 100644
--- a/test/registered/debug_utils/source_patcher/test_dumper_integration.py
+++ b/test/registered/debug_utils/source_patcher/test_dumper_integration.py
@@ -55,3 +55,11 @@ class TestDumperApplySourcePatches:
cls.greet.__code__ = original_code
assert obj.greet("world") == "hello world"
+
+
+if __name__ == "__main__":
+ import sys
+
+ import pytest
+
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/debug_utils/source_patcher/test_source_editor.py b/test/registered/debug_utils/source_patcher/test_source_editor.py
index 560563dde..fd5334be4 100644
--- a/test/registered/debug_utils/source_patcher/test_source_editor.py
+++ b/test/registered/debug_utils/source_patcher/test_source_editor.py
@@ -290,3 +290,9 @@ class TestApplyEdits:
]
result = apply_edits(source=source, edits=edits)
assert result == ("def foo():\n" " x = 1\n" " y = 20\n" " return x\n")
+
+
+if __name__ == "__main__":
+ import sys
+
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/debug_utils/test_dump_comparator.py b/test/registered/debug_utils/test_dump_comparator.py
index 8b2d66d54..5d4945189 100644
--- a/test/registered/debug_utils/test_dump_comparator.py
+++ b/test/registered/debug_utils/test_dump_comparator.py
@@ -1,6 +1,3 @@
-from argparse import Namespace
-from pathlib import Path
-
import pytest
import torch
@@ -9,9 +6,7 @@ from sglang.srt.debug_utils.dump_comparator import (
_calc_rel_diff,
_compute_smaller_dtype,
_try_unify_shape,
- main,
)
-from sglang.srt.debug_utils.dumper import DumperConfig, _Dumper
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=30, suite="stage-a-test-cpu", nightly=True)
@@ -59,101 +54,7 @@ class TestComputeSmallerDtype:
assert _compute_smaller_dtype(torch.float32, torch.float32) is None
-# ----------------------------- Integration tests -----------------------------
+if __name__ == "__main__":
+ import sys
-
-def _make_dumper(directory: Path) -> _Dumper:
- return _Dumper(
- config=DumperConfig(
- enable=True,
- dir=str(directory),
- )
- )
-
-
-def _create_dumps(
- tmp_path: Path,
- tensor_names: list[str],
- *,
- baseline_names: list[str] | None = None,
-) -> tuple[Path, Path]:
- if baseline_names is None:
- baseline_names = tensor_names
-
- d_baseline: Path = tmp_path / "baseline"
- d_target: Path = tmp_path / "target"
- d_baseline.mkdir()
- d_target.mkdir()
-
- torch.manual_seed(42)
- baseline_tensor: torch.Tensor = torch.randn(10, 10)
- target_tensor: torch.Tensor = baseline_tensor + torch.randn(10, 10) * 0.01
-
- exp_paths: list[Path] = []
- for d, names, tensor in [
- (d_baseline, baseline_names, baseline_tensor),
- (d_target, tensor_names, target_tensor),
- ]:
- dumper: _Dumper = _make_dumper(d)
- for name in names:
- dumper.dump(name, tensor)
- dumper.step()
- exp_paths.append(d / dumper._config.exp_name)
-
- return exp_paths[0], exp_paths[1]
-
-
-def _make_args(
- baseline_path: Path,
- target_path: Path,
- *,
- filter_pattern: str | None = None,
-) -> Namespace:
- return Namespace(
- baseline_path=str(baseline_path),
- target_path=str(target_path),
- start_step=0,
- end_step=1000000,
- diff_threshold=1e-3,
- filter=filter_pattern,
- )
-
-
-class TestMainBasic:
- def test_matching_tensors(
- self, tmp_path: Path, capsys: pytest.CaptureFixture
- ) -> None:
- baseline_path, target_path = _create_dumps(tmp_path, ["tensor_a", "tensor_b"])
- args: Namespace = _make_args(baseline_path, target_path)
-
- main(args)
-
- captured: str = capsys.readouterr().out
- assert "✅" in captured
-
- def test_with_filter(self, tmp_path: Path, capsys: pytest.CaptureFixture) -> None:
- baseline_path, target_path = _create_dumps(tmp_path, ["tensor_a", "tensor_b"])
- args: Namespace = _make_args(
- baseline_path, target_path, filter_pattern="tensor_a"
- )
-
- main(args)
-
- captured: str = capsys.readouterr().out
- assert "tensor_a" in captured
- assert "Check:" in captured
-
- def test_no_match_skips(
- self, tmp_path: Path, capsys: pytest.CaptureFixture
- ) -> None:
- baseline_path, target_path = _create_dumps(
- tmp_path,
- ["only_in_target"],
- baseline_names=["only_in_baseline"],
- )
- args: Namespace = _make_args(baseline_path, target_path)
-
- main(args)
-
- captured: str = capsys.readouterr().out
- assert "Skip" in captured
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/ops/test_repeat_interleave.py b/test/registered/ops/test_repeat_interleave.py
deleted file mode 100644
index ca2c26b91..000000000
--- a/test/registered/ops/test_repeat_interleave.py
+++ /dev/null
@@ -1,148 +0,0 @@
-import time
-from typing import Tuple
-
-import numpy as np
-import pytest
-import torch
-
-from sglang.srt.models.utils import compute_cu_seqlens_from_grid_numpy as cpu_numpy_impl
-from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
-
-# Ops - Repeat Interleave tests (1-GPU)
-
-
-register_cuda_ci(est_time=7, suite="stage-b-test-1-gpu-small")
-register_amd_ci(est_time=75, suite="stage-b-test-1-gpu-small-amd")
-
-
-def torch_ref_impl(grid_thw: torch.Tensor) -> torch.Tensor:
- """
- Pure PyTorch implementation of cu_seqlens computation.
- Assumes grid_thw is already on the correct device (CPU here).
- Shape: [T, 3], columns: [repeat_count, H, W]
- """
- cu_seqlens = torch.repeat_interleave(
- grid_thw[:, 1] * grid_thw[:, 2], grid_thw[:, 0]
- ).cumsum(dim=0)
- cu_seqlens = torch.cat(
- [
- torch.zeros(1, dtype=torch.int32, device=cu_seqlens.device),
- cu_seqlens.to(torch.int32),
- ]
- )
- return cu_seqlens
-
-
-def benchmark_once(fn, grid_thw, iters: int = 1000):
- """
- Run a function `fn` on the same input `grid_thw` for `iters` times
- and measure total elapsed time.
- """
- start = time.perf_counter()
- for _ in range(iters):
- out = fn(grid_thw)
- end = time.perf_counter()
- return (end - start), out
-
-
-# (T, repeat_min, repeat_max)
-GRID_TEST_CONFIGS: list[Tuple[int, int, int]] = [
- (16, 1, 4), # small T, small repeat counts
- (128, 0, 4), # allow repeat=0 to test edge cases
- (512, 1, 8),
- (1024, 1, 16),
-]
-
-NUM_CASES_PER_CONFIG = 10
-
-
-def _generate_random_grid(T: int, repeat_min: int, repeat_max: int) -> torch.Tensor:
- """
- grid_thw: [T, 3]
- col0: repeat count
- col1, col2: arbitrary positive integers (here 1..16)
- """
- repeats = torch.randint(repeat_min, repeat_max + 1, (T, 1), dtype=torch.int32)
- th = torch.randint(1, 17, (T, 1), dtype=torch.int32)
- tw = torch.randint(1, 17, (T, 1), dtype=torch.int32)
- grid_thw = torch.cat([repeats, th, tw], dim=1)
- return grid_thw
-
-
-class TestRepeatInterleave:
- @classmethod
- def setup_class(cls):
- torch.set_num_threads(1)
-
- def setup_method(self, method):
- torch.manual_seed(0)
- np.random.seed(0)
-
- @pytest.mark.parametrize(
- "T,repeat_min,repeat_max",
- GRID_TEST_CONFIGS,
- )
- @pytest.mark.parametrize("case_idx", range(NUM_CASES_PER_CONFIG))
- def test_cpu_correctness_random_cases(
- self,
- T: int,
- repeat_min: int,
- repeat_max: int,
- case_idx: int,
- ):
- torch.manual_seed(case_idx)
- np.random.seed(case_idx)
-
- grid_thw = _generate_random_grid(T, repeat_min, repeat_max)
-
- grid_clone = grid_thw.clone()
-
- out_torch = torch_ref_impl(grid_thw)
- out_numpy = cpu_numpy_impl(grid_thw)
-
- assert torch.equal(grid_thw, grid_clone), "Function modified input grid_thw!"
-
- assert (
- out_torch.shape == out_numpy.shape
- ), f"Shape mismatch: torch={out_torch.shape}, numpy={out_numpy.shape}"
-
- assert (
- out_torch.dtype == torch.int32
- ), f"Unexpected torch dtype: {out_torch.dtype}"
- assert (
- out_numpy.dtype == torch.int32
- ), f"Unexpected numpy impl dtype: {out_numpy.dtype}"
-
- if not torch.equal(out_torch.cpu(), out_numpy.cpu()):
- diff_idx = (out_torch.cpu() != out_numpy.cpu()).nonzero(as_tuple=False)
- idx0 = diff_idx[0].item()
- pytest.fail(
- f"Value mismatch, T={T}, case_idx={case_idx}, first differing index={idx0}, "
- f"torch={out_torch[idx0].item()}, "
- f"numpy={out_numpy[idx0].item()}"
- )
-
- def test_zero_repeat_edge_case(self):
- T = 4
- grid_thw = torch.tensor(
- [
- [0, 4, 4],
- [1, 2, 3], # 6
- [2, 1, 5], # 5, 5
- [0, 7, 7], # 0
- ],
- dtype=torch.int32,
- )
-
- grid_clone = grid_thw.clone()
-
- out_torch = torch_ref_impl(grid_thw)
- out_numpy = cpu_numpy_impl(grid_thw)
-
- assert torch.equal(
- grid_thw, grid_clone
- ), "Function modified input grid_thw with zero repeats!"
-
- assert torch.equal(
- out_torch.cpu(), out_numpy.cpu()
- ), f"Zero-repeat case mismatch: torch={out_torch}, numpy={out_numpy}"
diff --git a/test/registered/quant/test_bnb.py b/test/registered/quant/test_bnb.py
deleted file mode 100644
index ee7964256..000000000
--- a/test/registered/quant/test_bnb.py
+++ /dev/null
@@ -1,310 +0,0 @@
-"""
-Usage:
-python3 -m unittest test_bnb.TestVisionModel.test_vlm
-python3 -m unittest test_bnb.TestLanguageModel.test_mmlu
-"""
-
-import multiprocessing as mp
-import random
-from concurrent.futures import ThreadPoolExecutor
-from types import SimpleNamespace
-
-import openai
-
-from sglang.srt.utils import kill_process_tree
-from sglang.test.ci.ci_register import register_cuda_ci
-from sglang.test.run_eval import run_eval
-from sglang.test.test_utils import (
- DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
- DEFAULT_URL_FOR_TEST,
- CustomTestCase,
- is_in_ci,
- popen_launch_server,
-)
-
-register_cuda_ci(est_time=6, suite="stage-b-test-1-gpu-small")
-
-VISION_MODELS = [
- "unsloth/Qwen2.5-VL-7B-Instruct-bnb-4bit",
- "unsloth/Qwen2-VL-7B-Instruct-bnb-4bit",
- "unsloth/Llama-3.2-11B-Vision-Instruct-bnb-4bit",
- "unsloth/Llama-3.2-11B-Vision-bnb-4bit",
- "unsloth/gemma-3-4b-it-bnb-4bit",
- "unsloth/gemma-3-4b-it-unsloth-bnb-4bit",
-]
-LANGUAGE_MODELS = [
- "unsloth/Qwen2.5-7B-Instruct-bnb-4bit",
- "unsloth/Qwen2-7B-Instruct-bnb-4bit",
- "unsloth/Llama-3.2-3B-Instruct-bnb-4bit",
- "unsloth/gemma-3-1b-it-bnb-4bit",
-]
-
-# image
-IMAGE_MAN_IRONING_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/man_ironing_on_back_of_suv.png"
-IMAGE_SGL_LOGO_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/images/sgl_logo.png"
-
-# video
-VIDEO_JOBS_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/videos/jobs_presenting_ipod.mp4"
-
-# audio
-AUDIO_TRUMP_SPEECH_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/audios/Trump_WEF_2018_10s.mp3"
-AUDIO_BIRD_SONG_URL = "https://raw.githubusercontent.com/sgl-project/sgl-test-files/refs/heads/main/audios/bird_song.mp3"
-
-
-def popen_launch_server_wrapper(base_url, model, other_args):
- process = popen_launch_server(
- model,
- base_url,
- timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
- other_args=other_args,
- )
- return process
-
-
-class TestVisionModel(CustomTestCase):
- @classmethod
- def setUpClass(cls):
- mp.set_start_method("spawn", force=True)
- cls.base_url = DEFAULT_URL_FOR_TEST
- cls.base_url += "/v1"
- cls.api_key = "sk-123456"
-
- def _run_single_image_chat_completion(self):
- client = openai.Client(api_key=self.api_key, base_url=self.base_url)
-
- response = client.chat.completions.create(
- model="default",
- messages=[
- {
- "role": "user",
- "content": [
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_MAN_IRONING_URL},
- },
- {
- "type": "text",
- "text": "Describe this image in a very short sentence.",
- },
- ],
- },
- ],
- temperature=0,
- )
-
- assert response.choices[0].message.role == "assistant"
- text = response.choices[0].message.content
- assert isinstance(text, str)
- # `driver` is for gemma-3-it
- assert "man" in text or "person" or "driver" in text, text
- assert "cab" in text or "taxi" in text or "SUV" in text, text
- # MiniCPMO fails to recognize `iron`, but `hanging`
- assert "iron" in text or "hang" in text, text
- assert response.id
- assert response.created
- assert response.usage.prompt_tokens > 0
- assert response.usage.completion_tokens > 0
- assert response.usage.total_tokens > 0
-
- def _run_multi_turn_chat_completion(self):
- client = openai.Client(api_key=self.api_key, base_url=self.base_url)
-
- response = client.chat.completions.create(
- model="default",
- messages=[
- {
- "role": "user",
- "content": [
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_MAN_IRONING_URL},
- },
- {
- "type": "text",
- "text": "Describe this image in a very short sentence.",
- },
- ],
- },
- {
- "role": "assistant",
- "content": [
- {
- "type": "text",
- "text": "There is a man at the back of a yellow cab ironing his clothes.",
- }
- ],
- },
- {
- "role": "user",
- "content": [
- {"type": "text", "text": "Repeat your previous answer."}
- ],
- },
- ],
- temperature=0,
- )
-
- assert response.choices[0].message.role == "assistant"
- text = response.choices[0].message.content
- assert isinstance(text, str)
- assert "man" in text or "cab" in text, text
- assert response.id
- assert response.created
- assert response.usage.prompt_tokens > 0
- assert response.usage.completion_tokens > 0
- assert response.usage.total_tokens > 0
-
- def _run_multi_images_chat_completion(self):
- client = openai.Client(api_key=self.api_key, base_url=self.base_url)
- response = client.chat.completions.create(
- model="default",
- messages=[
- {
- "role": "user",
- "content": [
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_MAN_IRONING_URL},
- "modalities": "multi-images",
- },
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_SGL_LOGO_URL},
- "modalities": "multi-images",
- },
- {
- "type": "text",
- "text": "I have two very different images. They are not related at all. "
- "Please describe the first image in one sentence, and then describe the second image in another sentence.",
- },
- ],
- },
- ],
- temperature=0,
- )
-
- assert response.choices[0].message.role == "assistant"
- text = response.choices[0].message.content
- assert isinstance(text, str)
- print("-" * 30)
- print(f"Multi images response:\n{text}")
- print("-" * 30)
- assert "man" in text or "cab" in text or "SUV" in text or "taxi" in text, text
- assert "logo" in text or '"S"' in text or "SG" in text, text
- assert response.id
- assert response.created
- assert response.usage.prompt_tokens > 0
- assert response.usage.completion_tokens > 0
- assert response.usage.total_tokens > 0
-
- def run_decode_with_image(self, image_id):
- client = openai.Client(api_key=self.api_key, base_url=self.base_url)
-
- content = []
- if image_id == 0:
- content.append(
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_MAN_IRONING_URL},
- }
- )
- elif image_id == 1:
- content.append(
- {
- "type": "image_url",
- "image_url": {"url": IMAGE_SGL_LOGO_URL},
- }
- )
- else:
- pass
-
- content.append(
- {
- "type": "text",
- "text": "Describe this image in a very short sentence.",
- }
- )
-
- response = client.chat.completions.create(
- model="default",
- messages=[
- {"role": "user", "content": content},
- ],
- temperature=0,
- )
-
- assert response.choices[0].message.role == "assistant"
- text = response.choices[0].message.content
- assert isinstance(text, str)
-
- def _run_test_mixed_batch(self):
- image_ids = [0, 1, 2] * 4
- with ThreadPoolExecutor(4) as executor:
- list(executor.map(self.run_decode_with_image, image_ids))
-
- def test_vlm(self):
- models_to_test = VISION_MODELS
-
- if is_in_ci():
- models_to_test = [random.choice(VISION_MODELS)]
-
- for model in models_to_test:
- with self.subTest(model=model):
- other_args = [
- "--mem-fraction-static",
- "0.6",
- "--load-format",
- "bitsandbytes",
- "--enable-multimodal",
- ]
- try:
- process = popen_launch_server_wrapper(
- DEFAULT_URL_FOR_TEST, model, other_args
- )
- self._run_test_mixed_batch()
- self._run_multi_images_chat_completion()
- self._run_multi_turn_chat_completion()
- self._run_single_image_chat_completion()
- finally:
- kill_process_tree(process.pid)
-
-
-class TestLanguageModel(CustomTestCase):
- @classmethod
- def setUpClass(cls):
- mp.set_start_method("spawn", force=True)
- cls.base_url = DEFAULT_URL_FOR_TEST
- # cls.base_url += "/v1"
- cls.api_key = "sk-123456"
-
- def test_mmlu(self):
- models_to_test = LANGUAGE_MODELS
-
- if is_in_ci():
- models_to_test = [random.choice(LANGUAGE_MODELS)]
-
- for model in models_to_test:
- with self.subTest(model=model):
- other_args = [
- "--mem-fraction-static",
- "0.6",
- "--load-format",
- "bitsandbytes",
- ]
- try:
- process = popen_launch_server_wrapper(
- DEFAULT_URL_FOR_TEST, model, other_args
- )
- args = SimpleNamespace(
- base_url=self.base_url,
- model=model,
- eval_name="mmlu",
- num_examples=32,
- num_threads=16,
- )
-
- metrics = run_eval(args)
- print(f"{metrics=}")
- self.assertGreater(metrics["score"], 0.3)
- finally:
- kill_process_tree(process.pid)
diff --git a/test/registered/quant/test_int4fp8_moe.py b/test/registered/quant/test_int4fp8_moe.py
deleted file mode 100644
index c46c50447..000000000
--- a/test/registered/quant/test_int4fp8_moe.py
+++ /dev/null
@@ -1,59 +0,0 @@
-from types import SimpleNamespace
-
-from sglang.srt.utils import kill_process_tree
-from sglang.test.ci.ci_register import register_amd_ci
-from sglang.test.run_eval import run_eval
-from sglang.test.test_utils import (
- DEFAULT_URL_FOR_TEST,
- CustomTestCase,
- popen_launch_server,
-)
-
-register_amd_ci(est_time=313, suite="stage-b-test-1-gpu-small-amd")
-
-
-class TestMixtralAccuracy(CustomTestCase):
- @classmethod
- def setUpClass(cls):
- cls.model = "mistralai/Mixtral-8x7B-Instruct-v0.1"
- cls.base_url = DEFAULT_URL_FOR_TEST
-
- other_args = [
- "--tp",
- "2",
- "--mem-fraction-static",
- "0.9",
- "--context-length",
- "38768",
- "--quantization",
- "quark_int4fp8_moe",
- # The default aiter attention backend raises segmentation faults and other errors - as quark_int4fp8_moe is not related to attention, let's just use triton here.
- "--attention-backend",
- "triton",
- ]
-
- cls.process = popen_launch_server(
- cls.model,
- cls.base_url,
- timeout=45 * 60,
- other_args=other_args,
- )
-
- @classmethod
- def tearDownClass(cls):
- kill_process_tree(cls.process.pid)
-
- def test_gsm8k(self):
- args = SimpleNamespace(
- base_url=self.base_url,
- model=self.model,
- eval_name="gsm8k",
- api="completion",
- max_tokens=512,
- num_examples=1400,
- num_threads=128,
- num_shots=8,
- )
- metrics = run_eval(args)
- print(f"{metrics=}")
- self.assertGreater(metrics["score"], 0.56)
diff --git a/test/registered/rotary/test_mrope.py b/test/registered/rotary/test_mrope.py
deleted file mode 100644
index 87b34df91..000000000
--- a/test/registered/rotary/test_mrope.py
+++ /dev/null
@@ -1,174 +0,0 @@
-# Rotary Embedding - MRoPE tests (1-GPU)
-
-from typing import NamedTuple
-
-import pytest
-import torch
-from packaging.version import Version
-from transformers import AutoConfig
-from transformers import __version__ as TRANSFORMERS_VERSION
-
-from sglang.srt.layers.rotary_embedding import get_rope
-from sglang.srt.server_args import ServerArgs, set_global_server_args_for_scheduler
-from sglang.srt.utils import (
- cpu_has_amx_support,
- is_cpu,
- is_cuda,
- is_hip,
- is_npu,
- is_xpu,
-)
-from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
-
-register_cuda_ci(est_time=7, suite="stage-b-test-1-gpu-large")
-register_amd_ci(est_time=15, suite="stage-b-test-1-gpu-small-amd")
-
-_is_cuda = is_cuda()
-_is_hip = is_hip()
-_is_cpu = is_cpu()
-_is_cpu_amx_available = cpu_has_amx_support()
-_is_npu = is_npu()
-_is_xpu = is_xpu()
-
-device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-
-
-def generate_test_data(
- num_tokens: int,
- num_q_heads: int,
- num_kv_heads: int,
- head_size: int,
- max_position_embeddings: int,
- dtype: torch.dtype,
- device: torch.device,
-):
- """Generate test data for given configuration."""
- torch.manual_seed(42)
- # Create 2D positions (3, num_tokens) for multimodal case
- positions = torch.randint(
- 0, max_position_embeddings // 4, (3, num_tokens), device=device
- )
-
- # Create query and key tensors
- query = torch.randn(num_tokens, num_q_heads * head_size, dtype=dtype, device=device)
- key = torch.randn(num_tokens, num_kv_heads * head_size, dtype=dtype, device=device)
-
- return positions, query, key
-
-
-class MRoPETestInfo(NamedTuple):
- model_name: str
- atol: float = 1e-2
- rtol: float = 1.6e-2
- marks: list[pytest.MarkDecorator] = []
-
-
-TRANSFORMERS_BASE_VERSION = Version(TRANSFORMERS_VERSION).base_version
-
-MODELS_TO_TEST = [
- MRoPETestInfo(model_name="Qwen/Qwen2-VL-7B-Instruct"),
- MRoPETestInfo(model_name="Qwen/Qwen2-VL-72B-Instruct"),
- MRoPETestInfo(model_name="Qwen/Qwen2.5-VL-72B-Instruct"),
-]
-
-num_tokens_list = [11, 8192]
-
-
-def create_yarn_rope_scaling(original_config, scaling_factor=2.0):
- yarn_config = {
- "rope_type": "yarn",
- "factor": scaling_factor,
- "original_max_position_embeddings": original_config.max_position_embeddings,
- }
- if hasattr(original_config, "rope_scaling") and original_config.rope_scaling:
- if "mrope_section" in original_config.rope_scaling:
- yarn_config["mrope_section"] = original_config.rope_scaling["mrope_section"]
- if "mrope_interleaved" in original_config.rope_scaling:
- yarn_config["mrope_interleaved"] = original_config.rope_scaling[
- "mrope_interleaved"
- ]
- return yarn_config
-
-
-@pytest.mark.skipif(not (_is_cuda or _is_hip), reason="Skipping CUDA/ROCm only tests.")
-@pytest.mark.parametrize(
- "model_info, model_name",
- [
- pytest.param(test_config, test_config.model_name, marks=test_config.marks)
- for test_config in MODELS_TO_TEST
- ],
-)
-@pytest.mark.parametrize("tp_size", [1, 2])
-@pytest.mark.parametrize("dtype", [torch.bfloat16, torch.float32])
-@pytest.mark.parametrize("num_tokens", num_tokens_list)
-@pytest.mark.parametrize(
- "rope_scaling_type", ["default", "yarn"], ids=["mrope_default", "mrope_yarn"]
-)
-def test_mrope(
- model_name: str,
- model_info: MRoPETestInfo,
- tp_size: int,
- dtype: torch.dtype,
- num_tokens: int,
- rope_scaling_type: str,
-):
- set_global_server_args_for_scheduler(ServerArgs(model_path="dummy"))
-
- atol = model_info.atol
- rtol = model_info.rtol
-
- config = AutoConfig.from_pretrained(model_name)
- config = config.get_text_config()
-
- # get the model config
- total_num_kv_heads = config.num_key_value_heads
- total_num_heads = config.num_attention_heads
- num_heads = total_num_heads // tp_size
- num_kv_heads = max(1, total_num_kv_heads // tp_size)
- head_dim = (
- config.head_dim
- if hasattr(config, "head_dim")
- else config.hidden_size // total_num_heads
- )
- is_neox_style = True
-
- rope_theta = config.rope_theta
- max_position = config.max_position_embeddings
- partial_rotary_factor = getattr(config, "partial_rotary_factor", 1.0)
- rotary_dim = int(head_dim * partial_rotary_factor)
-
- if rope_scaling_type == "yarn":
- rope_scaling_config = create_yarn_rope_scaling(config, scaling_factor=2.0)
- else:
- rope_scaling_config = config.rope_scaling
-
- mrope_helper_class = get_rope(
- head_size=head_dim,
- rotary_dim=rotary_dim,
- max_position=max_position,
- base=rope_theta,
- is_neox_style=is_neox_style,
- rope_scaling=rope_scaling_config,
- dtype=dtype,
- ).to(device=device)
-
- # create q k v input tensors
- # create rotary pos emb input tensors
- positions, query, key = generate_test_data(
- num_tokens, num_heads, num_kv_heads, head_dim, max_position, dtype, device
- )
-
- query_native, key_native = mrope_helper_class.forward_native(
- positions,
- query.clone(),
- key.clone(),
- )
-
- query_cuda, key_cuda = mrope_helper_class.forward(
- positions,
- query.clone(),
- key.clone(),
- )
-
- torch.testing.assert_close(query_native, query_cuda, atol=atol, rtol=rtol)
- torch.testing.assert_close(key_native, key_cuda, atol=atol, rtol=rtol)
diff --git a/test/registered/unit/function_call/test_glm47_moe_detector.py b/test/registered/unit/function_call/test_glm47_moe_detector.py
deleted file mode 100644
index e0c192119..000000000
--- a/test/registered/unit/function_call/test_glm47_moe_detector.py
+++ /dev/null
@@ -1,1847 +0,0 @@
-import json
-import unittest
-
-from sglang.srt.entrypoints.openai.protocol import Function, Tool
-from sglang.srt.function_call.core_types import StreamingParseResult
-from sglang.srt.function_call.glm4_moe_detector import Glm4MoeDetector
-from sglang.srt.function_call.glm47_moe_detector import (
- Glm47MoeDetector,
- get_argument_type,
-)
-from sglang.test.ci.ci_register import register_cpu_ci
-
-register_cpu_ci(5, "stage-a-test-cpu")
-
-
-class TestGlm47MoeDetector(unittest.TestCase):
- def setUp(self):
- self.tools = [
- Tool(
- type="function",
- function=Function(
- name="get_weather",
- description="Get weather information",
- parameters={
- "type": "object",
- "properties": {
- "city": {"type": "string", "description": "City name"},
- "date": {"type": "string", "description": "Date"},
- },
- "required": ["city", "date"],
- },
- ),
- ),
- ]
- self.detector = Glm47MoeDetector()
-
- # ==================== Basic Parsing Tests (5) ====================
-
- def test_single_tool_call(self):
- """
- Test basic single tool call parsing.
-
- Scenario: Parse a complete tool call with two string parameters in a single text block.
- Purpose: Verify the detector can correctly identify and extract function name and parameters
- from a simple, well-formed tool call.
- """
- text = (
- "get_weather"
- "cityBeijing"
- "date2024-06-27"
- ""
- )
- result = self.detector.detect_and_parse(text, self.tools)
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "get_weather")
- self.assertEqual(
- result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
- )
- self.assertEqual(result.normal_text, "")
-
- def test_multiple_tool_calls(self):
- """
- Test parsing multiple consecutive tool calls.
-
- Scenario: Parse two complete tool calls back-to-back without any text in between.
- Purpose: Verify the detector correctly handles multiple tool calls and resets state
- between calls to avoid parameter leakage or ID conflicts.
- """
- text = (
- "get_weather"
- "cityBeijing"
- "date2024-06-27"
- ""
- "get_weather"
- "cityShanghai"
- "date2024-06-28"
- ""
- )
- result = self.detector.detect_and_parse(text, self.tools)
- self.assertEqual(len(result.calls), 2)
- self.assertEqual(result.calls[0].name, "get_weather")
- self.assertEqual(
- result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
- )
- self.assertEqual(result.calls[1].name, "get_weather")
- self.assertEqual(
- result.calls[1].parameters, '{"city": "Shanghai", "date": "2024-06-28"}'
- )
- self.assertEqual(result.normal_text, "")
-
- def test_no_arg_function_non_streaming(self):
- """
- Test no-argument function call without streaming.
-
- Scenario: Parse a tool call for a function that has no parameters (empty properties).
- Purpose: Verify the detector generates a single empty object "{}" for no-argument functions
- and does not duplicate empty parameter objects.
- """
- tools_with_no_args = [
- Tool(
- type="function",
- function=Function(
- name="list_filenames",
- description="List filenames",
- parameters={
- "type": "object",
- "properties": {},
- },
- ),
- ),
- ]
-
- text = "list_filenames"
- result = self.detector.detect_and_parse(text, tools_with_no_args)
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "list_filenames")
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params, {})
-
- def test_invalid_tool_call(self):
- """
- Test handling of invalid tool calls.
-
- Scenario: Attempt to parse a tool call with a function name that doesn't exist in the tool list.
- Purpose: Verify the detector gracefully rejects invalid function calls and returns no calls
- rather than throwing an error or accepting invalid input.
- """
- text = "invalid_funccityBeijing"
- result = self.detector.detect_and_parse(text, self.tools)
- self.assertEqual(len(result.calls), 0)
-
- def test_array_argument_with_escaped_json(self):
- """
- Test array arguments containing escaped JSON strings.
-
- Scenario: Parse tool calls with array parameters containing nested JSON objects with
- escaped quotes (both backslash-escaped and raw escaped strings).
- Purpose: Verify the detector properly handles JSON escaping without double-escaping,
- preserving special characters like backslashes in paths and newline sequences.
- """
- tools_with_array = [
- Tool(
- type="function",
- function=Function(
- name="todo_write",
- description="Write todos",
- parameters={
- "type": "object",
- "properties": {
- "todos": {
- "type": "array",
- "description": "The updated todo list",
- }
- },
- "required": ["todos"],
- },
- ),
- ),
- ]
-
- def check_params(result):
- self.assertEqual(1, len(result.calls))
- self.assertEqual("todo_write", result.calls[0].name)
- params = json.loads(result.calls[0].parameters)
- self.assertIsInstance(params["todos"], list)
- self.assertEqual(4, len(params["todos"]))
- self.assertEqual("1", params["todos"][0]["id"])
- self.assertEqual(
- "Check for hard-coded issues in the backend code",
- params["todos"][0]["task"],
- )
- self.assertEqual("in_progress", params["todos"][0]["status"])
- self.assertEqual("2", params["todos"][1]["id"])
- self.assertEqual(
- "Check for hard-coded issues in the frontend code",
- params["todos"][1]["task"],
- )
- self.assertEqual("pending", params["todos"][1]["status"])
- self.assertEqual("3", params["todos"][2]["id"])
- self.assertEqual(
- "Check for code violating the Single Responsibility Principle",
- params["todos"][2]["task"],
- )
- self.assertEqual("pending", params["todos"][2]["status"])
- self.assertEqual("4", params["todos"][3]["id"])
- self.assertEqual(
- "Generate a rectification proposal report", params["todos"][3]["task"]
- )
- self.assertEqual("pending", params["todos"][3]["status"])
-
- # Test with normal escaped JSON in XML
- result = self.detector.detect_and_parse(
- """todo_writetodos[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]
-""",
- tools_with_array,
- )
- check_params(result)
-
- # Test with raw string escaped JSON
- result = self.detector.detect_and_parse(
- r"""todo_writetodos[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]
-""",
- tools_with_array,
- )
- check_params(result)
-
- def check_single_todos(tool_result, expected):
- self.assertEqual(1, len(tool_result.calls))
- self.assertEqual("todo_write", tool_result.calls[0].name)
- params = json.loads(tool_result.calls[0].parameters)
- self.assertIsInstance(params["todos"], list)
- self.assertEqual(1, len(params["todos"]))
- self.assertEqual("1", params["todos"][0]["id"])
- self.assertEqual(expected, params["todos"][0]["task"])
- self.assertEqual("pending", params["todos"][0]["status"])
-
- # Test with escaped backslashes (Windows paths)
- expected_path = r"Check file at C:\Users\test.txt"
- result = self.detector.detect_and_parse(
- """todo_writetodos[{\"id\": \"1\", \"task\": \"Check file at C:\\\\Users\\\\test.txt\", \"status\": \"pending\"}]""",
- tools_with_array,
- )
- check_single_todos(result, expected_path)
-
- # Test with literal backslash-n (not newline)
- expected_output = r"Print \n to see newline"
- result = self.detector.detect_and_parse(
- """todo_writetodos[{\"id\": \"1\", \"task\": \"Print \\\\n to see newline\",\"status\": \"pending\"}]""",
- tools_with_array,
- )
- check_single_todos(result, expected_output)
-
- # ==================== MTP Core Scenarios (3) ====================
-
- def test_mtp_func_and_string_split(self):
- """
- Test MTP-style function name and string parameter value splitting across chunks.
-
- Scenario: Simulate Model Token Provider (MTP) behavior where function names and string
- parameter values are split mid-word across multiple chunks.
- Purpose: This is the MOST CRITICAL test - verify the detector correctly reassembles:
- - Function name split as "create_ta" + "sk"
- - String values split as "Go to Bei" + "jing" and "San Fran" + "cisco"
- These splits mimic real MTP output where tokenization breaks words arbitrarily.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="create_task",
- parameters={
- "type": "object",
- "properties": {
- "title": {"type": "string"},
- "location": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- chunks = [
- "I'll create a task.", # normal text before tool call
- "create_ta", # function name split mid-word
- "sktitleGo to Bei", # function name completes, param value splits
- "jing", # first parameter value completes
- "locationSan Fran", # second parameter value splits
- "cisco", # second parameter and tool call complete
- ]
-
- detector = Glm47MoeDetector()
- all_calls = []
- all_normal_text = ""
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, tools)
- all_calls.extend(result.calls)
- all_normal_text += result.normal_text
-
- # Verify normal text is preserved
- self.assertEqual(all_normal_text, "I'll create a task.")
-
- # Verify function call
- func_calls = [c for c in all_calls if c.name]
- self.assertEqual(len(func_calls), 1)
- self.assertEqual(
- func_calls[0].name, "create_task"
- ) # "create_ta" + "sk" reassembled
-
- # Verify parameter reassembly
- full_params = "".join([c.parameters for c in all_calls if c.parameters])
- params = json.loads(full_params)
- self.assertEqual(
- params["title"], "Go to Beijing"
- ) # "Go to Bei" + "jing" reassembled
- self.assertEqual(
- params["location"], "San Francisco"
- ) # "San Fran" + "cisco" reassembled
-
- def test_mtp_noarg_and_multiple_calls(self):
- """
- Test MTP-style no-argument function and multiple tool calls with state reset.
-
- Scenario: Stream a no-argument function call followed by a regular function call,
- simulating MTP's output pattern where function completion triggers state reset.
- Purpose: Verify:
- - No-argument functions emit exactly ONE empty object "{}", not duplicates
- - State properly resets between consecutive tool calls (tool_index increments)
- - Second tool call doesn't inherit parameters from first call
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="list_files",
- parameters={
- "type": "object",
- "properties": {},
- },
- ),
- ),
- Tool(
- type="function",
- function=Function(
- name="get_weather",
- parameters={
- "type": "object",
- "properties": {
- "city": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- chunks = [
- "list_files", # no-arg function, complete in one chunk
- "get_weathercityBeijing",
- ]
-
- detector = Glm47MoeDetector()
- all_calls = []
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, tools)
- all_calls.extend(result.calls)
-
- # Verify two distinct tool calls
- func_calls = [c for c in all_calls if c.name]
- self.assertEqual(len(func_calls), 2)
- self.assertEqual(func_calls[0].name, "list_files")
- self.assertEqual(func_calls[1].name, "get_weather")
-
- # Verify no duplicate empty objects for no-arg function
- empty_object_calls = [c for c in all_calls if c.parameters == "{}"]
- self.assertLessEqual(
- len(empty_object_calls),
- 1,
- "No-argument function should emit at most one empty object",
- )
-
- # Verify second call has correct parameters
- weather_params = [
- c.parameters for c in all_calls if c.parameters and c.parameters != "{}"
- ]
- if weather_params:
- full_params = "".join(weather_params)
- params = json.loads(full_params)
- self.assertEqual(params["city"], "Beijing")
-
- def test_mtp_number_and_complex_json(self):
- """
- Test MTP-style number parameters and complex JSON array splitting.
-
- Scenario: Parse tool calls with number parameters (int and float) and JSON arrays
- split across chunks, including splits within JSON structure.
- Purpose: Verify:
- - Number types (5.5, 10) are preserved as numbers, not strings
- - JSON array content split as "description" + ": \"" maintains validity
- - Nested JSON objects in arrays are correctly reconstructed
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="create_todos",
- parameters={
- "type": "object",
- "properties": {
- "priority": {"type": "number"},
- "count": {"type": "integer"},
- "items": {"type": "array"},
- },
- },
- ),
- ),
- ]
-
- chunks = [
- "create_todos",
- "priority5.5", # float number
- "count10", # integer number
- 'items[{"description', # JSON array splits mid-key
- '": "Test', # key completes, value starts
- 'Todo 1"}, {"description": "TestTodo 2"}]',
- ]
-
- detector = Glm47MoeDetector()
- all_calls = []
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, tools)
- all_calls.extend(result.calls)
-
- # Verify function name
- func_calls = [c for c in all_calls if c.name]
- self.assertEqual(len(func_calls), 1)
- self.assertEqual(func_calls[0].name, "create_todos")
-
- # Verify parameters - numbers and JSON array
- full_params = "".join([c.parameters for c in all_calls if c.parameters])
- params = json.loads(full_params)
-
- # Number types should be preserved
- self.assertIsInstance(params["priority"], (int, float))
- self.assertEqual(params["priority"], 5.5)
- self.assertIsInstance(params["count"], int)
- self.assertEqual(params["count"], 10)
-
- # JSON array should be correctly reconstructed
- self.assertIsInstance(params["items"], list)
- self.assertEqual(len(params["items"]), 2)
- self.assertEqual(params["items"][0]["description"], "TestTodo 1")
- self.assertEqual(params["items"][1]["description"], "TestTodo 2")
-
- # ==================== Streaming Basics (3) ====================
-
- def test_streaming_tool_call(self):
- """
- Test basic streaming incremental parsing of a single tool call.
-
- Scenario: Parse a tool call split across 4 chunks with natural boundaries
- (function name, first param, second param, closing tag).
- Purpose: Verify basic streaming functionality works correctly and accumulates
- parameters progressively across chunks.
- """
- chunks = [
- "get_weather",
- "cityBeijing",
- "date2024-06-27",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, self.tools)
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "get_weather")
- self.assertEqual(
- tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
- )
-
- def test_streaming_multiple_tool_calls(self):
- """
- Test streaming incremental parsing of multiple consecutive tool calls.
-
- Scenario: Stream two complete tool calls with the transition ""
- occurring within a single chunk.
- Purpose: Verify streaming correctly handles multiple tool calls and properly increments
- tool_index for each new call.
- """
- chunks = [
- "get_weather",
- "cityBeijing",
- "date2024-06-27",
- "get_weather", # two tool calls transition in same chunk
- "cityShanghai",
- "date2024-06-28",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, self.tools)
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
- self.assertEqual(len(tool_calls), 2)
- self.assertEqual(tool_calls[0]["name"], "get_weather")
- self.assertEqual(
- tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
- )
- self.assertEqual(tool_calls[1]["name"], "get_weather")
- self.assertEqual(
- tool_calls[1]["parameters"], '{"city": "Shanghai", "date": "2024-06-28"}'
- )
-
- def test_normal_text_before_tool_call(self):
- """
- Test preservation of normal text (including punctuation) before tool calls.
-
- Scenario: Parse chunks containing normal text with various punctuation marks
- (English and Chinese) immediately followed by tool call tags.
- Purpose: Verify normal text is preserved in result.normal_text and not lost when
- tool call parsing begins. This consolidates 6 previous Chinese punctuation tests.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="list_dir",
- parameters={
- "type": "object",
- "properties": {
- "path": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- test_cases = [
- ("Sure, let me help.list_dir", "English with period"),
- ("结构：list_dir", "Chinese colon"),
- ("问题。list_dir", "Chinese period"),
- ("Complete!list_dir", "English exclamation"),
- ("说明；list_dir", "Chinese semicolon"),
- ]
-
- for text, description in test_cases:
- with self.subTest(description=description):
- detector = Glm47MoeDetector()
- result = detector.parse_streaming_increment(text, tools)
-
- before_token = text.split("")[0]
- self.assertIn(
- before_token,
- result.normal_text,
- f"Should preserve '{before_token}' in '{description}'",
- )
-
- # ==================== Boundary Cases (9) ====================
-
- def test_boundary_empty_param_value(self):
- """
- Test handling of empty parameter values.
-
- Scenario: Parse a tool call where a parameter value is an empty string.
- Purpose: Verify the detector correctly handles empty strings as valid parameter values
- and doesn't skip or error on them.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="create_note",
- parameters={
- "type": "object",
- "properties": {
- "title": {"type": "string"},
- "content": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- text = "create_notetitleTestcontent"
- result = self.detector.detect_and_parse(text, tools)
-
- self.assertEqual(len(result.calls), 1)
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["title"], "Test")
- self.assertEqual(params["content"], "") # empty string should be preserved
-
- def test_boundary_param_value_extreme_split(self):
- """
- Test extreme parameter value splitting - one character per chunk.
-
- Scenario: Stream a parameter value where each character arrives in a separate chunk,
- representing worst-case MTP tokenization.
- Purpose: Stress test the buffer reassembly mechanism to ensure it can handle
- extremely granular chunk boundaries without data loss or corruption.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="search",
- parameters={
- "type": "object",
- "properties": {
- "query": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- chunks = [
- "searchqueryN",
- "e",
- "w ",
- "Y",
- "o",
- "rk",
- ]
-
- detector = Glm47MoeDetector()
- all_calls = []
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, tools)
- all_calls.extend(result.calls)
-
- full_params = "".join([c.parameters for c in all_calls if c.parameters])
- params = json.loads(full_params)
- self.assertEqual(
- params["query"], "New York"
- ) # all characters correctly reassembled
-
- def test_boundary_param_value_with_special_chars(self):
- """
- Test parameter values containing special characters and escape sequences.
-
- Scenario: Parse parameter values with quotes, backslashes, newlines, and other
- special characters that require JSON escaping.
- Purpose: Verify special characters are properly escaped/unescaped and preserved
- through the parsing pipeline without corruption.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="execute_command",
- parameters={
- "type": "object",
- "properties": {
- "command": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- # Test with single quotes (no escaping needed)
- text = "execute_commandcommandecho 'Hello World'"
- result = self.detector.detect_and_parse(text, tools)
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["command"], "echo 'Hello World'")
-
- # Test with spaces and special chars that don't need escaping
- text = "execute_commandcommandecho Hello & World"
- result = self.detector.detect_and_parse(text, tools)
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["command"], "echo Hello & World")
-
- def test_boundary_json_deeply_nested(self):
- """
- Test deeply nested JSON structures in parameter values.
-
- Scenario: Parse a parameter containing a deeply nested JSON object with multiple levels.
- Purpose: Verify the detector can handle complex nested structures without stack overflow
- or parsing errors.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="process_data",
- parameters={
- "type": "object",
- "properties": {
- "data": {"type": "object"},
- },
- },
- ),
- ),
- ]
-
- nested_json = (
- '{"level1": {"level2": {"level3": {"level4": {"value": "deep"}}}}}'
- )
- text = f"process_datadata{nested_json}"
-
- result = self.detector.detect_and_parse(text, tools)
- params = json.loads(result.calls[0].parameters)
-
- # Navigate through nested structure
- self.assertEqual(
- params["data"]["level1"]["level2"]["level3"]["level4"]["value"], "deep"
- )
-
- def test_boundary_json_empty_structures(self):
- """
- Test empty JSON structures (empty objects and arrays) in parameters.
-
- Scenario: Parse parameters containing empty objects {} and empty arrays [].
- Purpose: Verify empty structures are preserved and not confused with no-argument
- function empty parameter generation.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="create_structure",
- parameters={
- "type": "object",
- "properties": {
- "empty_obj": {"type": "object"},
- "empty_arr": {"type": "array"},
- },
- },
- ),
- ),
- ]
-
- text = "create_structureempty_obj{}empty_arr[]"
- result = self.detector.detect_and_parse(text, tools)
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["empty_obj"], {})
- self.assertEqual(params["empty_arr"], [])
-
- def test_boundary_multi_tags_one_chunk(self):
- """
- Test multiple XML tags appearing in a single chunk.
-
- Scenario: Parse chunks where multiple complete tags (arg_key, arg_value, etc.)
- appear together without any chunk boundaries between them.
- Purpose: Verify the regex-based tag extraction correctly handles multiple tags
- in one chunk and processes them in the correct order.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="multi_param",
- parameters={
- "type": "object",
- "properties": {
- "a": {"type": "string"},
- "b": {"type": "string"},
- "c": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- # All three parameters in one chunk
- text = "multi_parama1b2c3"
- result = self.detector.detect_and_parse(text, tools)
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["a"], "1")
- self.assertEqual(params["b"], "2")
- self.assertEqual(params["c"], "3")
-
- def test_boundary_normal_text_mixed_with_tool(self):
- """
- Test normal text interleaved with tool calls.
-
- Scenario: Parse text with normal text before and after tool calls.
- Purpose: Verify normal text segments are correctly separated from tool call parsing
- and preserved in the normal_text output.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="action",
- parameters={
- "type": "object",
- "properties": {},
- },
- ),
- ),
- ]
-
- text = "First I'll do this.actionThen I'll do that."
- result = self.detector.detect_and_parse(text, tools)
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "action")
- # Verify both text before and after tool calls are preserved
- self.assertIn("First I'll do this.", result.normal_text)
- self.assertIn("Then I'll do that.", result.normal_text)
-
- def test_boundary_number_edge_values(self):
- """
- Test edge-case number values (zero, negative, scientific notation).
-
- Scenario: Parse parameters with various numeric edge cases to ensure proper type handling.
- Purpose: Verify the detector correctly preserves number types for edge values and doesn't
- convert them to strings or lose precision.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="calculate",
- parameters={
- "type": "object",
- "properties": {
- "zero": {"type": "number"},
- "negative": {"type": "number"},
- "large": {"type": "number"},
- },
- },
- ),
- ),
- ]
-
- text = "calculatezero0negative-42.5large1e10"
- result = self.detector.detect_and_parse(text, tools)
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["zero"], 0)
- self.assertEqual(params["negative"], -42.5)
- self.assertEqual(params["large"], 1e10)
-
- def test_boundary_type_string_with_numeric_content(self):
- """
- Test string parameters that contain numeric-looking content.
-
- Scenario: Parse string parameters with values like "123" or "45.67" that look like
- numbers but should remain strings based on parameter schema.
- Purpose: Verify type preservation based on schema definition, not content appearance.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="store_data",
- parameters={
- "type": "object",
- "properties": {
- "id": {
- "type": "string"
- }, # string type despite numeric content
- "code": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- text = "store_dataid12345code67.89"
- result = self.detector.detect_and_parse(text, tools)
-
- params = json.loads(result.calls[0].parameters)
- # Should be strings, not numbers
- self.assertIsInstance(params["id"], str)
- self.assertIsInstance(params["code"], str)
- self.assertEqual(params["id"], "12345")
- self.assertEqual(params["code"], "67.89")
-
- # ==================== Error Handling (2) ====================
-
- def test_error_undefined_tool(self):
- """
- Test error handling for undefined tool names.
-
- Scenario: Attempt to call a function that doesn't exist in the provided tools list.
- Purpose: Verify the detector gracefully handles undefined tools by returning an empty
- call list rather than crashing or producing malformed output.
- """
- text = "nonexistent_functionparamvalue"
- result = self.detector.detect_and_parse(text, self.tools)
-
- # Should not crash, should return empty calls
- self.assertEqual(len(result.calls), 0)
-
- def test_error_incomplete_buffer_at_end(self):
- """
- Test handling of incomplete tool calls at end of stream.
-
- Scenario: Streaming ends with an incomplete tool call (e.g., missing closing tag).
- Purpose: Verify the detector handles incomplete buffers gracefully without throwing
- exceptions, as streaming may end mid-parse in real scenarios.
- """
- chunks = [
- "get_weathercityBeijing",
- # Stream ends here, no closing tags
- ]
-
- detector = Glm47MoeDetector()
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, self.tools)
- # Should not crash
- self.assertIsInstance(result, StreamingParseResult)
-
- # Incomplete call should not be in results
- # (or may be partially present - main thing is no exception)
-
- # ==================== Streamed Raw Length Bug Tests (3) ====================
-
- def test_streamed_raw_length_incomplete_xml_tag(self):
- """
- Test that _streamed_raw_length is updated even when json_increment is empty.
-
- Scenario: Stream XML content that is split at an incomplete tag boundary,
- causing the state machine to buffer without producing JSON output.
- Purpose: Verify that _streamed_raw_length is updated regardless of whether
- json_increment is empty, preventing reprocessing of the same input.
-
- This tests the bug where:
- 1. raw_increment is extracted from func_args_raw[self._streamed_raw_length:]
- 2. _process_xml_to_json_streaming() returns empty string (buffering state)
- 3. If _streamed_raw_length is NOT updated before the early return,
- the next call will reprocess the same raw_increment
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="get_weather",
- parameters={
- "type": "object",
- "properties": {
- "city": {"type": "string"},
- "temperature": {"type": "number"},
- },
- },
- ),
- ),
- ]
-
- # Simulate streaming chunks where XML tags are split
- chunks = [
- "get_weather",
- "cityBei", # Split in middle of value
- "jing", # Complete the value
- "temperature2", # Split numeric value
- "5",
- ]
-
- detector = Glm47MoeDetector()
- all_calls = []
- collected_params = ""
-
- for i, chunk in enumerate(chunks):
- result = detector.parse_streaming_increment(chunk, tools)
- all_calls.extend(result.calls)
-
- # Collect parameters
- for call in result.calls:
- if call.parameters:
- collected_params += call.parameters
-
- # Verify complete parameters were collected without duplication
- if collected_params:
- params = json.loads(collected_params)
- self.assertEqual(params["city"], "Beijing")
- self.assertEqual(params["temperature"], 25)
-
- # Critical: Verify no duplicate JSON output due to reprocessing
- # Count occurrences of "city" key - should appear exactly once
- city_count = collected_params.count('"city"')
- self.assertEqual(
- city_count,
- 1,
- f"'city' key appears {city_count} times, expected 1. "
- f"This indicates input reprocessing bug.",
- )
-
- def test_streamed_raw_length_tag_split_across_chunks(self):
- """
- Test _streamed_raw_length update when tag is split across chunk boundaries.
-
- Scenario: XML tags themselves are split across chunks (e.g., "").
- Purpose: Verify that even when the state machine is buffering partial tags,
- _streamed_raw_length is correctly updated to prevent reprocessing.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="search",
- parameters={
- "type": "object",
- "properties": {
- "query": {"type": "string"},
- "limit": {"type": "integer"},
- },
- },
- ),
- ),
- ]
-
- # Split tags in extreme positions
- chunks = [
- "searchqueryPython progra", # Complete tag, split value
- "mminglimit10",
- ]
-
- detector = Glm47MoeDetector()
- all_params = ""
-
- for chunk in chunks:
- result = detector.parse_streaming_increment(chunk, tools)
- for call in result.calls:
- if call.parameters:
- all_params += call.parameters
-
- # Verify correct reassembly
- params = json.loads(all_params)
- self.assertEqual(params["query"], "Python programming")
- self.assertEqual(params["limit"], 10)
-
- # Verify no duplication in output
- query_count = all_params.count('"query"')
- limit_count = all_params.count('"limit"')
- self.assertEqual(query_count, 1, "query key duplicated - reprocessing bug")
- self.assertEqual(limit_count, 1, "limit key duplicated - reprocessing bug")
-
- def test_streamed_raw_length_buffer_only_partial_tag(self):
- """
- Test that _streamed_raw_length updates even when state machine returns empty.
-
- Scenario: Send increment that is ONLY a partial opening tag that state machine
- must buffer completely without producing any JSON output.
- Purpose: Force json_increment to be empty string to expose the bug where
- _streamed_raw_length is not updated before early return.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="test_func",
- parameters={
- "type": "object",
- "properties": {
- "key1": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- # Manually call _process_arguments_streaming to have precise control
- detector = Glm47MoeDetector()
- detector.current_tool_id = 0
- detector.current_tool_name_sent = True
- detector._reset_streaming_state()
- detector.streamed_args_for_tool = [""]
- detector._streamed_raw_length = 0
-
- # First call: Complete tag that produces JSON output
- func_args_1 = "key1va"
- result_1 = detector._process_arguments_streaming(
- "test_func", func_args_1, tools
- )
-
- # Should produce JSON output: {"key1": "va (partial)
- self.assertIsNotNone(result_1)
- self.assertGreater(len(result_1.parameters), 0)
- initial_length = detector._streamed_raw_length
- self.assertEqual(initial_length, len(func_args_1))
-
- # Second call: Add just partial closing tag - state machine will buffer this
- # without producing JSON (it's waiting to see if is complete)
- func_args_2 = func_args_1 + "<" # Add partial tag
- result_2 = detector._process_arguments_streaming(
- "test_func", func_args_2, tools
- )
-
- # This is the critical test: if _streamed_raw_length is NOT updated when
- # json_increment is empty, then detector._streamed_raw_length will still be
- # at initial_length, and the next call will reprocess the "<" character
-
- # Check if length was updated (bug test)
- updated_length = detector._streamed_raw_length
-
- # BUG: If code has bug, updated_length will equal initial_length
- # FIXED: If code is correct, updated_length should equal len(func_args_2)
- self.assertEqual(
- updated_length,
- len(func_args_2),
- "Bug detected: _streamed_raw_length not updated when json_increment is empty. "
- f"Expected {len(func_args_2)}, got {updated_length}",
- )
-
- def test_streamed_raw_length_multiple_empty_returns(self):
- """
- Test consecutive chunks that produce empty json_increment.
-
- Scenario: Multiple consecutive chunks that all result in empty json_increment
- as the state machine buffers complex nested structures.
- Purpose: Verify _streamed_raw_length advances correctly through multiple
- empty-return cycles without getting stuck or reprocessing.
- """
- tools = [
- Tool(
- type="function",
- function=Function(
- name="update_settings",
- parameters={
- "type": "object",
- "properties": {
- "name": {"type": "string"},
- "value": {"type": "string"},
- },
- },
- ),
- ),
- ]
-
- # Split XML at positions that may cause state machine buffering
- chunks = [
- "update_settingsna", # Split in tag name
- "meco", # Complete tag start, split value # codespell:ignore ue
- "nf", # Continue value
- "ig_v1val", # Complete value, split next key
- "ueena", # Complete key name, split value # codespell:ignore ue
- "bled", # Complete everything
- ]
-
- detector = Glm47MoeDetector()
- all_params = ""
-
- for i, chunk in enumerate(chunks):
- result = detector.parse_streaming_increment(chunk, tools)
-
- for call in result.calls:
- if call.parameters:
- all_params += call.parameters
-
- # Verify final output is correct
- self.assertGreater(len(all_params), 0, "Should have generated some parameters")
- params = json.loads(all_params)
- self.assertEqual(params["name"], "config_v1")
- self.assertEqual(params["value"], "enabled")
-
- # Verify no duplicate keys due to reprocessing
- name_count = all_params.count('"name"')
- value_count = all_params.count('"value"')
- self.assertEqual(
- name_count,
- 1,
- f"'name' appears {name_count} times - indicates reprocessing bug",
- )
- self.assertEqual(
- value_count,
- 1,
- f"'value' appears {value_count} times - indicates reprocessing bug",
- )
-
-
-class TestGlm4ComplexJsonSchema(unittest.TestCase):
- """Test complex JSON Schema type inference for GLM function call parsers."""
-
- def setUp(self):
- """Set up test tools with complex JSON schemas."""
- self.tools_with_complex_schema = [
- Tool(
- type="function",
- function=Function(
- name="search",
- description="Search for information",
- parameters={
- "type": "object",
- "properties": {
- "query": {
- "description": "Search query, can be a string or a complex object",
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {
- "text": {"type": "string"},
- "filters": {"type": "object"},
- },
- },
- ],
- },
- "priority": {"enum": ["low", "medium", "high"]},
- "options": {
- "oneOf": [{"type": "string"}, {"type": "number"}]
- },
- "config": {
- "allOf": [
- {"type": "object"},
- {"properties": {"timeout": {"type": "number"}}},
- ]
- },
- "tags": {"type": ["string", "null"]},
- "data": {
- "type": "object",
- "properties": {
- "nested": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {
- "value": {"type": "string"}
- },
- },
- ]
- }
- },
- },
- },
- "required": ["query"],
- },
- ),
- ),
- Tool(
- type="function",
- function=Function(
- name="get_weather",
- description="Get weather information",
- parameters={
- "type": "object",
- "properties": {
- "location": {
- "type": "string",
- "description": "Location to get weather for",
- },
- "unit": {
- "type": "string",
- "description": "Temperature unit",
- "enum": ["celsius", "fahrenheit"],
- },
- },
- "required": ["location"],
- },
- ),
- ),
- ]
- self.glm4_detector = Glm4MoeDetector()
- self.glm47_detector = Glm47MoeDetector()
-
- def test_get_argument_type_simple_type(self):
- """Test that get_argument_type correctly handles simple type fields."""
- result = get_argument_type(
- "get_weather", "location", self.tools_with_complex_schema
- )
- self.assertEqual(result, "string")
-
- def test_get_argument_type_enum_type(self):
- """Test that get_argument_type correctly identifies enum as string type."""
- result = get_argument_type(
- "get_weather", "unit", self.tools_with_complex_schema
- )
- # Current implementation returns the direct type field, which is "string" for the enum parameter
- # But it doesn't handle enum-only schemas properly (without type field)
- self.assertEqual(result, "string")
-
- def test_get_argument_type_anyof_type(self):
- """Test that get_argument_type correctly handles anyOf type fields."""
- result = get_argument_type("search", "query", self.tools_with_complex_schema)
- # anyOf with [{"type": "string"}, {"type": "object", ...}] should return "string"
- self.assertEqual(result, "string") # Returns first common type
-
- def test_get_argument_type_oneof_type(self):
- """Test that get_argument_type correctly handles oneOf type fields."""
- result = get_argument_type("search", "options", self.tools_with_complex_schema)
- # oneOf with [{"type": "string"}, {"type": "number"}] should return "string" (prioritizes string)
- self.assertEqual(result, "string")
-
- def test_get_argument_type_allof_type(self):
- """Test that get_argument_type correctly handles allOf type fields."""
- result = get_argument_type("search", "config", self.tools_with_complex_schema)
- # allOf with [{"type": "object"}, ...] should return "object"
- self.assertEqual(result, "object")
-
- def test_get_argument_type_type_array(self):
- """Test that get_argument_type correctly handles type arrays."""
- result = get_argument_type("search", "tags", self.tools_with_complex_schema)
- # Type arrays should return the first non-null type
- self.assertEqual(
- result, "string"
- ) # ["string", "null"] -> "string" (non-null type)
-
- def test_glm4_detector_with_complex_schema_anyof(self):
- """Test GLM4 detector with anyOf schema - should demonstrate current issues."""
- # This test shows the current behavior with complex schemas
- text = (
- "search\n"
- "query\nHello world\n"
- "priority\nmedium\n"
- ""
- )
- result = self.glm4_detector.detect_and_parse(
- text, self.tools_with_complex_schema
- )
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "search")
-
- # Parse parameters to check if they are correctly handled
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "Hello world")
- self.assertEqual(params["priority"], "medium")
-
- def test_glm47_detector_with_complex_schema_anyof(self):
- """Test GLM47 detector with anyOf schema - should demonstrate current issues."""
- # This test shows the current behavior with complex schemas
- text = (
- "search"
- "queryHello world"
- "prioritymedium"
- ""
- )
- result = self.glm47_detector.detect_and_parse(
- text, self.tools_with_complex_schema
- )
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "search")
-
- # Parse parameters to check if they are correctly handled
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "Hello world")
- self.assertEqual(params["priority"], "medium")
-
- def test_glm4_detector_with_enum_values(self):
- """Test GLM4 detector with enum values in complex schema."""
- text = (
- "search\n"
- "query\ntest query\n"
- "priority\nhigh\n"
- ""
- )
- result = self.glm4_detector.detect_and_parse(
- text, self.tools_with_complex_schema
- )
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "search")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "test query")
- self.assertEqual(params["priority"], "high")
-
- def test_glm47_detector_with_enum_values(self):
- """Test GLM47 detector with enum values in complex schema."""
- text = (
- "search"
- "querytest query"
- "priorityhigh"
- ""
- )
- result = self.glm47_detector.detect_and_parse(
- text, self.tools_with_complex_schema
- )
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "search")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "test query")
- self.assertEqual(params["priority"], "high")
-
- def test_glm4_detector_streaming_with_complex_schema(self):
- """Test GLM4 detector streaming with complex schema."""
- chunks = [
- "search\n",
- "query\nnested object\n",
- "priority\nlow\n",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.glm4_detector.parse_streaming_increment(
- chunk, self.tools_with_complex_schema
- )
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
-
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "search")
-
- params = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params["query"], "nested object")
- self.assertEqual(params["priority"], "low")
-
- def test_glm47_detector_streaming_with_complex_schema(self):
- """Test GLM47 detector streaming with complex schema."""
- chunks = [
- "search",
- "querynested object",
- "prioritylow",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.glm47_detector.parse_streaming_increment(
- chunk, self.tools_with_complex_schema
- )
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
-
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "search")
-
- params = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params["query"], "nested object")
- self.assertEqual(params["priority"], "low")
-
- def test_type_inference_issue_reproduction(self):
- """Reproduce the issue where complex JSON schemas are not properly handled."""
- # This test demonstrates the current limitations
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="complex_function",
- parameters={
- "type": "object",
- "properties": {
- "complex_param": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"value": {"type": "string"}},
- },
- ]
- },
- "enum_param": {"enum": ["option1", "option2", "option3"]},
- },
- },
- ),
- )
- ]
-
- # Test that get_argument_type returns appropriate types for complex schemas
- anyof_result = get_argument_type(
- "complex_function", "complex_param", complex_tools
- )
- enum_result = get_argument_type("complex_function", "enum_param", complex_tools)
-
- # Verify complex schema types are correctly inferred
- self.assertEqual(anyof_result, "string") # anyOf prioritizes string type
- self.assertEqual(enum_result, "string") # enum values are strings
-
- def test_expected_behavior_for_complex_schemas(self):
- """Test cases that should work but currently fail - demonstrating the issue."""
- # This test shows what the behavior SHOULD be after the fix
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="complex_function",
- parameters={
- "type": "object",
- "properties": {
- "complex_param": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"value": {"type": "string"}},
- },
- ]
- },
- "enum_param": {"enum": ["option1", "option2", "option3"]},
- "oneof_param": {
- "oneOf": [{"type": "string"}, {"type": "number"}]
- },
- "allof_param": {
- "allOf": [
- {"type": "object"},
- {"properties": {"timeout": {"type": "number"}}},
- ]
- },
- },
- },
- ),
- )
- ]
-
- # These assertions represent the EXPECTED behavior after implementing RFC improvements
- # Currently they will fail, demonstrating the issue
- anyof_result = get_argument_type(
- "complex_function", "complex_param", complex_tools
- )
- enum_result = get_argument_type("complex_function", "enum_param", complex_tools)
- oneof_result = get_argument_type(
- "complex_function", "oneof_param", complex_tools
- )
- allof_result = get_argument_type(
- "complex_function", "allof_param", complex_tools
- )
-
- # These should pass after implementing the RFC improvements, but will currently fail
- # This demonstrates the issue exists
- self.assertIsNotNone(
- anyof_result, "anyOf should return a type after RFC implementation"
- )
- self.assertEqual(
- enum_result,
- "string",
- "enum should return 'string' type after RFC implementation",
- )
- self.assertIsNotNone(
- oneof_result, "oneOf should return a type after RFC implementation"
- )
- self.assertIsNotNone(
- allof_result, "allOf should return a type after RFC implementation"
- )
-
- def test_complex_schema_type_inference_scenarios(self):
- """Test various complex schema scenarios mentioned in the RFC."""
- # Create tools with different complex schema structures
- complex_schema_tools = [
- Tool(
- type="function",
- function=Function(
- name="search_complex",
- parameters={
- "type": "object",
- "properties": {
- # anyOf example - parameter can be string or object
- "query": {
- "description": "Search query, can be a string or a complex object",
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {
- "text": {"type": "string"},
- "filters": {"type": "object"},
- },
- },
- ],
- },
- # oneOf example - parameter must be one of the specified types
- "priority": {
- "oneOf": [{"type": "string"}, {"type": "integer"}]
- },
- # enum example - parameter must be one of the enum values
- "category": {"enum": ["news", "sports", "tech"]},
- # allOf example - parameter must satisfy all schemas
- "config": {
- "allOf": [
- {"type": "object"},
- {"properties": {"timeout": {"type": "number"}}},
- ]
- },
- # Type array example
- "tags": {"type": ["string", "null"]},
- },
- },
- ),
- ),
- Tool(
- type="function",
- function=Function(
- name="get_data",
- parameters={
- "type": "object",
- "properties": {
- # Complex nested anyOf
- "input": {
- "anyOf": [
- {"type": "string"},
- {"type": "number"},
- {
- "type": "object",
- "properties": {
- "type": {"type": "string"},
- "value": {},
- },
- },
- ]
- }
- },
- },
- ),
- ),
- ]
-
- # Test each complex type scenario
- query_type = get_argument_type("search_complex", "query", complex_schema_tools)
- priority_type = get_argument_type(
- "search_complex", "priority", complex_schema_tools
- )
- category_type = get_argument_type(
- "search_complex", "category", complex_schema_tools
- )
- config_type = get_argument_type(
- "search_complex", "config", complex_schema_tools
- )
- tags_type = get_argument_type("search_complex", "tags", complex_schema_tools)
- input_type = get_argument_type("get_data", "input", complex_schema_tools)
-
- # All of these should return appropriate types according to RFC
- self.assertEqual(query_type, "string") # anyOf: string | object -> string
- self.assertEqual(priority_type, "string") # oneOf: string | integer -> string
- self.assertEqual(
- category_type, "string"
- ) # enum: ["news", "sports", "tech"] -> string
- self.assertEqual(config_type, "object") # allOf with object -> object
- self.assertEqual(
- tags_type, "string"
- ) # type array: ["string", "null"] -> string
- self.assertEqual(
- input_type, "string"
- ) # nested anyOf: string | number | object -> string
-
- def test_glm4_detector_type_handling_with_complex_schema(self):
- """Test how GLM4 detector handles type inference for complex schemas in practice."""
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="complex_search",
- parameters={
- "type": "object",
- "properties": {
- "query": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"text": {"type": "string"}},
- },
- ]
- },
- "category": {"enum": ["tech", "news", "sports"]},
- },
- },
- ),
- )
- ]
-
- # Test with string value for anyOf parameter
- text = (
- "complex_search\n"
- "query\ntest search\n"
- "category\ntech\n"
- ""
- )
- result = self.glm4_detector.detect_and_parse(text, complex_tools)
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "complex_search")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "test search")
- self.assertEqual(params["category"], "tech")
-
- def test_glm47_detector_type_handling_with_complex_schema(self):
- """Test how GLM47 detector handles type inference for complex schemas in practice."""
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="complex_search",
- parameters={
- "type": "object",
- "properties": {
- "query": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"text": {"type": "string"}},
- },
- ]
- },
- "category": {"enum": ["tech", "news", "sports"]},
- },
- },
- ),
- )
- ]
-
- # Test with string value for anyOf parameter
- text = (
- "complex_search"
- "querytest search"
- "categorytech"
- ""
- )
- result = self.glm47_detector.detect_and_parse(text, complex_tools)
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "complex_search")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["query"], "test search")
- self.assertEqual(params["category"], "tech")
-
- def test_streaming_with_complex_schema_type_inference(self):
- """Test streaming behavior with complex schema type inference."""
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="stream_test",
- parameters={
- "type": "object",
- "properties": {
- "data": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"value": {"type": "string"}},
- },
- ]
- },
- "status": {"enum": ["active", "inactive"]},
- },
- },
- ),
- )
- ]
-
- # Test GLM4 detector streaming
- chunks = [
- "stream_test\n",
- "data\nnested data\n",
- "status\nactive\n",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.glm4_detector.parse_streaming_increment(chunk, complex_tools)
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
-
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "stream_test")
-
- params = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params["data"], "nested data")
- self.assertEqual(params["status"], "active")
-
- def test_streaming_with_complex_schema_type_inference_glm47(self):
- """Test GLM47 streaming behavior with complex schema type inference."""
- complex_tools = [
- Tool(
- type="function",
- function=Function(
- name="stream_test",
- parameters={
- "type": "object",
- "properties": {
- "data": {
- "anyOf": [
- {"type": "string"},
- {
- "type": "object",
- "properties": {"value": {"type": "string"}},
- },
- ]
- },
- "status": {"enum": ["active", "inactive"]},
- },
- },
- ),
- )
- ]
-
- # Test GLM47 detector streaming
- chunks = [
- "stream_test",
- "datanested data",
- "statusactive",
- "",
- ]
- tool_calls = []
- for chunk in chunks:
- result = self.glm47_detector.parse_streaming_increment(chunk, complex_tools)
- for tool_call_chunk in result.calls:
- if (
- hasattr(tool_call_chunk, "tool_index")
- and tool_call_chunk.tool_index is not None
- ):
- while len(tool_calls) <= tool_call_chunk.tool_index:
- tool_calls.append({"name": "", "parameters": ""})
- tc = tool_calls[tool_call_chunk.tool_index]
- if tool_call_chunk.name:
- tc["name"] = tool_call_chunk.name
- if tool_call_chunk.parameters:
- tc["parameters"] += tool_call_chunk.parameters
-
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "stream_test")
-
- params = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params["data"], "nested data")
- self.assertEqual(params["status"], "active")
-
- if __name__ == "__main__":
- unittest.main()
diff --git a/test/registered/unit/mem_cache/test_streaming_session_unit.py b/test/registered/unit/mem_cache/test_streaming_session_unit.py
index 3caee6887..b4ac9e8c3 100644
--- a/test/registered/unit/mem_cache/test_streaming_session_unit.py
+++ b/test/registered/unit/mem_cache/test_streaming_session_unit.py
@@ -4,7 +4,6 @@ import torch
from sglang.srt.managers.schedule_batch import FINISH_ABORT
from sglang.srt.mem_cache.base_prefix_cache import MatchResult
-from sglang.srt.mem_cache.common import release_kv_cache
from sglang.srt.session.streaming_session import SessionSlot, StreamingSession
from sglang.test.ci.ci_register import register_cpu_ci
@@ -76,6 +75,7 @@ class _FakeReq:
self.pop_overallocated_calls = 0
self.to_finish = None
self.finished_reason = None
+ self.finished_len = None
def pop_committed_kv_cache(self):
assert not self.kv_committed_freed
@@ -89,36 +89,6 @@ class _FakeReq:
return self.kv_committed_len, self.kv_allocated_len
-def test_streaming_release_kv_cache_defers_tail_free(monkeypatch):
- """Spec tail is NOT trimmed in cache_finished_req; it is deferred to
- match_prefix's orphan tail free on the next turn. cache_finished_req
- only sets bookkeeping flags and saves the slot as-is."""
- page_size = 16
- req_to_token = torch.arange(128, dtype=torch.int32).reshape(1, 128)
- req_to_token_pool = SimpleNamespace(req_to_token=req_to_token, free_slots=[])
- allocator = _FakeAllocator()
- tree_cache = StreamingSession(
- _FakeInnerCache(req_to_token_pool, allocator, page_size)
- )
- req = _FakeReq("session-a", req_pool_idx=0, committed=17, allocated=40)
-
- monkeypatch.setattr(
- "sglang.srt.mem_cache.common.get_global_server_args",
- lambda: SimpleNamespace(page_size=page_size, speculative_algorithm="eagle"),
- )
-
- release_kv_cache(req, tree_cache)
-
- slot = tree_cache.slots["session-a"]
- assert req.kv_committed_freed is True
- assert req.kv_overallocated_freed is True
- assert req.req_pool_idx is None
- # Slot keeps the full allocation — tail free is deferred to match_prefix.
- assert slot.kv_committed_len == 17
- assert slot.kv_allocated_len == 40
- assert len(allocator.freed) == 0
-
-
def test_preabort_detaches_session_and_preserves_slot():
"""Pre-aborted req (to_finish set before match_prefix) is detached from
the session: session=None, abort_req() called. Slot stays intact."""
@@ -271,3 +241,11 @@ def test_trim_overshoot_postcondition():
# Tail [38, 44) freed by _free_kv_aligned.
assert len(allocator.freed) == 1
assert allocator.freed[0].tolist() == list(range(38, 44))
+
+
+if __name__ == "__main__":
+ import sys
+
+ import pytest
+
+ sys.exit(pytest.main([__file__, "-v"]))
diff --git a/test/registered/unit/model_executor/test_model_hooks.py b/test/registered/unit/model_executor/test_model_hooks.py
deleted file mode 100644
index 33c9ee82a..000000000
--- a/test/registered/unit/model_executor/test_model_hooks.py
+++ /dev/null
@@ -1,156 +0,0 @@
-import argparse
-import json
-
-import torch
-import torch.nn as nn
-
-from sglang.srt.model_executor.hook_manager import register_forward_hooks
-from sglang.srt.server_args import ServerArgs
-from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
-from sglang.test.test_utils import CustomTestCase
-
-register_cuda_ci(est_time=6, suite="stage-b-test-1-gpu-small")
-register_amd_ci(est_time=10, suite="stage-b-test-1-gpu-small-amd")
-
-HOOK_CALLS = []
-
-
-def dummy_hook_factory(config):
- """Factory that returns a forward hook capturing a tag from config."""
- tag = config.get("tag", "default")
-
- def hook(module, inputs, output):
- HOOK_CALLS.append(
- {
- "module_type": type(module).__name__,
- "tag": tag,
- "shape": tuple(output.shape),
- }
- )
- return output
-
- return hook
-
-
-class TinyModel(nn.Module):
- def __init__(self):
- super().__init__()
- self.inner = nn.Sequential(
- nn.Linear(4, 2),
- nn.ReLU(),
- )
- self.outer = nn.Sequential(
- nn.Linear(4, 4),
- nn.ReLU(),
- self.inner,
- )
-
- def forward(self, x):
- return self.outer(x)
-
-
-class TestAttachHooks(CustomTestCase):
- """Tests for register_forward_hooks / resolve_callable integration."""
-
- def setUp(self):
- HOOK_CALLS.clear()
-
- def test_hook_is_attached(self):
- """Hook from a factory string is registered and fired."""
- hook_specs = [
- {
- "target_modules": ["outer.0", "outer.1"],
- "hook_factory": "test_model_hooks:dummy_hook_factory",
- "config": {"tag": "forward-ok"},
- },
- {
- "target_modules": ["inner.*"],
- "hook_factory": "test_model_hooks:dummy_hook_factory",
- "config": {"tag": "forward-ok"},
- },
- ]
-
- model = TinyModel()
- register_forward_hooks(model, hook_specs)
-
- x = torch.randn(3, 4)
- _ = model(x)
-
- self.assertEqual(
- len(HOOK_CALLS),
- 4,
- "Forward hook was not called correct number of times",
- )
- tags = {call["tag"] for call in HOOK_CALLS}
- self.assertIn("forward-ok", tags)
-
- def test_no_matching_modules_does_not_crash(self):
- """Hook spec with no matching modules should not crash."""
- model = TinyModel()
- hook_specs = [
- {
- "name": "no_match",
- "target_modules": ["does_not_exist.*"],
- "hook_factory": "test_model_hooks:dummy_hook_factory",
- "config": {"tag": "unused"},
- }
- ]
-
- register_forward_hooks(model, hook_specs)
-
- x = torch.randn(3, 4)
- _ = model(x)
-
- # No hooks should have fired
- self.assertEqual(len(HOOK_CALLS), 0)
-
- def test_cli_hooks_reach_model(self):
- """
- Ensure that when hooks are provided via CLI, they are parsed into
- ServerArgs, passed to register_forward_hooks, and actually
- run during a forward pass.
- """
- parser = argparse.ArgumentParser()
- ServerArgs.add_cli_args(parser)
-
- hooks_spec = [
- {
- "name": "outer_and_inner_from_cli",
- "target_modules": ["outer.0", "outer.1", "inner.*"],
- "hook_factory": "test_model_hooks:dummy_hook_factory",
- "config": {"tag": "cli-hook"},
- }
- ]
-
- cli_args = [
- "--model-path",
- "Qwen/Qwen2-7B-Instruct", # Dummy value; not used in this test
- "--forward-hooks",
- json.dumps(hooks_spec),
- ]
-
- args = parser.parse_args(cli_args)
- server_args = ServerArgs.from_cli_args(args)
-
- self.assertEqual(server_args.forward_hooks, hooks_spec)
-
- model = TinyModel()
- register_forward_hooks(model, server_args.forward_hooks)
-
- x = torch.randn(3, 4)
- _ = model(x)
-
- # We expect hooks on outer.0, outer.1, inner.0, inner.1 => 4 calls
- self.assertEqual(
- len(HOOK_CALLS),
- 4,
- "CLI-configured hooks did not fire expected number of times",
- )
-
- tags = {call["tag"] for call in HOOK_CALLS}
- self.assertEqual(tags, {"cli-hook"})
-
-
-if __name__ == "__main__":
- pass
- # unittest.main()
diff --git a/test/registered/vlm/test_patch_embed_perf.py b/test/registered/vlm/test_patch_embed_perf.py
deleted file mode 100644
index 052d17095..000000000
--- a/test/registered/vlm/test_patch_embed_perf.py
+++ /dev/null
@@ -1,166 +0,0 @@
-import os
-import statistics
-
-import pytest
-import torch
-import torch.nn as nn
-
-from sglang.srt.models.glm4v import Glm4vVisionPatchEmbed
-from sglang.test.ci.ci_register import register_cuda_ci
-
-register_cuda_ci(est_time=8, suite="stage-b-test-1-gpu-large")
-
-PATCH_SIZE = 14
-TEMPORAL_PATCH_SIZE = 2
-IN_CHANNELS = 3
-HIDDEN_SIZE = 1536
-FLAT_DIM = IN_CHANNELS * TEMPORAL_PATCH_SIZE * PATCH_SIZE * PATCH_SIZE
-
-
-class ReferenceConv3dPatchEmbed(nn.Module):
- def __init__(
- self,
- patch_size=PATCH_SIZE,
- temporal_patch_size=TEMPORAL_PATCH_SIZE,
- in_channels=IN_CHANNELS,
- hidden_size=HIDDEN_SIZE,
- ):
- super().__init__()
- self.patch_size = patch_size
- self.temporal_patch_size = temporal_patch_size
- self.in_channels = in_channels
- self.hidden_size = hidden_size
-
- kernel_size = (temporal_patch_size, patch_size, patch_size)
- self.proj = nn.Conv3d(
- in_channels,
- hidden_size,
- kernel_size=kernel_size,
- stride=kernel_size,
- bias=True,
- )
-
- def forward(self, x: torch.Tensor) -> torch.Tensor:
- x = x.view(
- -1,
- self.in_channels,
- self.temporal_patch_size,
- self.patch_size,
- self.patch_size,
- )
- x = self.proj(x).view(-1, self.hidden_size)
- return x
-
-
-def _build_modules(device: str, dtype: torch.dtype):
- conv_mod = ReferenceConv3dPatchEmbed().to(device=device, dtype=dtype).eval()
- linear_mod = (
- Glm4vVisionPatchEmbed(
- patch_size=PATCH_SIZE,
- temporal_patch_size=TEMPORAL_PATCH_SIZE,
- in_channels=IN_CHANNELS,
- hidden_size=HIDDEN_SIZE,
- )
- .to(device=device, dtype=dtype)
- .eval()
- )
-
- with torch.no_grad():
- linear_mod.proj.weight.copy_(conv_mod.proj.weight)
- linear_mod.proj.bias.copy_(conv_mod.proj.bias)
-
- linear_mod.copy_conv3d_weight_to_linear()
-
- return conv_mod, linear_mod
-
-
-def _benchmark_cuda_module(
- module: nn.Module,
- x: torch.Tensor,
- warmup: int = 50,
- inner_iters: int = 200,
- repeats: int = 10,
-) -> float:
- assert x.is_cuda
- module.eval()
-
- with torch.inference_mode():
- for _ in range(warmup):
- module(x)
- torch.cuda.synchronize()
-
- samples = []
- for _ in range(repeats):
- start = torch.cuda.Event(enable_timing=True)
- end = torch.cuda.Event(enable_timing=True)
-
- start.record()
- for _ in range(inner_iters):
- module(x)
- end.record()
-
- torch.cuda.synchronize()
- samples.append(start.elapsed_time(end) / inner_iters)
-
- return statistics.median(samples)
-
-
-def test_patch_embed_linear_matches_conv3d():
- torch.manual_seed(0)
-
- device = "cpu"
- dtype = torch.float32
-
- conv_mod, linear_mod = _build_modules(device=device, dtype=dtype)
-
- x = torch.randn(512, FLAT_DIM, device=device, dtype=dtype)
-
- with torch.inference_mode():
- y_conv = conv_mod(x)
- y_linear = linear_mod(x)
-
- torch.testing.assert_close(
- y_conv,
- y_linear,
- rtol=1e-5,
- atol=1e-5,
- )
-
-
-@pytest.mark.skipif(
- not torch.cuda.is_available(), reason="CUDA is required for perf benchmark"
-)
-def test_patch_embed_linear_conv3d():
- torch.manual_seed(0)
- torch.backends.cudnn.benchmark = True
-
- device = "cuda"
- dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
-
- conv_mod, linear_mod = _build_modules(device=device, dtype=dtype)
-
- num_patches = int(os.getenv("GLM4V_NUM_PATCHES", "4096"))
- warmup = int(os.getenv("GLM4V_WARMUP", "50"))
- inner_iters = int(os.getenv("GLM4V_INNER_ITERS", "200"))
- repeats = int(os.getenv("GLM4V_REPEATS", "10"))
-
- x = torch.randn(num_patches, FLAT_DIM, device=device, dtype=dtype).contiguous()
-
- conv_ms = _benchmark_cuda_module(
- conv_mod, x, warmup=warmup, inner_iters=inner_iters, repeats=repeats
- )
- linear_ms = _benchmark_cuda_module(
- linear_mod, x, warmup=warmup, inner_iters=inner_iters, repeats=repeats
- )
-
- speedup = conv_ms / linear_ms
- print(
- f"\n[patch_embed perf] conv3d={conv_ms:.4f} ms | "
- f"linear={linear_ms:.4f} ms | speedup={speedup:.3f}x"
- )
-
- min_speedup = float(os.getenv("GLM4V_MIN_SPEEDUP", "1.00"))
- assert speedup >= min_speedup, (
- f"Expected speedup >= {min_speedup:.3f}x, but got {speedup:.3f}x "
- f"(conv3d={conv_ms:.4f} ms, linear={linear_ms:.4f} ms)"
- )