Add UT for watsonx/passthrough/transformation & watsonx api route

Vibe coded with IBM Bob Signed-off-by: T K Chandra Hasan <t.k.chandra.hasan@ibm.com>
2026-09-26 01:12:21 +00:00 · 2026-05-20 19:33:07 +05:30 · 2026-05-20 19:33:07 +05:30 · 56582e2934
commit 56582e2934
parent eeae9bf6e3
2 changed files with 726 additions and 0 deletions
--- a/tests/test_litellm/llms/watsonx/passthrough/test_watsonx_passthrough_transformation.py
+++ b/tests/test_litellm/llms/watsonx/passthrough/test_watsonx_passthrough_transformation.py
@ -0,0 +1,282 @@
+"""
+Unit tests for WatsonxPassthroughConfig transformation.
+
+Tests the Watsonx-specific passthrough configuration including URL construction,
+streaming detection, and authentication handling.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../.."))
+
+import litellm
+from litellm.llms.watsonx.passthrough.transformation import WatsonxPassthroughConfig
+
+
+class TestWatsonxPassthroughConfig:
+    """Tests for WatsonxPassthroughConfig class."""
+
+    def test_is_streaming_request_true(self):
+        """Test that streaming is detected when stream=True in request data."""
+        config = WatsonxPassthroughConfig()
+        request_data = {"stream": True, "input": "test"}
+
+        result = config.is_streaming_request(
+            endpoint="ml/v1/text/generation", request_data=request_data
+        )
+
+        assert result is True
+
+    def test_is_streaming_request_false(self):
+        """Test that streaming is not detected when stream=False in request data."""
+        config = WatsonxPassthroughConfig()
+        request_data = {"stream": False, "input": "test"}
+
+        result = config.is_streaming_request(
+            endpoint="ml/v1/text/generation", request_data=request_data
+        )
+
+        assert result is False
+
+    def test_is_streaming_request_missing_stream_key(self):
+        """Test that streaming defaults to False when stream key is missing."""
+        config = WatsonxPassthroughConfig()
+        request_data = {"input": "test"}
+
+        result = config.is_streaming_request(
+            endpoint="ml/v1/text/generation", request_data=request_data
+        )
+
+        assert result is False
+
+    def test_get_complete_url_with_api_base(self):
+        """Test URL construction with explicit api_base."""
+        config = WatsonxPassthroughConfig()
+        api_base = "https://us-south.ml.cloud.ibm.com"
+        endpoint = "ml/v1/text/generation"
+        request_query_params = {"version": "2024-03-19"}
+
+        complete_url, base_target_url = config.get_complete_url(
+            api_base=api_base,
+            api_key=None,
+            model="ibm/granite-13b-chat-v2",
+            endpoint=endpoint,
+            request_query_params=request_query_params,
+            litellm_params={},
+        )
+
+        assert isinstance(complete_url, httpx.URL)
+        assert str(complete_url).startswith(api_base)
+        assert endpoint in str(complete_url)
+        assert "version=2024-03-19" in str(complete_url)
+        assert base_target_url == api_base
+
+    @patch("litellm.llms.watsonx.passthrough.transformation.get_secret_str")
+    def test_get_complete_url_with_env_api_base(self, mock_get_secret):
+        """Test URL construction with api_base from environment."""
+        config = WatsonxPassthroughConfig()
+        env_api_base = "https://eu-de.ml.cloud.ibm.com"
+        mock_get_secret.return_value = env_api_base
+
+        endpoint = "ml/v1/text/tokenization"
+        request_query_params = {"version": "2024-03-19"}
+
+        complete_url, base_target_url = config.get_complete_url(
+            api_base=None,
+            api_key=None,
+            model="ibm/granite-13b-chat-v2",
+            endpoint=endpoint,
+            request_query_params=request_query_params,
+            litellm_params={},
+        )
+
+        assert isinstance(complete_url, httpx.URL)
+        assert str(complete_url).startswith(env_api_base)
+        assert endpoint in str(complete_url)
+        assert base_target_url == env_api_base
+
+    def test_get_complete_url_with_query_params(self):
+        """Test that query parameters are correctly added to URL."""
+        config = WatsonxPassthroughConfig()
+        api_base = "https://us-south.ml.cloud.ibm.com"
+        endpoint = "ml/v1/text/generation"
+        request_query_params = {
+            "version": "2024-03-19",
+        }
+
+        complete_url, _ = config.get_complete_url(
+            api_base=api_base,
+            api_key=None,
+            model="ibm/granite-13b-chat-v2",
+            endpoint=endpoint,
+            request_query_params=request_query_params,
+            litellm_params={},
+        )
+
+        url_str = str(complete_url)
+        assert "version=2024-03-19" in url_str
+
+    def test_get_complete_url_without_query_params(self):
+        """Test URL construction without query parameters."""
+        config = WatsonxPassthroughConfig()
+        api_base = "https://us-south.ml.cloud.ibm.com"
+        endpoint = "ml/v1/models"
+
+        complete_url, base_target_url = config.get_complete_url(
+            api_base=api_base,
+            api_key=None,
+            model="",
+            endpoint=endpoint,
+            request_query_params=None,
+            litellm_params={},
+        )
+
+        assert isinstance(complete_url, httpx.URL)
+        assert str(complete_url) == f"{api_base}/{endpoint}"
+        assert base_target_url == api_base
+        assert "version=2024-03-19" not in str(complete_url)
+
+    @patch("litellm.llms.watsonx.passthrough.transformation.get_secret_str")
+    def test_get_api_base_with_explicit_value(self, mock_get_secret):
+        """Test get_api_base returns explicit value when provided."""
+        explicit_base = "https://custom.watsonx.com"
+
+        result = WatsonxPassthroughConfig.get_api_base(api_base=explicit_base)
+
+        assert result == explicit_base
+        mock_get_secret.assert_not_called()
+
+    @patch("litellm.llms.watsonx.passthrough.transformation.get_secret_str")
+    def test_get_api_base_from_environment(self, mock_get_secret):
+        """Test get_api_base retrieves from environment when not provided."""
+        env_base = "https://env.watsonx.com"
+        mock_get_secret.return_value = env_base
+
+        result = WatsonxPassthroughConfig.get_api_base(api_base=None)
+
+        assert result == env_base
+        mock_get_secret.assert_called_once_with("WATSONX_API_BASE")
+
+    @patch("litellm.llms.watsonx.passthrough.transformation.get_secret_str")
+    def test_get_api_key_with_explicit_value(self, mock_get_secret):
+        """Test get_api_key returns explicit value when provided."""
+        explicit_key = "test-api-key-123"
+
+        result = WatsonxPassthroughConfig.get_api_key(api_key=explicit_key)
+
+        assert result == explicit_key
+        mock_get_secret.assert_not_called()
+
+    @patch("litellm.llms.watsonx.passthrough.transformation.get_secret_str")
+    def test_get_api_key_from_environment(self, mock_get_secret):
+        """Test get_api_key retrieves from environment when not provided."""
+        env_key = "env-api-key-456"
+        mock_get_secret.return_value = env_key
+
+        result = WatsonxPassthroughConfig.get_api_key(api_key=None)
+
+        assert result == env_key
+        mock_get_secret.assert_called_once_with("WATSON_API_KEY")
+
+    def test_get_base_model_returns_model(self):
+        """Test get_base_model returns the model as-is."""
+        model = "ibm/granite-13b-chat-v2"
+
+        result = WatsonxPassthroughConfig.get_base_model(model)
+
+        assert result == model
+
+    def test_get_base_model_with_deployment(self):
+        """Test get_base_model with deployment model."""
+        model = "deployment/test-deployment-id"
+
+        result = WatsonxPassthroughConfig.get_base_model(model)
+
+        assert result == model
+
+    def test_get_complete_url_with_different_endpoints(self):
+        """Test URL construction with various endpoint paths."""
+        config = WatsonxPassthroughConfig()
+        api_base = "https://us-south.ml.cloud.ibm.com"
+
+        endpoints = [
+            "ml/v1/text/generation",
+            "ml/v1/text/tokenization",
+            "ml/v1/deployments/test-id/text/generation",
+            "ml/v1/models",
+            "ml/v1/foundation_model_specs",
+        ]
+
+        for endpoint in endpoints:
+            complete_url, base_target_url = config.get_complete_url(
+                api_base=api_base,
+                api_key=None,
+                model="",
+                endpoint=endpoint,
+                request_query_params={"version": "2024-03-19"},
+                litellm_params={},
+            )
+
+            assert isinstance(complete_url, httpx.URL)
+            assert endpoint in str(complete_url)
+            assert base_target_url == api_base
+
+    def test_get_complete_url_preserves_query_param_order(self):
+        """Test that query parameters maintain their values correctly."""
+        config = WatsonxPassthroughConfig()
+        api_base = "https://us-south.ml.cloud.ibm.com"
+        endpoint = "ml/v1/text/generation"
+        request_query_params = {
+            "version": "2024-03-19",
+            "project_id": "abc-123",
+            "space_id": "xyz-789",
+        }
+
+        complete_url, _ = config.get_complete_url(
+            api_base=api_base,
+            api_key=None,
+            model="",
+            endpoint=endpoint,
+            request_query_params=request_query_params,
+            litellm_params={},
+        )
+
+        url_str = str(complete_url)
+        # Verify all params are present
+        assert "version=2024-03-19" in url_str
+        assert "project_id=abc-123" in url_str
+        assert "space_id=xyz-789" in url_str
+
+    def test_is_streaming_request_with_various_stream_values(self):
+        """Test streaming detection with different stream value types."""
+        config = WatsonxPassthroughConfig()
+
+        # Test with boolean True
+        assert config.is_streaming_request("endpoint", {"stream": True}) is True
+
+        # Test with boolean False
+        assert config.is_streaming_request("endpoint", {"stream": False}) is False
+
+        # Test with string "true" (truthy string)
+        result = config.is_streaming_request("endpoint", {"stream": "true"})
+        assert result == "true"  # Returns the value as-is from .get()
+
+        # Test with integer 1 (truthy)
+        result = config.is_streaming_request("endpoint", {"stream": 1})
+        assert result == 1
+
+        # Test with integer 0 (falsy)
+        result = config.is_streaming_request("endpoint", {"stream": 0})
+        assert result == 0
+
+        # Test with None
+        result = config.is_streaming_request("endpoint", {"stream": None})
+        assert result is None
+
+        # Test with empty dict (defaults to False)
+        assert config.is_streaming_request("endpoint", {}) is False
--- a/tests/test_litellm/proxy/pass_through_endpoints/test_watsonx_proxy_route.py
+++ b/tests/test_litellm/proxy/pass_through_endpoints/test_watsonx_proxy_route.py
@ -0,0 +1,444 @@
+"""
+Unit tests for watsonx_proxy_route endpoint.
+
+Tests the Watsonx pass-through endpoint that handles automatic IAM token management
+and version parameter injection.
+"""
+
+import json
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
+
+import pytest
+from fastapi import HTTPException, Request, Response
+
+sys.path.insert(
+    0, os.path.abspath("../../../..")
+)  # Adds the parent directory to the system path
+
+import litellm
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
+    watsonx_proxy_route,
+)
+
+
+class TestWatsonxProxyRoute:
+    """Tests for the Watsonx pass-through route."""
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_success_non_streaming(self):
+        """Test successful non-streaming request through Watsonx proxy route."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "application/json"}
+        mock_request.json = AsyncMock(return_value={"stream": False, "input": "test"})
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock provider config
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/text/generation",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token"
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(
+            return_value={"model_id": "ibm/granite-13b-chat-v2", "results": []}
+        )
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            result = await watsonx_proxy_route(
+                endpoint="ml/v1/text/generation",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify provider config was called correctly
+            mock_provider_config.get_complete_url.assert_called_once()
+            mock_provider_config.validate_environment.assert_called_once()
+
+            # Verify create_pass_through_route was called with correct parameters
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert call_args["endpoint"] == "ml/v1/text/generation"
+            assert (
+                call_args["target"]
+                == "https://us-south.ml.cloud.ibm.com/ml/v1/text/generation"
+            )
+            assert (
+                call_args["custom_headers"]["Authorization"] == "Bearer test-iam-token"
+            )
+            assert call_args["is_streaming_request"] is False
+            assert call_args["custom_llm_provider"] == "watsonx"
+            assert (
+                call_args["query_params"]["version"]
+                == litellm.WATSONX_DEFAULT_API_VERSION
+            )
+
+            # Verify endpoint function was called
+            mock_endpoint_func.assert_called_once_with(
+                mock_request, mock_response, mock_user_api_key_dict
+            )
+
+            assert result == {"model_id": "ibm/granite-13b-chat-v2", "results": []}
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_success_streaming(self):
+        """Test successful streaming request through Watsonx proxy route."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "application/json"}
+        mock_request.json = AsyncMock(return_value={"stream": True, "input": "test"})
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock provider config
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/text/generation_stream",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token"
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(return_value="streaming_response")
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            result = await watsonx_proxy_route(
+                endpoint="ml/v1/text/generation_stream",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify create_pass_through_route was called with streaming enabled
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert call_args["is_streaming_request"] is True
+
+            assert result == "streaming_response"
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_get_request(self):
+        """Test GET request through Watsonx proxy route."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "GET"
+        mock_request.query_params = {"project_id": "test-project"}
+        mock_request.headers = {}
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock provider config
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/models",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token"
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(return_value={"resources": []})
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            result = await watsonx_proxy_route(
+                endpoint="ml/v1/models",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify is_streaming_request is False for GET requests
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert call_args["is_streaming_request"] is False
+
+            assert result == {"resources": []}
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_multipart_form_data(self):
+        """Test multipart/form-data request through Watsonx proxy route."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "multipart/form-data; boundary=----"}
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock form data
+        mock_form_data = {"file": "test_file", "stream": False}
+
+        # Mock provider config
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/text/tokenization",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token"
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(return_value={"token_count": 10})
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.get_form_data",
+                return_value=mock_form_data,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            result = await watsonx_proxy_route(
+                endpoint="ml/v1/text/tokenization",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify is_streaming_request is False for non-streaming form data
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert call_args["is_streaming_request"] is False
+
+            assert result == {"token_count": 10}
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_no_provider_config(self):
+        """Test that HTTPException is raised when provider config is not found."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "application/json"}
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=None,
+            ),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                await watsonx_proxy_route(
+                    endpoint="ml/v1/text/generation",
+                    request=mock_request,
+                    fastapi_response=mock_response,
+                    user_api_key_dict=mock_user_api_key_dict,
+                )
+
+            assert exc_info.value.status_code == 404
+            assert exc_info.value.detail == "Watsonx passthrough config not found"
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_version_parameter_injection(self):
+        """Test that version parameter is correctly injected into query params."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "application/json"}
+        mock_request.json = AsyncMock(return_value={"input": "test"})
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock provider config
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/text/generation",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token"
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(return_value={})
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            await watsonx_proxy_route(
+                endpoint="ml/v1/text/generation",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify version parameter is injected
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert "query_params" in call_args
+            assert "version" in call_args["query_params"]
+            assert (
+                call_args["query_params"]["version"]
+                == litellm.WATSONX_DEFAULT_API_VERSION
+            )
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_custom_headers_from_validate_environment(self):
+        """Test that custom headers from validate_environment are passed through."""
+        # Setup mocks
+        mock_request = MagicMock(spec=Request)
+        mock_request.method = "POST"
+        mock_request.query_params = {}
+        mock_request.headers = {"content-type": "application/json"}
+        mock_request.json = AsyncMock(return_value={"input": "test"})
+        mock_response = MagicMock(spec=Response)
+        mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+        # Mock provider config with custom headers
+        mock_provider_config = MagicMock()
+        mock_provider_config.get_complete_url.return_value = (
+            "https://us-south.ml.cloud.ibm.com/ml/v1/text/generation",
+            {},
+        )
+        mock_provider_config.validate_environment.return_value = {
+            "Authorization": "Bearer test-iam-token",
+            "X-Custom-Header": "custom-value",
+        }
+
+        # Mock endpoint function
+        mock_endpoint_func = AsyncMock(return_value={})
+
+        with (
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                return_value=mock_provider_config,
+            ),
+            patch(
+                "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                return_value=mock_endpoint_func,
+            ) as mock_create_route,
+        ):
+            await watsonx_proxy_route(
+                endpoint="ml/v1/text/generation",
+                request=mock_request,
+                fastapi_response=mock_response,
+                user_api_key_dict=mock_user_api_key_dict,
+            )
+
+            # Verify custom headers are passed through
+            mock_create_route.assert_called_once()
+            call_args = mock_create_route.call_args[1]
+            assert "custom_headers" in call_args
+            assert (
+                call_args["custom_headers"]["Authorization"] == "Bearer test-iam-token"
+            )
+            assert call_args["custom_headers"]["X-Custom-Header"] == "custom-value"
+
+    @pytest.mark.asyncio
+    async def test_watsonx_proxy_route_different_endpoints(self):
+        """Test various Watsonx endpoint paths."""
+        endpoints = [
+            "ml/v1/text/generation",
+            "ml/v1/text/tokenization",
+            "ml/v1/deployments/test-deployment/text/generation",
+            "ml/v1/models",
+        ]
+
+        for endpoint_path in endpoints:
+            # Setup mocks
+            mock_request = MagicMock(spec=Request)
+            mock_request.method = "POST"
+            mock_request.query_params = {}
+            mock_request.headers = {"content-type": "application/json"}
+            mock_request.json = AsyncMock(return_value={"input": "test"})
+            mock_response = MagicMock(spec=Response)
+            mock_user_api_key_dict = MagicMock(spec=UserAPIKeyAuth)
+
+            # Mock provider config
+            mock_provider_config = MagicMock()
+            mock_provider_config.get_complete_url.return_value = (
+                f"https://us-south.ml.cloud.ibm.com/{endpoint_path}",
+                {},
+            )
+            mock_provider_config.validate_environment.return_value = {
+                "Authorization": "Bearer test-iam-token"
+            }
+
+            # Mock endpoint function
+            mock_endpoint_func = AsyncMock(return_value={})
+
+            with (
+                patch(
+                    "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.ProviderConfigManager.get_provider_passthrough_config",
+                    return_value=mock_provider_config,
+                ),
+                patch(
+                    "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
+                    return_value=mock_endpoint_func,
+                ) as mock_create_route,
+            ):
+                await watsonx_proxy_route(
+                    endpoint=endpoint_path,
+                    request=mock_request,
+                    fastapi_response=mock_response,
+                    user_api_key_dict=mock_user_api_key_dict,
+                )
+
+                # Verify endpoint is passed correctly
+                mock_create_route.assert_called_once()
+                call_args = mock_create_route.call_args[1]
+                assert call_args["endpoint"] == endpoint_path
+                assert (
+                    call_args["target"]
+                    == f"https://us-south.ml.cloud.ibm.com/{endpoint_path}"
+                )