test: move backend unit tests (#3958)

Signed-off-by: pvijayakrish <pvijayakrish@nvidia.com>

tests/unit/conftest.py → components/src/dynamo/sglang/tests/conftest.py

 # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Conftest for dynamo backend unit tests.
-
-Handles conditional test collection to prevent import errors when backend
-frameworks are not installed in the current container.
+"""Conftest for dynamo.sglang unit tests only.
+Handles conditional test collection to prevent import errors when the sglang
+framework is not installed in the current container.
 """
 
 import importlib.util
@@ -14,62 +13,29 @@ import pytest
 
 
 def pytest_ignore_collect(collection_path, config):
-    """Skip collecting backend test files if their framework isn't installed.
-
-    Checks test file naming pattern: test_<backend>_*.py
+    """Skip collecting sglang test files if sglang module isn't installed.
+    Checks test file naming pattern: test_sglang_*.py
     """
     filename = collection_path.name
-
-    # Map test file prefixes to required modules
-    backend_requirements = {
-        "test_vllm_": "vllm",
-        "test_sglang_": "sglang",
-        "test_trtllm_": "tensorrt_llm",
-    }
-
-    for prefix, required_module in backend_requirements.items():
-        if filename.startswith(prefix):
-            if importlib.util.find_spec(required_module) is None:
-                return True  # Module not available, skip this file
-
-    return None  # Not a backend test or module available
+    if filename.startswith("test_sglang_"):
+        if importlib.util.find_spec("sglang") is None:
+            return True  # sglang not available, skip this file
+    return None
 
 
 def make_cli_args_fixture(module_name: str):
-    """Create a pytest fixture for mocking CLI arguments for a backend.
-
-    The returned fixture supports two call styles:
-
-    1. Explicit:
-        mock_vllm_cli("--model", "gpt-2", "--custom-jinja-template", "path.jinja")
-
-    2. Kwargs (auto-converts underscores to hyphens):
-        mock_vllm_cli(model="gpt-2", custom_jinja_template="path.jinja")
-
-    Both produce: ["dynamo.vllm", "--model", "gpt-2", "--custom-jinja-template", "path.jinja"]
-
-    Args:
-        module_name: Module identifier for argv[0] (e.g., "dynamo.vllm")
-
-    Returns:
-        Pytest fixture that mocks sys.argv via monkeypatch
-    """
+    """Create a pytest fixture for mocking CLI arguments for sglang backend."""
 
     @pytest.fixture
     def mock_cli_args(monkeypatch):
-        """Mock sys.argv with CLI arguments (explicit or kwargs style)."""
-
         def set_args(*args, **kwargs):
             if args:
-                # Explicit style: pass argv elements directly
                 argv = [module_name, *args]
             else:
-                # Kwargs style: convert python arg to CLI arg
                 argv = [module_name]
                 for param_name, param_value in kwargs.items():
                     cli_flag = f"--{param_name.replace('_', '-')}"
                     argv.extend([cli_flag, str(param_value)])
-
             monkeypatch.setattr(sys, "argv", argv)
 
         return set_args

components/src/dynamo/sglang/tests/test_sglang_prometheus_utils.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for Prometheus utilities."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from dynamo.common.utils.prometheus import get_prometheus_expfmt
+
+pytestmark = [
+    pytest.mark.unit,
+]
+
+
+class TestGetPrometheusExpfmt:
+    """Test class for get_prometheus_expfmt function."""
+
+    @pytest.fixture
+    def sglang_registry(self):
+        """Create a mock registry with SGLang-style metrics."""
+        registry = Mock()
+
+        sample_metrics = """# HELP python_gc_objects_collected_total Objects collected during gc
+# TYPE python_gc_objects_collected_total counter
+python_gc_objects_collected_total{generation="0"} 123.0
+# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds
+# TYPE process_cpu_seconds_total counter
+process_cpu_seconds_total 45.6
+# HELP sglang:prompt_tokens_total Number of prefill tokens processed
+# TYPE sglang:prompt_tokens_total counter
+sglang:prompt_tokens_total{model_name="meta-llama/Llama-3.1-8B-Instruct"} 8128902.0
+# HELP sglang:generation_tokens_total Number of generation tokens processed
+# TYPE sglang:generation_tokens_total counter
+sglang:generation_tokens_total{model_name="meta-llama/Llama-3.1-8B-Instruct"} 7557572.0
+# HELP sglang:cache_hit_rate The cache hit rate
+# TYPE sglang:cache_hit_rate gauge
+sglang:cache_hit_rate{model_name="meta-llama/Llama-3.1-8B-Instruct"} 0.0075
+"""
+
+        def mock_generate_latest(reg):
+            return sample_metrics.encode("utf-8")
+
+        import dynamo.common.utils.prometheus
+
+        original_generate_latest = dynamo.common.utils.prometheus.generate_latest
+        dynamo.common.utils.prometheus.generate_latest = mock_generate_latest
+
+        yield registry
+
+        dynamo.common.utils.prometheus.generate_latest = original_generate_latest
+
+    def test_sglang_use_case(self, sglang_registry):
+        """Test SGLang use case: filter to sglang: metrics and exclude python_/process_."""
+        result = get_prometheus_expfmt(
+            sglang_registry,
+            metric_prefix_filter="sglang:",
+            exclude_prefixes=["python_", "process_"],
+        )
+
+        # Should only contain sglang: metrics
+        assert "sglang:prompt_tokens_total" in result
+        assert "sglang:generation_tokens_total" in result
+        assert "sglang:cache_hit_rate" in result
+        assert "# HELP sglang:prompt_tokens_total" in result
+
+        # Should not contain excluded metrics
+        assert "python_gc_objects_collected_total" not in result
+        assert "process_cpu_seconds_total" not in result
+
+        # Check specific content
+        assert 'model_name="meta-llama/Llama-3.1-8B-Instruct"' in result
+        assert "8128902.0" in result  # prompt tokens value
+        assert result.endswith("\n")
+
+    def test_error_handling(self):
+        """Test error handling when registry fails."""
+        # Create a registry that raises an exception
+        bad_registry = Mock()
+        bad_registry.side_effect = Exception("Registry error")
+
+        result = get_prometheus_expfmt(bad_registry)
+
+        # Should return empty string on error
+        assert result == ""

tests/unit/test_sglang_unit.py → components/src/dynamo/sglang/tests/test_sglang_unit.py

@@ -10,13 +10,15 @@ from pathlib import Path
 import pytest
 
 from dynamo.sglang.args import parse_args
-from tests.unit.conftest import make_cli_args_fixture
+from dynamo.sglang.tests.conftest import make_cli_args_fixture
 
 # Get path relative to this test file
-TEST_DIR = Path(__file__).parent.parent
-JINJA_TEMPLATE_PATH = str(TEST_DIR / "serve" / "fixtures" / "custom_template.jinja")
-
-
+REPO_ROOT = Path(__file__).resolve().parents[5]
+TEST_DIR = REPO_ROOT / "tests"
+# Now construct the full path to the shared test fixture
+JINJA_TEMPLATE_PATH = str(
+    REPO_ROOT / "tests" / "serve" / "fixtures" / "custom_template.jinja"
+)
 pytestmark = [
     pytest.mark.unit,
     pytest.mark.sglang,

components/src/dynamo/trtllm/tests/conftest.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Conftest for dynamo.trtllm unit tests only.
+Handles conditional test collection to prevent import errors when the tensorrt_llm
+framework is not installed in the current container.
+"""
+
+import importlib.util
+import sys
+
+import pytest
+
+
+def pytest_ignore_collect(collection_path, config):
+    """Skip collecting trtllm test files if tensorrt_llm module isn't installed.
+    Checks test file naming pattern: test_trtllm_*.py
+    """
+    filename = collection_path.name
+    if filename.startswith("test_trtllm_"):
+        if importlib.util.find_spec("tensorrt_llm") is None:
+            return True  # tensorrt_llm not available, skip this file
+    return None
+
+
+def make_cli_args_fixture(module_name: str):
+    """Create a pytest fixture for mocking CLI arguments for trtllm backend."""
+
+    @pytest.fixture
+    def mock_cli_args(monkeypatch):
+        def set_args(*args, **kwargs):
+            if args:
+                argv = [module_name, *args]
+            else:
+                argv = [module_name]
+                for param_name, param_value in kwargs.items():
+                    cli_flag = f"--{param_name.replace('_', '-')}"
+                    argv.extend([cli_flag, str(param_value)])
+            monkeypatch.setattr(sys, "argv", argv)
+
+        return set_args
+
+    return mock_cli_args

tests/unit/test_prometheus_utils.py → components/src/dynamo/trtllm/tests/test_trtllm_prometheus_utils.py

@@ -17,72 +17,6 @@ pytestmark = [
 class TestGetPrometheusExpfmt:
     """Test class for get_prometheus_expfmt function."""
 
-    @pytest.fixture
-    def vllm_registry(self):
-        """Create a mock registry with vLLM-style metrics."""
-        registry = Mock()
-
-        sample_metrics = """# HELP python_gc_objects_collected_total Objects collected during gc
-# TYPE python_gc_objects_collected_total counter
-python_gc_objects_collected_total{generation="0"} 123.0
-# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds
-# TYPE process_cpu_seconds_total counter
-process_cpu_seconds_total 45.6
-# HELP vllm:request_success_total Number of successfully finished requests
-# TYPE vllm:request_success_total counter
-vllm:request_success_total{finished_reason="stop",model_name="meta-llama/Llama-3.1-8B"} 150.0
-# HELP vllm:time_to_first_token_seconds Histogram of time to first token in seconds
-# TYPE vllm:time_to_first_token_seconds histogram
-vllm:time_to_first_token_seconds_bucket{le="0.005",model_name="meta-llama/Llama-3.1-8B"} 5.0
-vllm:time_to_first_token_seconds_count{model_name="meta-llama/Llama-3.1-8B"} 165.0
-"""
-
-        def mock_generate_latest(reg):
-            return sample_metrics.encode("utf-8")
-
-        import dynamo.common.utils.prometheus
-
-        original_generate_latest = dynamo.common.utils.prometheus.generate_latest
-        dynamo.common.utils.prometheus.generate_latest = mock_generate_latest
-
-        yield registry
-
-        dynamo.common.utils.prometheus.generate_latest = original_generate_latest
-
-    @pytest.fixture
-    def sglang_registry(self):
-        """Create a mock registry with SGLang-style metrics."""
-        registry = Mock()
-
-        sample_metrics = """# HELP python_gc_objects_collected_total Objects collected during gc
-# TYPE python_gc_objects_collected_total counter
-python_gc_objects_collected_total{generation="0"} 123.0
-# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds
-# TYPE process_cpu_seconds_total counter
-process_cpu_seconds_total 45.6
-# HELP sglang:prompt_tokens_total Number of prefill tokens processed
-# TYPE sglang:prompt_tokens_total counter
-sglang:prompt_tokens_total{model_name="meta-llama/Llama-3.1-8B-Instruct"} 8128902.0
-# HELP sglang:generation_tokens_total Number of generation tokens processed
-# TYPE sglang:generation_tokens_total counter
-sglang:generation_tokens_total{model_name="meta-llama/Llama-3.1-8B-Instruct"} 7557572.0
-# HELP sglang:cache_hit_rate The cache hit rate
-# TYPE sglang:cache_hit_rate gauge
-sglang:cache_hit_rate{model_name="meta-llama/Llama-3.1-8B-Instruct"} 0.0075
-"""
-
-        def mock_generate_latest(reg):
-            return sample_metrics.encode("utf-8")
-
-        import dynamo.common.utils.prometheus
-
-        original_generate_latest = dynamo.common.utils.prometheus.generate_latest
-        dynamo.common.utils.prometheus.generate_latest = mock_generate_latest
-
-        yield registry
-
-        dynamo.common.utils.prometheus.generate_latest = original_generate_latest
-
     @pytest.fixture
     def trtllm_registry(self):
         """Create a mock registry with TensorRT-LLM-style metrics (no existing prefixes)."""
@@ -118,51 +52,6 @@ tokens_per_second 245.7
 
         dynamo.common.utils.prometheus.generate_latest = original_generate_latest
 
-    def test_vllm_use_case(self, vllm_registry):
-        """Test vLLM use case: filter to vllm: metrics and exclude python_/process_."""
-        result = get_prometheus_expfmt(
-            vllm_registry,
-            metric_prefix_filter="vllm:",
-            exclude_prefixes=["python_", "process_"],
-        )
-
-        # Should only contain vllm: metrics
-        assert "vllm:request_success_total" in result
-        assert "vllm:time_to_first_token_seconds" in result
-        assert "# HELP vllm:request_success_total" in result
-
-        # Should not contain excluded metrics
-        assert "python_gc_objects_collected_total" not in result
-        assert "process_cpu_seconds_total" not in result
-
-        # Check specific content
-        assert 'finished_reason="stop"' in result
-        assert 'model_name="meta-llama/Llama-3.1-8B"' in result
-        assert result.endswith("\n")
-
-    def test_sglang_use_case(self, sglang_registry):
-        """Test SGLang use case: filter to sglang: metrics and exclude python_/process_."""
-        result = get_prometheus_expfmt(
-            sglang_registry,
-            metric_prefix_filter="sglang:",
-            exclude_prefixes=["python_", "process_"],
-        )
-
-        # Should only contain sglang: metrics
-        assert "sglang:prompt_tokens_total" in result
-        assert "sglang:generation_tokens_total" in result
-        assert "sglang:cache_hit_rate" in result
-        assert "# HELP sglang:prompt_tokens_total" in result
-
-        # Should not contain excluded metrics
-        assert "python_gc_objects_collected_total" not in result
-        assert "process_cpu_seconds_total" not in result
-
-        # Check specific content
-        assert 'model_name="meta-llama/Llama-3.1-8B-Instruct"' in result
-        assert "8128902.0" in result  # prompt tokens value
-        assert result.endswith("\n")
-
     def test_trtllm_use_case(self, trtllm_registry):
         """Test TensorRT-LLM use case: exclude python_/process_ and add trtllm: prefix."""
         result = get_prometheus_expfmt(

tests/unit/test_trtllm_unit.py → components/src/dynamo/trtllm/tests/test_trtllm_unit.py

@@ -8,13 +8,16 @@ from pathlib import Path
 
 import pytest
 
+from dynamo.trtllm.tests.conftest import make_cli_args_fixture
 from dynamo.trtllm.utils.trtllm_utils import cmd_line_args
-from tests.unit.conftest import make_cli_args_fixture
 
 # Get path relative to this test file
-TEST_DIR = Path(__file__).parent.parent
-JINJA_TEMPLATE_PATH = str(TEST_DIR / "serve" / "fixtures" / "custom_template.jinja")
-
+REPO_ROOT = Path(__file__).resolve().parents[5]
+TEST_DIR = REPO_ROOT / "tests"
+# Now construct the full path to the shared test fixture
+JINJA_TEMPLATE_PATH = str(
+    REPO_ROOT / "tests" / "serve" / "fixtures" / "custom_template.jinja"
+)
 
 pytestmark = [
     pytest.mark.unit,

components/src/dynamo/vllm/tests/conftest.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Conftest for dynamo.vllm unit tests only.
+Handles conditional test collection to prevent import errors when the vllm
+framework is not installed in the current container.
+"""
+
+import importlib.util
+import sys
+
+import pytest
+
+
+def pytest_ignore_collect(collection_path, config):
+    """Skip collecting vllm test files if vllm module isn't installed.
+    Checks test file naming pattern: test_vllm_*.py
+    """
+    filename = collection_path.name
+    if filename.startswith("test_vllm_"):
+        if importlib.util.find_spec("vllm") is None:
+            return True  # vllm not available, skip this file
+    return None
+
+
+def make_cli_args_fixture(module_name: str):
+    """Create a pytest fixture for mocking CLI arguments for vllm backend."""
+
+    @pytest.fixture
+    def mock_cli_args(monkeypatch):
+        def set_args(*args, **kwargs):
+            if args:
+                argv = [module_name, *args]
+            else:
+                argv = [module_name]
+                for param_name, param_value in kwargs.items():
+                    cli_flag = f"--{param_name.replace('_', '-')}"
+                    argv.extend([cli_flag, str(param_value)])
+            monkeypatch.setattr(sys, "argv", argv)
+
+        return set_args
+
+    return mock_cli_args

components/src/dynamo/vllm/tests/test_vllm_prometheus_utils.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for Prometheus utilities."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from dynamo.common.utils.prometheus import get_prometheus_expfmt
+
+pytestmark = [
+    pytest.mark.unit,
+]
+
+
+class TestGetPrometheusExpfmt:
+    """Test class for get_prometheus_expfmt function."""
+
+    @pytest.fixture
+    def vllm_registry(self):
+        """Create a mock registry with vLLM-style metrics."""
+        registry = Mock()
+
+        sample_metrics = """# HELP python_gc_objects_collected_total Objects collected during gc
+# TYPE python_gc_objects_collected_total counter
+python_gc_objects_collected_total{generation="0"} 123.0
+# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds
+# TYPE process_cpu_seconds_total counter
+process_cpu_seconds_total 45.6
+# HELP vllm:request_success_total Number of successfully finished requests
+# TYPE vllm:request_success_total counter
+vllm:request_success_total{finished_reason="stop",model_name="meta-llama/Llama-3.1-8B"} 150.0
+# HELP vllm:time_to_first_token_seconds Histogram of time to first token in seconds
+# TYPE vllm:time_to_first_token_seconds histogram
+vllm:time_to_first_token_seconds_bucket{le="0.005",model_name="meta-llama/Llama-3.1-8B"} 5.0
+vllm:time_to_first_token_seconds_count{model_name="meta-llama/Llama-3.1-8B"} 165.0
+"""
+
+        def mock_generate_latest(reg):
+            return sample_metrics.encode("utf-8")
+
+        import dynamo.common.utils.prometheus
+
+        original_generate_latest = dynamo.common.utils.prometheus.generate_latest
+        dynamo.common.utils.prometheus.generate_latest = mock_generate_latest
+
+        yield registry
+
+        dynamo.common.utils.prometheus.generate_latest = original_generate_latest
+
+    def test_vllm_use_case(self, vllm_registry):
+        """Test vLLM use case: filter to vllm: metrics and exclude python_/process_."""
+        result = get_prometheus_expfmt(
+            vllm_registry,
+            metric_prefix_filter="vllm:",
+            exclude_prefixes=["python_", "process_"],
+        )
+
+        # Should only contain vllm: metrics
+        assert "vllm:request_success_total" in result
+        assert "vllm:time_to_first_token_seconds" in result
+        assert "# HELP vllm:request_success_total" in result
+
+        # Should not contain excluded metrics
+        assert "python_gc_objects_collected_total" not in result
+        assert "process_cpu_seconds_total" not in result
+
+        # Check specific content
+        assert 'finished_reason="stop"' in result
+        assert 'model_name="meta-llama/Llama-3.1-8B"' in result
+        assert result.endswith("\n")
+
+    def test_error_handling(self):
+        """Test error handling when registry fails."""
+        # Create a registry that raises an exception
+        bad_registry = Mock()
+        bad_registry.side_effect = Exception("Registry error")
+
+        result = get_prometheus_expfmt(bad_registry)
+
+        # Should return empty string on error
+        assert result == ""

tests/unit/test_vllm_unit.py → components/src/dynamo/vllm/tests/test_vllm_unit.py

@@ -9,12 +9,15 @@ from pathlib import Path
 import pytest
 
 from dynamo.vllm.args import parse_args
-from tests.unit.conftest import make_cli_args_fixture
+from dynamo.vllm.tests.conftest import make_cli_args_fixture
 
 # Get path relative to this test file
-TEST_DIR = Path(__file__).parent.parent
-JINJA_TEMPLATE_PATH = str(TEST_DIR / "serve" / "fixtures" / "custom_template.jinja")
-
+REPO_ROOT = Path(__file__).resolve().parents[5]
+TEST_DIR = REPO_ROOT / "tests"
+# Now construct the full path to the shared test fixture
+JINJA_TEMPLATE_PATH = str(
+    REPO_ROOT / "tests" / "serve" / "fixtures" / "custom_template.jinja"
+)
 
 pytestmark = [
     pytest.mark.unit,