diff --git a/docs/my-website/docs/providers/oci.md b/docs/my-website/docs/providers/oci.md index 1f52fba04f3..cea5d6824a0 100644 --- a/docs/my-website/docs/providers/oci.md +++ b/docs/my-website/docs/providers/oci.md @@ -29,17 +29,38 @@ Check the [OCI Models List](https://docs.oracle.com/en-us/iaas/Content/generativ ## Authentication -LiteLLM uses OCI signing key authentication. Follow the [official Oracle tutorial](https://docs.oracle.com/en-us/iaas/Content/API/Concepts/apisigningkey.htm) to create a signing key and obtain the following parameters: +LiteLLM supports two authentication methods for OCI: + +### Method 1: Manual Credentials +Provide individual OCI credentials directly to LiteLLM. Follow the [official Oracle tutorial](https://docs.oracle.com/en-us/iaas/Content/API/Concepts/apisigningkey.htm) to create a signing key and obtain the following parameters: - `user` - `fingerprint` - `tenancy` - `region` -- `key_file` +- `key_file` or `key` +- `compartment_id` + +This is the default method for LiteLLM AI Gateway (LLM Proxy) access to OCI GenAI models. + +### Method 2: OCI SDK Signer +Use an OCI SDK `Signer` object for authentication. This method: +- Leverages the official [OCI SDK for signing](https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html) +- Supports additional authentication methods (instance principals, workload identity, etc.) + +To use this method, install the OCI SDK: +```bash +pip install oci +``` + +This method is an alternative when using the LiteLLM SDK on Oracle Cloud Infrastructure (instances or Oracle Kubernetes Engine). ## Usage -Input the parameters obtained from the OCI signing key creation process into the `completion` function. + + + +Input the parameters obtained from the OCI signing key creation process into the `completion` function: ```python import os @@ -64,10 +85,119 @@ response = completion( print(response) ``` + + + +Use the OCI SDK `Signer` for authentication: + +```python +from litellm import completion +from oci.signer import Signer + +# Create an OCI Signer +signer = Signer( + tenancy="ocid1.tenancy.oc1..", + user="ocid1.user.oc1..", + fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx", + private_key_file_location="~/.oci/key.pem", + # Or use private_key_content="" +) + +messages = [{"role": "user", "content": "Hey! how's it going?"}] +response = completion( + model="oci/xai.grok-4", + messages=messages, + oci_signer=signer, + oci_region="us-chicago-1", # Optional, defaults to us-ashburn-1 + oci_serving_mode="ON_DEMAND", # Optional, default is "ON_DEMAND". Other option is "DEDICATED" + oci_compartment_id="", +) +print(response) +``` + +**Alternative: Use OCI Config File** + +The OCI SDK can automatically load credentials from `~/.oci/config`: + +```python +from litellm import completion +from oci.config import from_file +from oci.signer import Signer + +# Load config from file +config = from_file("~/.oci/config", "DEFAULT") # "DEFAULT" is the profile name +signer = Signer( + tenancy=config["tenancy"], + user=config["user"], + fingerprint=config["fingerprint"], + private_key_file_location=config["key_file"], + pass_phrase=config.get("pass_phrase") # Optional if key is encrypted +) + +messages = [{"role": "user", "content": "Hey! how's it going?"}] +response = completion( + model="oci/xai.grok-4", + messages=messages, + oci_signer=signer, + oci_region=config["region"], + oci_compartment_id="", +) +print(response) +``` + +**Instance Principal Authentication** + +For applications running on OCI compute instances: + +```python +from litellm import completion +from oci.auth.signers import InstancePrincipalsSecurityTokenSigner + +oci.auth.signers.get_oke_workload_identity_resource_principal_signer() +# Use instance principal authentication +signer = InstancePrincipalsSecurityTokenSigner() + +messages = [{"role": "user", "content": "Hey! how's it going?"}] +response = completion( + model="oci/xai.grok-4", + messages=messages, + oci_signer=signer, + oci_region="us-chicago-1", + oci_compartment_id="", +) +print(response) +``` + +**Use workload identity authentication** + +For applications running in Oracle Kubernetes Engine (OKE): + +```python +from litellm import completion +from oci.auth.signers import get_oke_workload_identity_resource_principal_signer + +# Use instance principal authentication +signer = get_oke_workload_identity_resource_principal_signer() + +messages = [{"role": "user", "content": "Hey! how's it going?"}] +response = completion( + model="oci/xai.grok-4", + messages=messages, + oci_signer=signer, + oci_region="us-chicago-1", + oci_compartment_id="", +) +print(response) +``` + + ## Usage - Streaming Just set `stream=True` when calling completion. + + + ```python import os from litellm import completion @@ -93,10 +223,68 @@ for chunk in response: print(chunk["choices"][0]["delta"]["content"]) # same as openai format ``` + + + +```python +from litellm import completion +from oci.signer import Signer + +signer = Signer( + tenancy="ocid1.tenancy.oc1..", + user="ocid1.user.oc1..", + fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx", + private_key_file_location="~/.oci/key.pem", +) + +messages = [{"role": "user", "content": "Hey! how's it going?"}] +response = completion( + model="oci/xai.grok-4", + messages=messages, + stream=True, + oci_signer=signer, + oci_region="us-chicago-1", + oci_compartment_id="", +) +for chunk in response: + print(chunk["choices"][0]["delta"]["content"]) # same as openai format +``` + + + + ## Usage Examples by Model Type ### Using Cohere Models + + + +```python +from litellm import completion +from oci.signer import Signer + +signer = Signer( + tenancy="ocid1.tenancy.oc1..", + user="ocid1.user.oc1..", + fingerprint="xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx", + private_key_file_location="~/.oci/key.pem", +) + +messages = [{"role": "user", "content": "Explain quantum computing"}] +response = completion( + model="oci/cohere.command-latest", + messages=messages, + oci_signer=signer, + oci_region="us-chicago-1", + oci_compartment_id="", +) +print(response) +``` + + + + ```python from litellm import completion @@ -112,4 +300,7 @@ response = completion( oci_compartment_id=, ) print(response) -``` \ No newline at end of file +``` + + + \ No newline at end of file diff --git a/litellm/llms/oci/chat/transformation.py b/litellm/llms/oci/chat/transformation.py index 3ab827797c5..167ba26bacb 100644 --- a/litellm/llms/oci/chat/transformation.py +++ b/litellm/llms/oci/chat/transformation.py @@ -2,7 +2,8 @@ import base64 import datetime import hashlib import json -from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Tuple, Union +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Protocol, Tuple, Union from urllib.parse import urlparse import httpx @@ -62,6 +63,47 @@ else: LiteLLMLoggingObj = Any +class OCISignerProtocol(Protocol): + """ + Protocol for OCI request signers (e.g., oci.signer.Signer). + + This protocol defines the interface expected for OCI SDK signer objects. + Compatible with the OCI Python SDK's Signer class. + + See: https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html + """ + + def do_request_sign(self, request: Any, *, enforce_content_headers: bool = False) -> None: + """ + Sign an HTTP request by adding authentication headers. + + Args: + request: Request object with method, url, headers, body, and path_url attributes + enforce_content_headers: Whether to enforce content-type and content-length headers + """ + ... + + +@dataclass +class OCIRequestWrapper: + """ + Wrapper for HTTP requests compatible with OCI signer interface. + + This class wraps request data in a format compatible with OCI SDK signers, + which expect objects with method, url, headers, body, and path_url attributes. + """ + method: str + url: str + headers: dict + body: bytes + + @property + def path_url(self) -> str: + """Returns the path + query string for OCI signing.""" + parsed_url = urlparse(self.url) + return parsed_url.path + ("?" + parsed_url.query if parsed_url.query else "") + + def sha256_base64(data: bytes) -> str: digest = hashlib.sha256(data).digest() return base64.b64encode(digest).decode() @@ -228,29 +270,89 @@ class OCIChatConfig(BaseConfig): return adapted_params - def sign_request( + def _sign_with_oci_signer( self, headers: dict, optional_params: dict, request_data: dict, api_base: str, - api_key: Optional[str] = None, - model: Optional[str] = None, - stream: Optional[bool] = None, - fake_stream: Optional[bool] = None, - ) -> Tuple[dict, Optional[bytes]]: + ) -> Tuple[dict, bytes]: """ - Some providers like Bedrock require signing the request. The sign request funtion needs access to `request_data` and `complete_url` - Args: - headers: dict - optional_params: dict - request_data: dict - the request body being sent in http request - api_base: str - the complete url being sent in http request - Returns: - dict - the signed headers - """ - import json + Sign request using OCI SDK Signer object. + Args: + headers: Request headers to be signed + optional_params: Optional parameters including oci_signer + request_data: The request body dict to be sent in HTTP request + api_base: The complete URL for the HTTP request + + Returns: + Tuple of (signed_headers, encoded_body) + + Raises: + OCIError: If signing fails + ValueError: If HTTP method is unsupported + """ + oci_signer = optional_params.get("oci_signer") + body = json.dumps(request_data).encode("utf-8") + method = str(optional_params.get("method", "POST")).upper() + + if method not in ["POST", "GET", "PUT", "DELETE", "PATCH"]: + raise ValueError(f"Unsupported HTTP method: {method}") + + prepared_headers = headers.copy() + prepared_headers.setdefault("content-type", "application/json") + prepared_headers.setdefault("content-length", str(len(body))) + + request_wrapper = OCIRequestWrapper( + method=method, + url=api_base, + headers=prepared_headers, + body=body + ) + + if oci_signer is None: + raise ValueError("oci_signer cannot be None when calling _sign_with_oci_signer") + + try: + oci_signer.do_request_sign(request_wrapper, enforce_content_headers=True) + except Exception as e: + raise OCIError( + status_code=500, + message=( + f"Failed to sign request with provided oci_signer: {str(e)}. " + "The signer must implement the OCI SDK Signer interface with a " + "do_request_sign(request, enforce_content_headers=True) method. " + "See: https://docs.oracle.com/en-us/iaas/tools/python/latest/api/signing.html" + ) + ) from e + + headers.update(request_wrapper.headers) + return headers, body + + def _sign_with_manual_credentials( + self, + headers: dict, + optional_params: dict, + request_data: dict, + api_base: str, + ) -> Tuple[dict, None]: + """ + Sign request using manual OCI credentials. + + Args: + headers: Request headers to be signed + optional_params: Optional parameters including OCI credentials + request_data: The request body dict to be sent in HTTP request + api_base: The complete URL for the HTTP request + + Returns: + Tuple of (signed_headers, None) + + Raises: + Exception: If required credentials are missing + ImportError: If cryptography package is not installed + """ oci_region = optional_params.get("oci_region", "us-ashburn-1") api_base = ( api_base @@ -355,6 +457,69 @@ class OCIChatConfig(BaseConfig): return headers, None + def sign_request( + self, + headers: dict, + optional_params: dict, + request_data: dict, + api_base: str, + api_key: Optional[str] = None, + model: Optional[str] = None, + stream: Optional[bool] = None, + fake_stream: Optional[bool] = None, + ) -> Tuple[dict, Optional[bytes]]: + """ + Sign the OCI request by adding authentication headers. + + Supports two signing modes: + 1. OCI SDK Signer: Use an oci_signer object to sign the request + 2. Manual Signing: Use OCI credentials to manually sign the request + + Args: + headers: Request headers to be signed + optional_params: Optional parameters including auth credentials or oci_signer + request_data: The request body dict to be sent in HTTP request + api_base: The complete URL for the HTTP request + api_key: Optional API key (not used for OCI) + model: Optional model name + stream: Optional streaming flag + fake_stream: Optional fake streaming flag + + Returns: + Tuple of (signed_headers, encoded_body): + - If oci_signer is provided: Returns (headers, body) where body is the encoded JSON + - If manual credentials are provided: Returns (headers, None) as body is not returned + for the manual signing path + + Raises: + OCIError: If signing fails with oci_signer + Exception: If required credentials are missing + ImportError: If cryptography package is not installed (manual signing only) + + Example: + >>> from oci.signer import Signer + >>> signer = Signer( + ... tenancy="ocid1.tenancy.oc1..", + ... user="ocid1.user.oc1..", + ... fingerprint="xx:xx:xx", + ... private_key_file_location="~/.oci/key.pem" + ... ) + >>> headers, body = config.sign_request( + ... headers={}, + ... optional_params={"oci_signer": signer}, + ... request_data={"message": "Hello"}, + ... api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/..." + ... ) + """ + oci_signer = optional_params.get("oci_signer") + + # If a signer is provided, use it for request signing + if oci_signer is not None: + return self._sign_with_oci_signer(headers, optional_params, request_data, api_base) + + # Standard manual credential signing + return self._sign_with_manual_credentials(headers, optional_params, request_data, api_base) + def validate_environment( self, headers: dict, @@ -365,36 +530,67 @@ class OCIChatConfig(BaseConfig): api_key: Optional[str] = None, api_base: Optional[str] = None, ) -> dict: + """ + Validate the OCI environment and credentials. + + Supports two authentication modes: + 1. OCI SDK Signer: Pass an oci_signer object (e.g., oci.signer.Signer) + 2. Manual Credentials: Pass oci_user, oci_fingerprint, oci_tenancy, and oci_key/oci_key_file + + Args: + headers: Request headers to populate + model: Model name + messages: List of chat messages + optional_params: Optional parameters including authentication credentials + litellm_params: LiteLLM parameters + api_key: Optional API key (not used for OCI) + api_base: Optional API base URL + + Returns: + Updated headers dict + + Raises: + Exception: If required parameters are missing or invalid + """ + oci_signer = optional_params.get("oci_signer") oci_region = optional_params.get("oci_region", "us-ashburn-1") + + # Determine api_base api_base = ( api_base or litellm.api_base or f"https://inference.generativeai.{oci_region}.oci.oraclecloud.com" ) - oci_user = optional_params.get("oci_user") - oci_fingerprint = optional_params.get("oci_fingerprint") - oci_tenancy = optional_params.get("oci_tenancy") - oci_key = optional_params.get("oci_key") - oci_key_file = optional_params.get("oci_key_file") - oci_compartment_id = optional_params.get("oci_compartment_id") - - if ( - not oci_user - or not oci_fingerprint - or not oci_tenancy - or not (oci_key or oci_key_file) - or not oci_compartment_id - ): - raise Exception( - "Missing required parameters: oci_user, oci_fingerprint, oci_tenancy, oci_compartment_id " - "and at least one of oci_key or oci_key_file." - ) if not api_base: raise Exception( - "Either `api_base` must be provided or `litellm.api_base` must be set. Alternatively, you can set the `oci_region` optional parameter to use the default OCI region." + "Either `api_base` must be provided or `litellm.api_base` must be set. " + "Alternatively, you can set the `oci_region` optional parameter to use the default OCI region." ) + # Validate credentials only if signer is not provided + if oci_signer is None: + oci_user = optional_params.get("oci_user") + oci_fingerprint = optional_params.get("oci_fingerprint") + oci_tenancy = optional_params.get("oci_tenancy") + oci_key = optional_params.get("oci_key") + oci_key_file = optional_params.get("oci_key_file") + oci_compartment_id = optional_params.get("oci_compartment_id") + + if ( + not oci_user + or not oci_fingerprint + or not oci_tenancy + or not (oci_key or oci_key_file) + or not oci_compartment_id + ): + raise Exception( + "Missing required parameters: oci_user, oci_fingerprint, oci_tenancy, oci_compartment_id " + "and at least one of oci_key or oci_key_file. " + "Alternatively, provide an oci_signer object from the OCI SDK." + ) + + # Common header setup headers.update( { "content-type": "application/json", @@ -442,12 +638,12 @@ class OCIChatConfig(BaseConfig): for openai_key, oci_key in open_ai_to_oci_param_map.items(): if oci_key and openai_key in optional_params: selected_params[oci_key] = optional_params[openai_key] # type: ignore[index] - + # Also check for already-mapped OCI params (for backward compatibility) for oci_value in open_ai_to_oci_param_map.values(): if oci_value and oci_value in optional_params and oci_value not in selected_params: selected_params[oci_value] = optional_params[oci_value] # type: ignore[index] - + if "tools" in selected_params: if vendor == OCIVendors.COHERE: selected_params["tools"] = self.adapt_tool_definitions_to_cohere_standard( # type: ignore[assignment] @@ -465,7 +661,7 @@ class OCIChatConfig(BaseConfig): for msg in messages[:-1]: # All messages except the last one role = msg.get("role") content = msg.get("content") - + if isinstance(content, list): # Extract text from content array text_content = "" @@ -473,11 +669,11 @@ class OCIChatConfig(BaseConfig): if isinstance(content_item, dict) and content_item.get("type") == "text": text_content += content_item.get("text", "") content = text_content - + # Ensure content is a string if not isinstance(content, str): content = str(content) if content is not None else "" - + # Handle tool calls tool_calls: Optional[List[CohereToolCall]] = None if role == "assistant" and "tool_calls" in msg and msg.get("tool_calls"): # type: ignore[union-attr,typeddict-item] @@ -492,12 +688,12 @@ class OCIChatConfig(BaseConfig): arguments = {} else: arguments = raw_arguments - + tool_calls.append(CohereToolCall( name=str(tool_call.get("function", {}).get("name", "")), parameters=arguments )) - + if role == "user": chat_history.append(CohereMessage(role="USER", message=content)) elif role == "assistant": @@ -505,11 +701,11 @@ class OCIChatConfig(BaseConfig): elif role == "tool": # Tool messages need special handling chat_history.append(CohereMessage( - role="TOOL", + role="TOOL", message=content, toolCalls=None # Tool messages don't have tool calls )) - + return chat_history def adapt_tool_definitions_to_cohere_standard(self, tools: List[Dict[str, Any]]) -> List[CohereTool]: @@ -519,7 +715,7 @@ class OCIChatConfig(BaseConfig): function_def = tool.get("function", {}) parameters = function_def.get("parameters", {}).get("properties", {}) required = function_def.get("parameters", {}).get("required", []) - + parameter_definitions = {} for param_name, param_schema in parameters.items(): parameter_definitions[param_name] = CohereParameterDefinition( @@ -527,13 +723,13 @@ class OCIChatConfig(BaseConfig): type=param_schema.get("type", "string"), isRequired=param_name in required ) - + cohere_tools.append(CohereTool( name=function_def.get("name", ""), description=function_def.get("description", ""), parameterDefinitions=parameter_definitions )) - + return cohere_tools def _extract_text_content(self, content: Any) -> str: @@ -586,7 +782,7 @@ class OCIChatConfig(BaseConfig): user_messages = [msg for msg in messages if msg.get("role") == "user"] if not user_messages: raise Exception("No user message found for Cohere model") - + # Create Cohere-specific chat request chat_request = CohereChatRequest( @@ -595,7 +791,7 @@ class OCIChatConfig(BaseConfig): chatHistory=self.adapt_messages_to_cohere_standard(messages), **self._get_optional_params(OCIVendors.COHERE, optional_params) ) - + data = OCICompletionPayload( compartmentId=oci_compartment_id, servingMode=servingMode, @@ -616,24 +812,24 @@ class OCIChatConfig(BaseConfig): return data.model_dump(exclude_none=True) def _handle_cohere_response( - self, - json_response: dict, - model: str, + self, + json_response: dict, + model: str, model_response: ModelResponse ) -> ModelResponse: """Handle Cohere-specific response format.""" cohere_response = CohereChatResult(**json_response) # Cohere response format (uses camelCase) model_id = model - + # Set basic response info model_response.model = model_id model_response.created = int(datetime.datetime.now().timestamp()) - + # Extract the response text response_text = cohere_response.chatResponse.text oci_finish_reason = cohere_response.chatResponse.finishReason - + # Map finish reason if oci_finish_reason == "COMPLETE": finish_reason = "stop" @@ -641,7 +837,7 @@ class OCIChatConfig(BaseConfig): finish_reason = "length" else: finish_reason = "stop" - + # Handle tool calls tool_calls: Optional[List[Dict[str, Any]]] = None if cohere_response.chatResponse.toolCalls: @@ -655,7 +851,7 @@ class OCIChatConfig(BaseConfig): "arguments": json.dumps(tool_call.parameters) } }) - + # Create choice from litellm.types.utils import Choices choice = Choices( @@ -668,7 +864,7 @@ class OCIChatConfig(BaseConfig): finish_reason=finish_reason ) model_response.choices = [choice] - + # Extract usage info usage_info = cohere_response.chatResponse.usage from litellm.types.utils import Usage @@ -677,13 +873,13 @@ class OCIChatConfig(BaseConfig): completion_tokens=usage_info.completionTokens, # type: ignore[union-attr] total_tokens=usage_info.totalTokens # type: ignore[union-attr] ) - + return model_response def _handle_generic_response( - self, - json: dict, - model: str, + self, + json: dict, + model: str, model_response: ModelResponse, raw_response: httpx.Response ) -> ModelResponse: @@ -695,7 +891,7 @@ class OCIChatConfig(BaseConfig): message=f"Response cannot be casted to OCICompletionResponse: {str(e)}", status_code=raw_response.status_code, ) - + iso_str = completion_response.chatResponse.timeCreated dt = datetime.datetime.fromisoformat(iso_str.replace("Z", "+00:00")) model_response.created = int(dt.timestamp()) @@ -751,7 +947,7 @@ class OCIChatConfig(BaseConfig): ) vendor = get_vendor_from_model(model) - + # Handle response based on vendor type if vendor == OCIVendors.COHERE: model_response = self._handle_cohere_response(json, model, model_response) @@ -1080,7 +1276,7 @@ class OCIStreamWrapper(CustomStreamWrapper): if not chunk.startswith("data:"): raise ValueError(f"Chunk does not start with 'data:': {chunk}") dict_chunk = json.loads(chunk[5:]) # Remove 'data: ' prefix and parse JSON - + # Check if this is a Cohere stream chunk if "apiFormat" in dict_chunk and dict_chunk.get("apiFormat") == "COHERE": return self._handle_cohere_stream_chunk(dict_chunk) diff --git a/tests/test_litellm/llms/oci/chat/test_oci_chat_transformation.py b/tests/test_litellm/llms/oci/chat/test_oci_chat_transformation.py index 9ff93dfc1a2..9a43a4220ca 100644 --- a/tests/test_litellm/llms/oci/chat/test_oci_chat_transformation.py +++ b/tests/test_litellm/llms/oci/chat/test_oci_chat_transformation.py @@ -11,7 +11,7 @@ import litellm sys.path.insert(0, os.path.abspath("../../../../..")) from litellm import ModelResponse -from litellm.llms.oci.chat.transformation import OCIChatConfig, version +from litellm.llms.oci.chat.transformation import OCIChatConfig, OCIRequestWrapper, version TEST_MODEL_NAME = "xai.grok-4" TEST_MODEL = f"oci/{TEST_MODEL_NAME}" @@ -309,3 +309,190 @@ class TestOCIChatConfig: assert usage.prompt_tokens == 10 # type: ignore assert usage.completion_tokens == 20 # type: ignore assert usage.total_tokens == 30 # type: ignore + + +class TestOCISignerSupport: + """Tests for OCI SDK Signer integration.""" + + def test_validate_environment_with_oci_signer(self): + """Test validation when using oci_signer instead of manual credentials.""" + config = OCIChatConfig() + headers = {} + + # Mock signer object + class MockSigner: + def do_request_sign(self, request, enforce_content_headers=True): + request.headers["authorization"] = "Signature version=\"1\"" + + optional_params = { + "oci_signer": MockSigner(), + "oci_region": "us-ashburn-1" + } + + result = config.validate_environment( + headers=headers, + model=TEST_MODEL, + messages=TEST_MESSAGES, # type: ignore + optional_params=optional_params, + litellm_params={}, + ) + + assert result["content-type"] == "application/json" + assert result["user-agent"] == f"litellm/{version}" + + def test_validate_environment_with_oci_signer_no_compartment_id_in_validate(self): + """Test that oci_compartment_id is not required in validate_environment when using signer.""" + config = OCIChatConfig() + headers = {} + + class MockSigner: + def do_request_sign(self, request, enforce_content_headers=True): + request.headers["authorization"] = "Signature version=\"1\"" + + optional_params = { + "oci_signer": MockSigner(), + "oci_region": "us-phoenix-1" + } + + # Should not raise an exception even without oci_compartment_id + result = config.validate_environment( + headers=headers, + model=TEST_MODEL, + messages=TEST_MESSAGES, # type: ignore + optional_params=optional_params, + litellm_params={}, + ) + + assert result["content-type"] == "application/json" + + def test_sign_request_with_oci_signer(self): + """Test request signing with oci_signer.""" + config = OCIChatConfig() + + class MockSigner: + def do_request_sign(self, request, enforce_content_headers=True): + request.headers["authorization"] = "Signature version=\"1\"" + request.headers["date"] = "Mon, 01 Jan 2024 00:00:00 GMT" + + optional_params = { + "oci_signer": MockSigner(), + "method": "POST" + } + + headers, body = config.sign_request( + headers={}, + optional_params=optional_params, + request_data={"test": "data"}, + api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat" + ) + + assert "authorization" in headers + assert "Signature" in headers["authorization"] + assert body is not None # oci_signer path returns body + assert json.loads(body.decode("utf-8")) == {"test": "data"} + + def test_sign_request_with_oci_signer_updates_headers(self): + """Test that signer properly updates request headers.""" + config = OCIChatConfig() + + class MockSigner: + def do_request_sign(self, request, enforce_content_headers=True): + # Verify the request has the expected attributes + assert hasattr(request, "method") + assert hasattr(request, "url") + assert hasattr(request, "headers") + assert hasattr(request, "body") + assert hasattr(request, "path_url") + + # Add signature headers + request.headers["authorization"] = "Signature keyId=\"test\"" + request.headers["date"] = "Mon, 01 Jan 2024 00:00:00 GMT" + request.headers["x-content-sha256"] = "test-hash" + + optional_params = { + "oci_signer": MockSigner(), + } + + headers, body = config.sign_request( + headers={"custom-header": "custom-value"}, + optional_params=optional_params, + request_data={"message": "Hello"}, + api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat" + ) + + # Check that all signer-added headers are present + assert headers["authorization"] == "Signature keyId=\"test\"" + assert headers["date"] == "Mon, 01 Jan 2024 00:00:00 GMT" + assert headers["x-content-sha256"] == "test-hash" + # Original headers should be preserved + assert headers["custom-header"] == "custom-value" + + def test_sign_request_with_failing_oci_signer(self): + """Test error handling when oci_signer fails.""" + config = OCIChatConfig() + + class FailingSigner: + def do_request_sign(self, request, enforce_content_headers=True): + raise RuntimeError("Signing failed due to invalid credentials") + + optional_params = { + "oci_signer": FailingSigner(), + } + + from litellm.llms.oci.common_utils import OCIError + + with pytest.raises(OCIError) as excinfo: + config.sign_request( + headers={}, + optional_params=optional_params, + request_data={"test": "data"}, + api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat" + ) + + assert "Failed to sign request with provided oci_signer" in str(excinfo.value) + assert excinfo.value.status_code == 500 + + def test_sign_request_with_invalid_http_method(self): + """Test that invalid HTTP methods are rejected.""" + config = OCIChatConfig() + + class MockSigner: + def do_request_sign(self, request, enforce_content_headers=True): + pass + + optional_params = { + "oci_signer": MockSigner(), + "method": "INVALID" + } + + with pytest.raises(ValueError) as excinfo: + config.sign_request( + headers={}, + optional_params=optional_params, + request_data={"test": "data"}, + api_base="https://inference.generativeai.us-ashburn-1.oci.oraclecloud.com/20231130/actions/chat" + ) + + assert "Unsupported HTTP method: INVALID" in str(excinfo.value) + + def test_oci_request_wrapper_path_url(self): + """Test OCIRequestWrapper path_url property.""" + wrapper = OCIRequestWrapper( + method="POST", + url="https://example.com/api/v1/chat?param1=value1¶m2=value2", + headers={}, + body=b"test" + ) + + assert wrapper.path_url == "/api/v1/chat?param1=value1¶m2=value2" + + def test_oci_request_wrapper_path_url_no_query(self): + """Test OCIRequestWrapper path_url property without query string.""" + wrapper = OCIRequestWrapper( + method="POST", + url="https://example.com/api/v1/chat", + headers={}, + body=b"test" + ) + + assert wrapper.path_url == "/api/v1/chat"