test: move backend unit tests (#3958)

Signed-off-by: pvijayakrish <[email protected]>

Author: pvijayakrish

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

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""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
+import sys
+
+import pytest
+
+
+def pytest_ignore_collect(collection_path, config):
+    """Skip collecting sglang test files if sglang module isn't installed.
+    Checks test file naming pattern: test_sglang_*.py
+    """
+    filename = collection_path.name
+    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 sglang 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/sglang/tests/test_sglang_prometheus_utils.py

@@ -0,0 +1,86 @@
+# 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 renamed to components/src/dynamo/sglang/tests/test_sglang_unit.py

@@ -10,13 +10,15 @@
 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

@@ -0,0 +1,43 @@
+# 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_trtllm_main_init.py renamed to components/src/dynamo/trtllm/tests/test_trtllm_main_init.py

@@ -0,0 +1,111 @@
+# 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 trtllm_registry(self):
+        """Create a mock registry with TensorRT-LLM-style metrics (no existing prefixes)."""
+        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 request_latency_seconds Request latency in seconds
+# TYPE request_latency_seconds histogram
+request_latency_seconds_bucket{le="0.1"} 10.0
+request_latency_seconds_count 25.0
+# HELP num_requests_running Number of requests currently running
+# TYPE num_requests_running gauge
+num_requests_running 3.0
+# HELP tokens_per_second Tokens generated per second
+# TYPE tokens_per_second gauge
+tokens_per_second 245.7
+"""
+
+        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_trtllm_use_case(self, trtllm_registry):
+        """Test TensorRT-LLM use case: exclude python_/process_ and add trtllm: prefix."""
+        result = get_prometheus_expfmt(
+            trtllm_registry,
+            exclude_prefixes=["python_", "process_"],
+            add_prefix="trtllm:",
+        )
+
+        # Should not contain excluded metrics
+        assert "python_gc_objects_collected_total" not in result
+        assert "process_cpu_seconds_total" not in result
+
+        # All remaining metrics should have trtllm: prefix
+        assert "trtllm:request_latency_seconds" in result
+        assert "trtllm:num_requests_running" in result
+        assert "trtllm:tokens_per_second" in result
+
+        # HELP/TYPE comments should have prefix
+        assert "# HELP trtllm:request_latency_seconds" in result
+        assert "# TYPE trtllm:num_requests_running" in result
+
+        # Check specific content and structure preservation
+        assert 'trtllm:request_latency_seconds_bucket{le="0.1"} 10.0' in result
+        assert "trtllm:tokens_per_second 245.7" in result
+        assert result.endswith("\n")
+
+    def test_no_filtering_all_frameworks(self, trtllm_registry):
+        """Test that without any filters, all metrics are returned."""
+        result = get_prometheus_expfmt(trtllm_registry)
+
+        # Should contain all metrics including excluded ones
+        assert "python_gc_objects_collected_total" in result
+        assert "process_cpu_seconds_total" in result
+        assert "request_latency_seconds" in result
+        assert "num_requests_running" in result
+        assert result.endswith("\n")
+
+    def test_empty_result_handling(self, trtllm_registry):
+        """Test handling when all metrics are filtered out."""
+        result = get_prometheus_expfmt(
+            trtllm_registry,
+            exclude_prefixes=["python_", "process_", "request_", "num_", "tokens_"],
+        )
+
+        # Should return empty string with newline or just newline
+        assert result == "\n" or result == ""
+
+    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 == ""

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

@@ -8,13 +8,16 @@
 
 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,

tests/unit/test_trtllm_unit.py renamed to components/src/dynamo/trtllm/tests/test_trtllm_unit.py

@@ -0,0 +1,43 @@
+# 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