[Test] Remove more unit tests that mirror implementation or never run in CI

test/registered/unit/configs/test_linear_attn_model_registry.py

@@ -7,7 +7,6 @@
     LinearAttnModelSpec,
     get_linear_attn_config,
     get_linear_attn_spec_by_arch,
-    import_backend_class,
     register_linear_attn_model,
 )
 from sglang.test.ci.ci_register import register_cpu_ci
@@ -138,24 +137,6 @@ def test_first_match_wins(self):
         result = get_linear_attn_config(FakeLinearAttnConfig())
         self.assertIs(result[0], spec1)
 
-    def test_import_backend_class(self):
-        # Import a real stdlib class to verify the mechanism
-        cls = import_backend_class("collections.OrderedDict")
-        from collections import OrderedDict
-
-        self.assertIs(cls, OrderedDict)
-
-    def test_spec_defaults(self):
-        spec = LinearAttnModelSpec(
-            config_class=FakeLinearAttnConfig,
-            backend_class_name="pkg.mod.Cls",
-        )
-        self.assertEqual(spec.arch_names, [])
-        self.assertTrue(spec.uses_mamba_radix_cache)
-        self.assertTrue(spec.support_mamba_cache)
-        self.assertFalse(spec.support_mamba_cache_extra_buffer)
-        self.assertFalse(spec.unwrap_text_config)
-
 
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/constrained/test_base_grammar_backend.py

@@ -2,14 +2,10 @@
 Unit tests for sglang.srt.constrained.base_grammar_backend.
 
 Test Coverage:
-- GrammarStats: default values, mutable default isolation
-- BaseGrammarObject: default behavior
-- InvalidGrammarObject: error message
 - BaseGrammarBackend: caching, dispatch routing, unsupported fallback,
   thread pool execution, cache hit/miss
 - create_grammar_backend: factory routing, "none" backend, invalid name,
   custom registry, reasoner wrapping
-- register_grammar_backend: registration and lookup
 
 Usage:
     python -m pytest test_base_grammar_backend.py -v
@@ -36,44 +32,6 @@
 register_cpu_ci(2.0, "base-a-test-cpu")
 
 
-class TestGrammarStats(unittest.TestCase):
-    """Test GrammarStats dataclass."""
-
-    def test_defaults(self):
-        stats = GrammarStats()
-        self.assertIsNone(stats.compilation_time)
-        self.assertIsNone(stats.schema_count)
-        self.assertIsNone(stats.ebnf_size)
-        self.assertFalse(stats.is_cache_hit)
-        self.assertFalse(stats.is_grammar_aborted)
-        self.assertEqual(stats.tree_traversal_time, [])
-        self.assertIsNone(stats.dispatch_type)
-        self.assertEqual(stats.num_timeout, 0)
-
-    def test_tree_traversal_time_mutable_default(self):
-        """Ensure each instance gets its own list."""
-        s1 = GrammarStats()
-        s2 = GrammarStats()
-        s1.tree_traversal_time.append(0.1)
-        self.assertEqual(len(s2.tree_traversal_time), 0)
-
-
-class TestBaseGrammarObject(unittest.TestCase):
-    """Test BaseGrammarObject base class."""
-
-
-class TestInvalidGrammarObject(unittest.TestCase):
-    """Test InvalidGrammarObject."""
-
-    def test_default_error_message(self):
-        obj = InvalidGrammarObject()
-        self.assertEqual(obj.error_message, "Unknown grammar error")
-
-    def test_custom_error_message(self):
-        obj = InvalidGrammarObject("Regex compilation failed")
-        self.assertEqual(obj.error_message, "Regex compilation failed")
-
-
 class TestBaseGrammarBackend(unittest.TestCase):
     """Test BaseGrammarBackend caching and dispatch."""
 
@@ -206,24 +164,6 @@ def test_cache_miss_duplicate_key_submits_separate_futures(self):
         self.assertIsInstance(result2.result(timeout=5), InvalidGrammarObject)
 
 
-class TestRegisterGrammarBackend(unittest.TestCase):
-    """Test grammar backend registry."""
-
-    def setUp(self):
-        self._saved = dict(GRAMMAR_BACKEND_REGISTRY)
-
-    def tearDown(self):
-        GRAMMAR_BACKEND_REGISTRY.clear()
-        GRAMMAR_BACKEND_REGISTRY.update(self._saved)
-
-    def test_overwrite_registration(self):
-        register_grammar_backend("dup", lambda *a: "first")
-        register_grammar_backend("dup", lambda *a: "second")
-        self.assertEqual(
-            GRAMMAR_BACKEND_REGISTRY["dup"](None, None, None, None), "second"
-        )
-
-
 class TestCreateGrammarBackend(unittest.TestCase):
     """Test create_grammar_backend factory function."""
 
@@ -314,36 +254,6 @@ def test_custom_backend_skips_reasoner_wrapping(self):
         # Custom backends return early, no reasoner wrapping applied
         self.assertIs(result, mock_inner)
 
-    @patch("sglang.srt.constrained.outlines_backend.OutlinesGrammarBackend")
-    def test_outlines_backend(self, mock_outlines_cls):
-        mock_backend = MagicMock(spec=BaseGrammarBackend)
-        mock_outlines_cls.return_value = mock_backend
-        args = self._make_server_args(
-            "outlines", constrained_json_whitespace_pattern=r"\s*"
-        )
-
-        result = create_grammar_backend(args, "tok", 32000)
-        mock_outlines_cls.assert_called_once_with("tok", whitespace_pattern=r"\s*")
-        self.assertIs(result, mock_backend)
-
-    @patch("sglang.srt.constrained.xgrammar_backend.XGrammarGrammarBackend")
-    def test_xgrammar_backend(self, mock_xgrammar_cls):
-        mock_backend = MagicMock(spec=BaseGrammarBackend)
-        mock_xgrammar_cls.return_value = mock_backend
-        args = self._make_server_args(
-            "xgrammar", constrained_json_disable_any_whitespace=True
-        )
-
-        result = create_grammar_backend(args, "tok", 32000, {1, 2})
-        mock_xgrammar_cls.assert_called_once_with(
-            "tok",
-            vocab_size=32000,
-            model_eos_token_ids=[1, 2],
-            any_whitespace=False,
-            max_whitespace_cnt=None,
-        )
-        self.assertIs(result, mock_backend)
-
     @patch("sglang.srt.constrained.xgrammar_backend.XGrammarGrammarBackend")
     def test_xgrammar_unsupported_tokenizer_falls_back_to_none(self, mock_xgrammar_cls):
         from sglang.srt.constrained.xgrammar_backend import TokenizerNotSupportedError
@@ -366,26 +276,6 @@ def test_xgrammar_unsupported_tokenizer_falls_back_to_none(self, mock_xgrammar_c
         # (asserted above). The two are meant to differ here.
         self.assertEqual(resolution_result(server_args, "grammar_backend"), "xgrammar")
 
-    @patch("sglang.srt.constrained.llguidance_backend.GuidanceBackend")
-    def test_llguidance_backend(self, mock_guidance_cls):
-        mock_backend = MagicMock(spec=BaseGrammarBackend)
-        mock_guidance_cls.return_value = mock_backend
-        args = self._make_server_args(
-            "llguidance",
-            constrained_json_disable_any_whitespace=False,
-            constrained_json_whitespace_pattern=r"\s+",
-        )
-
-        result = create_grammar_backend(args, "tok", 32000, {1, 2})
-        mock_guidance_cls.assert_called_once_with(
-            tokenizer="tok",
-            any_whitespace=True,
-            whitespace_pattern=r"\s+",
-            n_vocab=32000,
-            eos_token_ids={1, 2},
-        )
-        self.assertIs(result, mock_backend)
-
     @patch("sglang.srt.constrained.outlines_backend.OutlinesGrammarBackend")
     def test_reasoner_wrapping_on_builtin_backend(self, mock_outlines_cls):
         """Non-custom backends get wrapped with ReasonerGrammarBackend."""

test/registered/unit/constrained/test_grammar_manager.py

@@ -111,18 +111,6 @@ def setUp(self):
 
     """Test GrammarManager initialization."""
 
-    @patch("sglang.srt.constrained.grammar_manager.create_grammar_backend")
-    def test_init_with_backend(self, mock_create):
-        mock_create.return_value = MagicMock(spec=BaseGrammarBackend)
-        scheduler = _make_scheduler("xgrammar")
-        enter_override(
-            self, get_context().override_server_args(skip_tokenizer_init=False)
-        )
-
-        mgr = GrammarManager(scheduler)
-        self.assertIsNotNone(mgr.grammar_backend)
-        self.assertEqual(len(mgr), 0)
-
     def test_init_skip_tokenizer(self):
         scheduler = _make_scheduler(skip_tokenizer=True)
         mgr = GrammarManager(scheduler)
@@ -136,19 +124,6 @@ def test_len_and_has_waiting(self, mock_create):
         self.assertEqual(len(mgr), 0)
         self.assertFalse(mgr.has_waiting_grammars())
 
-    @patch("sglang.srt.constrained.grammar_manager.create_grammar_backend")
-    def test_clear_resets_backend(self, mock_create):
-        mock_backend = MagicMock(spec=BaseGrammarBackend)
-        mock_create.return_value = mock_backend
-        scheduler = _make_scheduler()
-        enter_override(
-            self, get_context().override_server_args(skip_tokenizer_init=False)
-        )
-
-        mgr = GrammarManager(scheduler)
-        mgr.clear()
-        mock_backend.reset.assert_called_once()
-
     @patch("sglang.srt.constrained.grammar_manager.create_grammar_backend")
     def test_clear_no_backend(self, mock_create):
         mock_create.return_value = None

test/registered/unit/disaggregation/test_encode_server.py

@@ -24,7 +24,6 @@
 from sglang.srt.disaggregation.encoder.server import (
     BadRequestError,
     EncodeContext,
-    EncoderDelivery,
     EncoderMetaRegistry,
     InternalError,
     MMEncoder,
@@ -637,16 +636,6 @@ async def fail():
 
         asyncio.run(run())
 
-    def test_contract_has_two_direct_implementations(self):
-        self.assertEqual(EncoderDelivery.__abstractmethods__, {"send", "release"})
-        self.assertEqual(
-            set(EncoderDelivery.__subclasses__()),
-            {
-                MooncakeDelivery,
-                ZmqDelivery,
-            },
-        )
-
     def test_failed_staged_send_releases_request(self):
         async def run():
             encoder = SimpleNamespace(

test/registered/unit/disaggregation/test_fake_kv_sender.py

@@ -146,17 +146,6 @@ def test_abort_sets_failed_state(self):
         self.assertEqual(self.sender.conclude_state, KVPoll.Failed)
         self.assertEqual(self.sender.poll(), KVPoll.Failed)
 
-    def test_get_transfer_metric(self):
-        metric = self.sender.get_transfer_metric()
-        self.assertIsNone(metric.transfer_latency_s)
-        self.assertIsNone(metric.alloc_latency_s)
-        self.assertIsNone(metric.transfer_total_bytes)
-
-    def test_failure_exception(self):
-        with self.assertRaises(Exception) as ctx:
-            self.sender.failure_exception()
-        self.assertIn("Fake KVSender Exception", str(ctx.exception))
-
 
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/disaggregation/test_pd_role_switch.py

@@ -1,4 +1,3 @@
-import argparse
 import concurrent.futures
 import unittest
 from types import SimpleNamespace
@@ -18,18 +17,6 @@
 register_cpu_ci(est_time=5, suite="base-a-test-cpu")
 
 
-class TestPdRoleSwitchServerArg(unittest.TestCase):
-    def test_cli_flag_parses(self):
-        parser = argparse.ArgumentParser()
-        ServerArgs.add_cli_args(parser)
-
-        off = parser.parse_args(["--model-path", "dummy"])
-        self.assertFalse(off.enable_pd_role_switch)
-
-        on = parser.parse_args(["--model-path", "dummy", "--enable-pd-role-switch"])
-        self.assertTrue(on.enable_pd_role_switch)
-
-
 class TestHandlePdRoleSwitch(unittest.TestCase):
     """Cover the control-plane contract of Scheduler.handle_pd_role_switch.
 
@@ -388,14 +375,6 @@ def test_no_role_switch_is_unaffected(self):
 
 import zmq  # noqa: E402
 
-try:
-    from sglang.srt.disaggregation.common.utils import FastQueue  # noqa: E402
-    from sglang.srt.disaggregation.mori.conn import MoriKVManager  # noqa: E402
-
-    _HAS_MORI = True
-except Exception:  # pragma: no cover - environment dependent
-    _HAS_MORI = False
-
 try:
     from sglang.srt.disaggregation.common.utils import (  # noqa: E402,F811
         FastQueue as _FQ,
@@ -417,38 +396,6 @@ def test_no_role_switch_is_unaffected(self):
     _HAS_ROLE_SWITCH = False
 
 
-@unittest.skipUnless(_HAS_MORI, "mori not importable in this environment")
-class TestMoriTeardownNoThreadLeak(unittest.TestCase):
-    """teardown() must stop+join the transfer workers it started, so a P->D->P
-    flip loop does not leak _num_shards transfer threads per cycle."""
-
-    def test_teardown_joins_transfer_workers(self):
-        m = MoriKVManager.__new__(MoriKVManager)
-        m.disaggregation_mode = DisaggregationMode.PREFILL
-        m._stopped = False
-        m._worker_threads = []
-        m._transfer_queues = [FastQueue() for _ in range(3)]
-        m.server_socket = MagicMock()
-        m._zmq_ctx = MagicMock()
-        m.engine = MagicMock()
-        m.kv_mem_descs = m.aux_mem_descs = m.state_mem_descs = []
-        for q in m._transfer_queues:
-            t = threading.Thread(target=m._transfer_worker, args=(q,), daemon=True)
-            t.start()
-            m._worker_threads.append(t)
-        started = list(m._worker_threads)
-        time.sleep(0.05)  # let workers park in FastQueue.get()
-        for t in started:
-            self.assertTrue(t.is_alive())
-
-        MoriKVManager.teardown(m)
-
-        for t in started:
-            self.assertFalse(t.is_alive(), "transfer worker survived teardown (leak)")
-        self.assertEqual(m._worker_threads, [])
-        self.assertEqual(m._transfer_queues, [])
-
-
 @unittest.skipUnless(_HAS_MOONCAKE, "mooncake not importable in this environment")
 class TestMooncakeTeardownNoThreadLeak(unittest.TestCase):
     """teardown() must stop+join the transfer workers it started, so a P->D->P