diff --git a/litellm/llms/oci/common_utils.py b/litellm/llms/oci/common_utils.py index 3f703564b5a..679a9c21e43 100644 --- a/litellm/llms/oci/common_utils.py +++ b/litellm/llms/oci/common_utils.py @@ -1,16 +1,21 @@ import base64 import hashlib +import importlib import json import os import re +from collections.abc import Mapping from dataclasses import dataclass from email.utils import formatdate -from typing import Final, Protocol +from pathlib import Path +from types import MappingProxyType +from typing import Final, Protocol, runtime_checkable from urllib.parse import urlparse import httpx -from pydantic import JsonValue +from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError, field_validator +from litellm._logging import verbose_logger from litellm.llms.base_llm.chat.transformation import BaseLLMException try: @@ -154,7 +159,7 @@ _OCI_KEY_ENV: Final = "OCI_KEY" _OCI_COMPARTMENT_ID_ENV: Final = "OCI_COMPARTMENT_ID" -def resolve_oci_credentials(optional_params: dict) -> dict: +def resolve_oci_credentials(optional_params: Mapping[str, object]) -> dict: """ Merge OCI credentials from optional_params (explicit, always wins) and environment variables (fallback). @@ -174,11 +179,140 @@ def resolve_oci_credentials(optional_params: dict) -> dict: } -_OCI_REGION_RE: Final = re.compile(r"^[a-z][a-z0-9-]{0,30}[a-z0-9]$") +_OCI_REGION_PATTERN: Final = r"^[a-z][a-z0-9-]{0,30}[a-z0-9]$" +_OCI_REALM_DOMAIN_PATTERN: Final = r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$" +_OCI_REGION_RE: Final = re.compile(_OCI_REGION_PATTERN) _OCI_ACTION_PATH_RE: Final = re.compile(rf"/{OCI_API_VERSION}/actions/[^/?#]+/?$") +_OCI_COMMERCIAL_REALM_DOMAIN: Final = "oraclecloud.com" +_OCI_INFERENCE_ENDPOINT_TEMPLATE: Final = "https://inference.generativeai.{region}.oci.{secondLevelDomain}" +_OCI_REGION_METADATA_ENV: Final = "OCI_REGION_METADATA" +_OCI_REGIONS_CONFIG_FILE: Final = "~/.oci/regions-config.json" +_OCID_REALM_RE: Final = re.compile(r"^ocid1\.[a-z0-9]+\.([a-z0-9]+)\.", re.IGNORECASE) +_OCI_REALM_DOMAINS: Final = MappingProxyType( + { + "oc1": "oraclecloud.com", + "oc2": "oraclegovcloud.com", + "oc3": "oraclegovcloud.com", + "oc4": "oraclegovcloud.uk", + "oc8": "oraclecloud8.com", + "oc9": "oraclecloud9.com", + "oc10": "oraclecloud10.com", + "oc14": "oraclecloud14.com", + "oc15": "oraclecloud15.com", + "oc19": "oraclecloud.eu", + "oc20": "oraclecloud20.com", + "oc21": "oraclecloud21.com", + "oc23": "oraclecloud23.com", + "oc24": "oraclecloud24.com", + "oc26": "oraclecloud26.com", + "oc29": "oraclecloud29.com", + "oc35": "oraclecloud35.com", + "oc42": "oraclecloud42.com", + "oc51": "oraclecloud51.com", + "oc52": "oraclecloud52.com", + } +) -def get_oci_base_url(optional_params: dict, api_base: str | None = None) -> str: +class OCIRegionMetadata(BaseModel): + """One entry of the OCI SDK's region metadata schema, as found in + ``~/.oci/regions-config.json`` (a JSON array) or ``OCI_REGION_METADATA`` (one object). + Values are lowercased before validation, as the SDK does.""" + + model_config = ConfigDict(frozen=True, extra="ignore") + + region_identifier: str = Field(alias="regionIdentifier", pattern=_OCI_REGION_PATTERN) + realm_domain_component: str = Field(alias="realmDomainComponent", pattern=_OCI_REALM_DOMAIN_PATTERN) + + @field_validator("region_identifier", "realm_domain_component", mode="before") + @classmethod + def _lowercase(cls, value: object) -> object: + return value.lower() if isinstance(value, str) else value + + +_JSON_ARRAY: Final = TypeAdapter(tuple[JsonValue, ...]) + + +def _validated_region_metadata(raw: JsonValue, source: str) -> OCIRegionMetadata | None: + try: + return OCIRegionMetadata.model_validate(raw) + except ValidationError as e: + verbose_logger.warning("Ignoring OCI region metadata entry in %s: %s", source, e) + return None + + +def _region_metadata_from_file() -> tuple[OCIRegionMetadata, ...]: + path: Final = Path(os.path.expanduser(_OCI_REGIONS_CONFIG_FILE)) + if not path.is_file(): + return () + try: + raw_entries: Final = _JSON_ARRAY.validate_json(path.read_bytes()) + except (OSError, ValidationError) as e: + verbose_logger.warning("Ignoring OCI region metadata in %s: %s", path, e) + return () + candidates: Final = (_validated_region_metadata(raw, str(path)) for raw in raw_entries) + return tuple(entry for entry in candidates if entry is not None) + + +def _region_metadata_from_env() -> tuple[OCIRegionMetadata, ...]: + raw: Final = os.environ.get(_OCI_REGION_METADATA_ENV) + if not raw: + return () + try: + return (OCIRegionMetadata.model_validate_json(raw),) + except ValidationError as e: + verbose_logger.warning("Ignoring OCI region metadata in %s: %s", _OCI_REGION_METADATA_ENV, e) + return () + + +def _realm_domain_from_ocid(ocid: str | None) -> str | None: + match: Final = _OCID_REALM_RE.match(ocid) if ocid else None + return _OCI_REALM_DOMAINS.get(match.group(1).lower()) if match else None + + +def _realm_domain_from_metadata(region: str) -> str | None: + entries: Final = (*_region_metadata_from_file(), *_region_metadata_from_env()) + return next((entry.realm_domain_component for entry in entries if entry.region_identifier == region), None) + + +@runtime_checkable +class _OCIRegionRegistry(Protocol): + def endpoint_for(self, service: str, region: str, service_endpoint_template: str) -> str: ... + + +def _load_oci_region_registry() -> _OCIRegionRegistry | None: + try: + registry: Final = importlib.import_module("oci.regions") + except ImportError: + return None + return registry if isinstance(registry, _OCIRegionRegistry) else None + + +def resolve_oci_inference_endpoint(region: str, compartment_id: str | None = None) -> str: + """Return the GenAI inference endpoint for ``region`` in whichever OCI realm hosts it. + + The realm's second-level domain comes first from the realm key inside ``compartment_id`` + (``ocid1.compartment.oc2..`` is the Government realm), then from the OCI SDK's region + registry when the SDK is installed, then from the per-region metadata sources the SDK + reads, ``~/.oci/regions-config.json`` and ``OCI_REGION_METADATA``, and otherwise defaults + to the commercial realm. Realm domains per ``oci/regions_definitions.py`` in oci 2.187.0. + A region that is not described anywhere therefore keeps its commercial endpoint, so one + government deployment never redirects the others. + """ + realm_domain: Final = _realm_domain_from_ocid(compartment_id) + if realm_domain is not None: + return _OCI_INFERENCE_ENDPOINT_TEMPLATE.format(region=region, secondLevelDomain=realm_domain) + registry: Final = _load_oci_region_registry() + if registry is not None: + return registry.endpoint_for( + "generative_ai_inference", region=region, service_endpoint_template=_OCI_INFERENCE_ENDPOINT_TEMPLATE + ) + return _OCI_INFERENCE_ENDPOINT_TEMPLATE.format( + region=region, secondLevelDomain=_realm_domain_from_metadata(region) or _OCI_COMMERCIAL_REALM_DOMAIN + ) + + +def get_oci_base_url(optional_params: Mapping[str, object], api_base: str | None = None) -> str: """Return the OCI inference base URL, respecting any explicit api_base override. If ``api_base`` already ends with a fully-formed OCI action path @@ -196,7 +330,8 @@ def get_oci_base_url(optional_params: dict, api_base: str | None = None) -> str: f"Invalid OCI region {region!r}: must match ^[a-z][a-z0-9-]{{0,30}}[a-z0-9]$ (e.g. 'us-ashburn-1')." ), ) - return f"https://inference.generativeai.{region}.oci.oraclecloud.com" + compartment_id: Final = creds["oci_compartment_id"] + return resolve_oci_inference_endpoint(region, compartment_id if isinstance(compartment_id, str) else None) # --------------------------------------------------------------------------- diff --git a/litellm/llms/oci/embed/transformation.py b/litellm/llms/oci/embed/transformation.py index 2300c6ee403..dec43717387 100644 --- a/litellm/llms/oci/embed/transformation.py +++ b/litellm/llms/oci/embed/transformation.py @@ -77,7 +77,11 @@ class OCIEmbedConfig(BaseEmbeddingConfig): Required call-time params (via optional_params or env vars): - ``oci_compartment_id`` / ``OCI_COMPARTMENT_ID`` - - ``oci_region`` / ``OCI_REGION`` (default: ``us-ashburn-1``) + - ``oci_region`` / ``OCI_REGION`` (default: ``us-ashburn-1``). The realm comes from the realm + key in ``oci_compartment_id`` (``ocid1.compartment.oc2..`` is the Government realm), so + non-commercial realms need no extra setting. A realm unknown to litellm can be described in + ``OCI_REGION_METADATA`` or ``~/.oci/regions-config.json``, resolved through the OCI SDK when + it is installed, or given as ``api_base``. Optional call-time params: - ``oci_serving_mode``: ``"ON_DEMAND"`` (default) or ``"DEDICATED"`` diff --git a/tests/unit/llms/oci/test_oci_common_utils.py b/tests/unit/llms/oci/test_oci_common_utils.py index d306d7351dd..e66645c4dcd 100644 --- a/tests/unit/llms/oci/test_oci_common_utils.py +++ b/tests/unit/llms/oci/test_oci_common_utils.py @@ -5,10 +5,16 @@ Covers schema utilities, signing helpers, and credential resolution paths that require no real OCI credentials or network calls. """ -import pytest +import sys +import types +from types import MappingProxyType +from typing import Final from unittest.mock import MagicMock, patch +import pytest + from litellm.llms.oci.common_utils import ( + _OCI_REALM_DOMAINS, OCI_API_VERSION, OCIError, OCIRequestWrapper, @@ -40,7 +46,8 @@ def test_oci_api_version_constant(): def test_sha256_base64_known_value(): - import base64, hashlib + import base64 + import hashlib data = b"hello" expected = base64.b64encode(hashlib.sha256(data).digest()).decode() @@ -60,9 +67,7 @@ def test_sha256_base64_empty(): def test_build_signature_string_request_target(): headers = {"host": "example.com", "date": "Mon, 01 Jan 2024 00:00:00 GMT"} - result = build_signature_string( - "POST", "/20231130/actions/chat", headers, ["(request-target)", "host", "date"] - ) + result = build_signature_string("POST", "/20231130/actions/chat", headers, ["(request-target)", "host", "date"]) lines = result.split("\n") assert lines[0] == "(request-target): post /20231130/actions/chat" assert lines[1] == "host: example.com" @@ -161,12 +166,10 @@ def test_get_oci_base_url_explicit_api_base(): ], ) def test_get_oci_base_url_strips_trailing_action_path(api_base): - assert ( - get_oci_base_url({}, api_base=api_base) - == "https://inference.generativeai.us-chicago-1.oci.oraclecloud.com" - ) + assert get_oci_base_url({}, api_base=api_base) == "https://inference.generativeai.us-chicago-1.oci.oraclecloud.com" +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") def test_get_oci_base_url_from_region(): url = get_oci_base_url({"oci_region": "eu-frankfurt-1"}) assert url == "https://inference.generativeai.eu-frankfurt-1.oci.oraclecloud.com" @@ -192,6 +195,7 @@ def test_get_oci_base_url_rejects_unsafe_region(region): get_oci_base_url({"oci_region": region}) +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") def test_get_oci_base_url_empty_region_falls_back_to_default(monkeypatch): monkeypatch.delenv("OCI_REGION", raising=False) url = get_oci_base_url({"oci_region": ""}) @@ -209,11 +213,248 @@ def test_get_oci_base_url_empty_region_falls_back_to_default(monkeypatch): "ap", ], ) +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") def test_get_oci_base_url_accepts_valid_region(region): url = get_oci_base_url({"oci_region": region}) assert url == f"https://inference.generativeai.{region}.oci.oraclecloud.com" +_NON_COMMERCIAL_REALMS: Final = ( + ("oc2", "us-luke-1", "oraclegovcloud.com"), + ("oc3", "us-gov-ashburn-1", "oraclegovcloud.com"), + ("oc4", "uk-gov-london-1", "oraclegovcloud.uk"), + ("oc19", "eu-frankfurt-2", "oraclecloud.eu"), +) +_UNKNOWN_REGION: Final = "xx-nowhere-1" +_UNKNOWN_REALM_COMPARTMENT: Final = "ocid1.compartment.oc99..aaaaaaaaexample" +_UNKNOWN_REGION_METADATA: Final = '{"realmKey": "OCX", "realmDomainComponent": "example.test", "regionKey": "XNW", "regionIdentifier": "xx-nowhere-1"}' + + +def _compartment(realm): + return f"ocid1.compartment.{realm}..aaaaaaaaexample" + + +def _params(region: str, compartment_id: object = None) -> MappingProxyType[str, object]: + return MappingProxyType({"oci_region": region, "oci_compartment_id": compartment_id}) + + +@pytest.fixture +def without_oci_sdk(monkeypatch): + monkeypatch.setitem(sys.modules, "oci", None) + monkeypatch.setitem(sys.modules, "oci.regions", None) + + +@pytest.fixture +def isolated_region_metadata(monkeypatch, tmp_path): + monkeypatch.delenv("OCI_REGION_METADATA", raising=False) + monkeypatch.delenv("OCI_COMPARTMENT_ID", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + return tmp_path + + +def test_realm_table_matches_installed_sdk(): + # Realm domains per the OCI Python SDK's oci.regions_definitions.REALMS (v2.187.0, checked 2026-09-27) + definitions: Final = pytest.importorskip("oci.regions_definitions") + assert ( + MappingProxyType({realm: definitions.REALMS.get(realm) for realm in _OCI_REALM_DOMAINS}) == _OCI_REALM_DOMAINS + ) + + +@pytest.mark.usefixtures("isolated_region_metadata") +@pytest.mark.parametrize(("realm", "region", "second_level_domain"), _NON_COMMERCIAL_REALMS) +def test_get_oci_base_url_resolves_realm_from_region_via_sdk(realm, region, second_level_domain): + pytest.importorskip("oci.regions") + # Realm domains per the OCI Python SDK's oci.regions_definitions (v2.187.0, checked 2026-09-27) + url: Final = get_oci_base_url(_params(region)) + assert url == f"https://inference.generativeai.{region}.oci.{second_level_domain}" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +@pytest.mark.parametrize(("realm", "region", "second_level_domain"), _NON_COMMERCIAL_REALMS) +def test_get_oci_base_url_resolves_realm_from_compartment_ocid(realm, region, second_level_domain): + url: Final = get_oci_base_url(_params(region, _compartment(realm))) + assert url == f"https://inference.generativeai.{region}.oci.{second_level_domain}" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_resolves_realm_from_compartment_env(monkeypatch): + monkeypatch.setenv("OCI_COMPARTMENT_ID", _compartment("oc2")) + url: Final = get_oci_base_url(_params("us-luke-1")) + assert url == "https://inference.generativeai.us-luke-1.oci.oraclegovcloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_reads_realm_key_case_insensitively(): + url: Final = get_oci_base_url(_params("us-luke-1", _compartment("OC2"))) + assert url == "https://inference.generativeai.us-luke-1.oci.oraclegovcloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_keeps_commercial_compartment_commercial(): + url: Final = get_oci_base_url(_params("us-chicago-1", _compartment("oc1"))) + assert url == "https://inference.generativeai.us-chicago-1.oci.oraclecloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +@pytest.mark.parametrize("compartment_id", (None, "not-an-ocid", _UNKNOWN_REALM_COMPARTMENT, 42)) +def test_get_oci_base_url_without_sdk_defaults_to_commercial_when_realm_unknown(compartment_id): + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION, compartment_id)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.oraclecloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_compartment_realm_wins_over_region_metadata(monkeypatch): + monkeypatch.setenv( + "OCI_REGION_METADATA", '{"regionIdentifier": "us-luke-1", "realmDomainComponent": "example.test"}' + ) + url: Final = get_oci_base_url(_params("us-luke-1", _compartment("oc2"))) + assert url == "https://inference.generativeai.us-luke-1.oci.oraclegovcloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_without_sdk_uses_region_metadata_env(monkeypatch): + monkeypatch.setenv("OCI_REGION_METADATA", _UNKNOWN_REGION_METADATA) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION, _UNKNOWN_REALM_COMPARTMENT)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.example.test" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +def test_get_oci_base_url_without_sdk_region_metadata_leaves_other_regions_commercial(monkeypatch): + monkeypatch.setenv("OCI_REGION_METADATA", _UNKNOWN_REGION_METADATA) + url: Final = get_oci_base_url(_params("us-chicago-1")) + assert url == "https://inference.generativeai.us-chicago-1.oci.oraclecloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk") +def test_get_oci_base_url_without_sdk_uses_regions_config_file(isolated_region_metadata): + oci_dir: Final = isolated_region_metadata / ".oci" + oci_dir.mkdir() + (oci_dir / "regions-config.json").write_text(f"[{_UNKNOWN_REGION_METADATA}]") + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.example.test" + + +@pytest.mark.usefixtures("without_oci_sdk") +def test_get_oci_base_url_without_sdk_keeps_valid_regions_config_entries_next_to_a_bad_one(isolated_region_metadata): + oci_dir: Final = isolated_region_metadata / ".oci" + oci_dir.mkdir() + (oci_dir / "regions-config.json").write_text( + f'[{{"regionIdentifier": "us-langley-1"}}, {_UNKNOWN_REGION_METADATA}]' + ) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.example.test" + + +@pytest.mark.usefixtures("without_oci_sdk") +@pytest.mark.parametrize("content", (b"\xff\xfe\x00[", b'{"regionIdentifier": "xx-nowhere-1"}', b"not json")) +def test_get_oci_base_url_without_sdk_ignores_unusable_regions_config_file(isolated_region_metadata, content): + oci_dir: Final = isolated_region_metadata / ".oci" + oci_dir.mkdir() + (oci_dir / "regions-config.json").write_bytes(content) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.oraclecloud.com" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +@pytest.mark.parametrize( + "metadata", + ( + '{"regionIdentifier": "xx-nowhere-1", "realmDomainComponent": "evil.com/#"}', + '{"regionIdentifier": "xx-nowhere-1", "realmDomainComponent": "-internal"}', + '{"regionIdentifier": "xx-nowhere-1"}', + "not json", + ), +) +def test_get_oci_base_url_without_sdk_ignores_invalid_region_metadata(monkeypatch, metadata): + monkeypatch.setenv("OCI_REGION_METADATA", metadata) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.oraclecloud.com" + + +def _fake_oci_regions(endpoint_for=None): + module: Final = types.ModuleType("oci.regions") + if endpoint_for is not None: + module.endpoint_for = endpoint_for + return module + + +@pytest.mark.usefixtures("isolated_region_metadata") +def test_get_oci_base_url_uses_sdk_region_registry_when_realm_unknown(monkeypatch): + endpoint_for: Final = MagicMock( + side_effect=lambda service, region, service_endpoint_template: service_endpoint_template.format( + region=region, secondLevelDomain="example.test" + ) + ) + monkeypatch.setitem(sys.modules, "oci", types.ModuleType("oci")) + monkeypatch.setitem(sys.modules, "oci.regions", _fake_oci_regions(endpoint_for)) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION, _UNKNOWN_REALM_COMPARTMENT)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.example.test" + endpoint_for.assert_called_once_with( + "generative_ai_inference", + region=_UNKNOWN_REGION, + service_endpoint_template="https://inference.generativeai.{region}.oci.{secondLevelDomain}", + ) + + +@pytest.mark.usefixtures("isolated_region_metadata") +def test_get_oci_base_url_skips_sdk_region_registry_when_compartment_realm_known(monkeypatch): + def endpoint_for(service, region, service_endpoint_template): + raise AssertionError("registry consulted") + + monkeypatch.setitem(sys.modules, "oci", types.ModuleType("oci")) + monkeypatch.setitem(sys.modules, "oci.regions", _fake_oci_regions(endpoint_for)) + url: Final = get_oci_base_url(_params("us-luke-1", _compartment("oc2"))) + assert url == "https://inference.generativeai.us-luke-1.oci.oraclegovcloud.com" + + +@pytest.mark.usefixtures("isolated_region_metadata") +def test_get_oci_base_url_prefers_sdk_region_registry_over_hand_parsed_metadata(monkeypatch): + def endpoint_for(service, region, service_endpoint_template): + return service_endpoint_template.format(region=region, secondLevelDomain="sdk.test") + + monkeypatch.setitem(sys.modules, "oci", types.ModuleType("oci")) + monkeypatch.setitem(sys.modules, "oci.regions", _fake_oci_regions(endpoint_for)) + monkeypatch.setenv("OCI_REGION_METADATA", _UNKNOWN_REGION_METADATA) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.sdk.test" + + +@pytest.mark.usefixtures("isolated_region_metadata") +def test_get_oci_base_url_falls_back_to_metadata_when_sdk_registry_lacks_endpoint_for(monkeypatch): + monkeypatch.setitem(sys.modules, "oci", types.ModuleType("oci")) + monkeypatch.setitem(sys.modules, "oci.regions", _fake_oci_regions()) + monkeypatch.setenv("OCI_REGION_METADATA", _UNKNOWN_REGION_METADATA) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.example.test" + + +@pytest.mark.usefixtures("without_oci_sdk", "isolated_region_metadata") +@pytest.mark.parametrize( + ("metadata", "second_level_domain"), + ( + ('{"regionIdentifier": "XX-NOWHERE-1", "realmDomainComponent": "Example.Test"}', "example.test"), + ('{"regionIdentifier": "xx-nowhere-1", "realmDomainComponent": "internal"}', "internal"), + ), +) +def test_get_oci_base_url_without_sdk_normalizes_region_metadata_like_the_sdk( + monkeypatch, metadata, second_level_domain +): + monkeypatch.setenv("OCI_REGION_METADATA", metadata) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.{second_level_domain}" + + +@pytest.mark.usefixtures("without_oci_sdk") +def test_get_oci_base_url_without_sdk_tolerates_unresolvable_home(monkeypatch): + def no_passwd_entry(uid): + raise KeyError(uid) + + monkeypatch.delenv("HOME", raising=False) + monkeypatch.setattr("pwd.getpwuid", no_passwd_entry) + url: Final = get_oci_base_url(_params(_UNKNOWN_REGION)) + assert url == f"https://inference.generativeai.{_UNKNOWN_REGION}.oci.oraclecloud.com" + + # --------------------------------------------------------------------------- # validate_oci_environment # --------------------------------------------------------------------------- @@ -247,17 +488,13 @@ def test_sign_with_oci_signer_exception_wrapped(): bad_signer = MagicMock() bad_signer.do_request_sign.side_effect = RuntimeError("signing failed") with pytest.raises(OCIError, match="Failed to sign request"): - sign_with_oci_signer( - {}, {"oci_signer": bad_signer}, {"key": "val"}, "https://example.com" - ) + sign_with_oci_signer({}, {"oci_signer": bad_signer}, {"key": "val"}, "https://example.com") def test_sign_with_oci_signer_success(): signer = MagicMock() signer.do_request_sign.return_value = None - headers, body = sign_with_oci_signer( - {}, {"oci_signer": signer}, {"key": "val"}, "https://example.com" - ) + headers, body = sign_with_oci_signer({}, {"oci_signer": signer}, {"key": "val"}, "https://example.com") assert isinstance(body, bytes) signer.do_request_sign.assert_called_once() @@ -270,9 +507,7 @@ def test_sign_with_oci_signer_success(): def test_sign_oci_request_routes_to_signer(): signer = MagicMock() signer.do_request_sign.return_value = None - headers, body = sign_oci_request( - {}, {"oci_signer": signer}, {}, "https://example.com" - ) + headers, body = sign_oci_request({}, {"oci_signer": signer}, {}, "https://example.com") signer.do_request_sign.assert_called_once()