mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-04 02:31:27 +00:00
fix(oci): resolve the GenAI endpoint realm from the compartment OCID instead of hardcoding oraclecloud.com (#43180)
* fix(oci): resolve the GenAI endpoint realm from the region instead of hardcoding oraclecloud.com Government (OC2/OC3/OC4) and other non-commercial realms live under a different second-level domain, so a request for us-luke-1 was sent to inference.generativeai.us-luke-1.oci.oraclecloud.com and failed DNS Delegate the lookup to the OCI SDK's region registry when it is installed, honour OCI_DEFAULT_REALM otherwise, and keep api_base as the explicit override * fix(oci): read per-region realm metadata without the SDK and cover the registry path Replace the global OCI_DEFAULT_REALM fallback, which would have redirected commercial regions too in a mixed deployment, with the SDK's own per-region sources: OCI_REGION_METADATA and ~/.oci/regions-config.json. Regions not described anywhere keep their commercial endpoint Exercise the SDK registry path with a fake oci.regions module so CI, which has no SDK, still covers it, and skip the real-SDK test on oci.regions so a namespace package named oci in the tests tree cannot masquerade as the SDK * fix(oci): validate regions-config.json entries individually and tolerate undecodable files One malformed entry no longer discards the valid ones, and the file is parsed from bytes so an undecodable file is logged and ignored instead of failing every OCI request built without the SDK * fix(oci): resolve the realm from the compartment OCID so Government regions work without the SDK The Docker image ships without the oci package and Government deployments rarely carry OCI_REGION_METADATA, so the reviewed fallback still sent us-luke-1 to oraclecloud.com. Every compartment OCID already names its realm (ocid1.compartment.oc2..), so map that key through the SDK's twenty realm domains first, then the metadata sources, then the SDK registry, then the commercial default. * fix(oci): consult the SDK registry before hand-parsed region metadata and harden the fallback Review follow-ups on the realm resolver. Read the compartment realm first, then the SDK registry when it is installed, and only then the hand-parsed metadata sources, so the same file is never parsed twice with different rules. Lowercase metadata values like the SDK does, accept single-label realm domains, and expand ~ with os.path so a container without a home directory cannot raise out of URL building. Type the compartment as str | None at the caller, drop populate_by_name, isolate the legacy region tests from the developer's ~/.oci, and keep the new tests on the immutable style.
This commit is contained in:
Fede Kamelhar
GitHub
parent
3ae491a06c
commit
56bba4fbbe
3 changed files with 399 additions and 25 deletions
|
|
@ -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)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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"``
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue