Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,11 @@ omit = [
"src/oci_genai_auth/__about__.py",
]

[tool.pytest.ini_options]
markers = [
"integration: live OCI endpoint tests (require OCI_GENAI_* env vars)",
]

[tool.coverage.paths]
oci_genai_auth = ["src/oci_genai_auth", "*/oci-genai-auth/src/oci_genai_auth"]
tests = ["tests", "*/oci-genai-auth/tests"]
Expand Down
4 changes: 2 additions & 2 deletions src/oci_genai_auth/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@

__all__ = [
"HttpxOciAuth",
"OciSessionAuth",
"OciResourcePrincipalAuth",
"OciInstancePrincipalAuth",
"OciResourcePrincipalAuth",
"OciSessionAuth",
"OciUserPrincipalAuth",
]
34 changes: 26 additions & 8 deletions src/oci_genai_auth/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ class HttpxOciAuth(httpx.Auth, ABC):
refresh_interval: Seconds between token refreshes (default: 3600 - 1 hour)
_lock: Threading lock for thread-safe token refresh
_last_refresh: Last refresh timestamp
_last_refresh_error: The last refresh exception, if any (None on success)
"""

def __init__(self, signer: OciAuthSigner, refresh_interval: int = 3600):
Expand All @@ -46,7 +47,8 @@ def __init__(self, signer: OciAuthSigner, refresh_interval: int = 3600):
self.refresh_interval = refresh_interval
self._lock = threading.Lock()
self._last_refresh: Optional[float] = time.time()
logger.info(
self._last_refresh_error: Optional[Exception] = None
logger.debug(
"Initialized %s with refresh interval: %d seconds",
self.__class__.__name__,
refresh_interval,
Expand Down Expand Up @@ -76,13 +78,20 @@ def _refresh_if_needed(self) -> OciAuthSigner:
"""
with self._lock:
if self._should_refresh_token():
logger.info("Time interval reached, refreshing %s ...", self.__class__.__name__)
logger.debug("Time interval reached, refreshing %s ...", self.__class__.__name__)
try:
self._refresh_signer()
self._last_refresh = time.time()
self._last_refresh_error = None
logger.info("%s token refresh completed successfully", self.__class__.__name__)
except Exception:
logger.exception("Token refresh failed")
except Exception as exc:
self._last_refresh_error = exc
logger.warning(
"Scheduled token refresh failed for %s, "
"continuing with existing signer: %s",
self.__class__.__name__,
exc,
)
return self.signer

def _sign_request(self, request: httpx.Request, content: bytes, signer: OciAuthSigner) -> None:
Expand Down Expand Up @@ -112,6 +121,8 @@ def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Re
2. Signs the request using OCI signer
3. Yields the signed request
4. If 401 error is received, attempts token refresh and retries once
5. If retry refresh also fails, the generator ends and the caller
receives the original 401 rather than a silently dropped response
Args:
request: The HTTPX request to be authenticated
Yields:
Expand All @@ -138,11 +149,18 @@ def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Re
try:
self._refresh_signer()
self._last_refresh = time.time()
self._last_refresh_error = None
signer = self.signer
self._sign_request(request, content, signer)
yield request
except Exception:
logger.exception("Token refresh on 401 failed")
except Exception as exc:
self._last_refresh_error = exc
logger.error(
"Token refresh on 401 failed for %s: %s. "
"The original 401 response will be returned to the caller.",
self.__class__.__name__,
exc,
)


class OciSessionAuth(HttpxOciAuth):
Expand Down Expand Up @@ -231,13 +249,13 @@ def _load_token(self, config: Mapping[str, Any]) -> str:
with open(token_file, "r") as f:
return f.read().strip()

def _load_private_key(self, config: Any) -> str:
def _load_private_key(self, config: Any) -> Any:
"""
Load private key from file specified in configuration.
Args:
config: OCI configuration dictionary
Returns:
Private key object
Private key object (RSA/EC key from cryptography library)
"""
return oci.signer.load_private_key_from_file(config["key_file"])

Expand Down
52 changes: 52 additions & 0 deletions tests/.env.example
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# OCI GenAI Auth - Integration Test Configuration
#
# Copy this file to tests/.env and fill in your values.
# These variables are used by the integration test suite.
# Without this file (or without these env vars), integration tests
# are automatically skipped -- unit tests still run.
#
# IMPORTANT: Never commit tests/.env -- it is gitignored.
#
# ── Prerequisites ────────────────────────────────────────────────────────
#
# 1. An OCI tenancy with Generative AI enabled.
#
# 2. An OCI config profile in ~/.oci/config with either:
# - API key auth (auth_type=user_principal) or
# - Session token auth (auth_type=session, run `oci session authenticate` first)
#
# 3. A GenAI Project. Create one in the OCI Console:
# Console > Analytics & AI > Generative AI > Projects > Create Project
# Or via CLI:
# oci generative-ai generative-ai-project create \
# --compartment-id <compartment-ocid> \
# --display-name "test-project" \
# --profile <profile-name> --region us-chicago-1
#
# 4. A model available on the /openai/v1 endpoint.
# Known working models: xai.grok-3-mini-fast, xai.grok-4-1-fast-reasoning,
# google.gemini-2.5-flash, openai.gpt-5.2 (if available on your tenancy).
# Known NOT working on /openai/v1: meta.llama-*, cohere.command-* (return 404).

# ── OCI Authentication ───────────────────────────────────────────────────
# Which OCI config profile to use (must exist in ~/.oci/config).
OCI_GENAI_PROFILE=DEFAULT

# Auth type: "session" (SecurityTokenSigner) or "user_principal" (API key signer).
OCI_GENAI_AUTH_TYPE=session

# ── OCI GenAI Project ────────────────────────────────────────────────────
# Required. The GenAI Project OCID for the /openai/v1 endpoint.
OCI_GENAI_PROJECT_ID=ocid1.generativeaiproject.oc1.us-chicago-1.aaaaaaaaexample

# Optional. Compartment OCID (required for /chat/completions endpoint).
OCI_GENAI_COMPARTMENT_ID=ocid1.compartment.oc1..aaaaaaaaexample

# ── Region & Model ───────────────────────────────────────────────────────
# OCI region where GenAI is enabled.
OCI_GENAI_REGION=us-chicago-1

# Model to use in integration tests. Must be available on your tenancy's
# /openai/v1 endpoint. xai.grok-3-mini-fast is recommended (fast, cheap,
# available on most tenancies).
OCI_GENAI_MODEL=xai.grok-3-mini-fast
89 changes: 88 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,22 @@
# Copyright (c) 2026 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/

"""Shared fixtures for both unit and integration tests."""

from __future__ import annotations

import os
from pathlib import Path

import pytest

# ---------------------------------------------------------------------------
# Global: disable noisy tracing from OpenAI Agents SDK
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True, scope="session")
def _disable_openai_agents_tracing():
# Prevent OpenAI Agents tracing from emitting external HTTP requests during tests.
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "true")
try:
from agents.tracing import set_tracing_disabled
Expand All @@ -17,3 +25,82 @@ def _disable_openai_agents_tracing():
return
set_tracing_disabled(True)
yield


# ---------------------------------------------------------------------------
# Integration test environment
# ---------------------------------------------------------------------------


def _load_env():
"""Load tests/.env if present (plain KEY=VALUE, no shell expansion)."""
env_file = Path(__file__).parent / ".env"
if not env_file.exists():
return
for line in env_file.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
key, _, value = line.partition("=")
key, value = key.strip(), value.strip()
if key:
os.environ.setdefault(key, value)


_load_env()


# Required env vars for integration tests
_REQUIRED_VARS = (
"OCI_GENAI_PROJECT_ID",
"OCI_GENAI_REGION",
"OCI_GENAI_MODEL",
"OCI_GENAI_PROFILE",
"OCI_GENAI_AUTH_TYPE",
)


def _env(var: str) -> str:
return os.environ.get(var, "")


def _integration_configured() -> bool:
"""Return True if all required env vars are set to non-placeholder values."""
return all(_env(v) and "example" not in _env(v).lower() for v in _REQUIRED_VARS)


# Marker: skip integration tests when env is not configured
requires_oci = pytest.mark.skipif(
not _integration_configured(),
reason="Integration tests require OCI_GENAI_* env vars (see tests/.env.example)",
)


@pytest.fixture(scope="session")
def oci_project_id():
return _env("OCI_GENAI_PROJECT_ID")


@pytest.fixture(scope="session")
def oci_compartment_id():
return _env("OCI_GENAI_COMPARTMENT_ID")


@pytest.fixture(scope="session")
def oci_region():
return _env("OCI_GENAI_REGION")


@pytest.fixture(scope="session")
def oci_model():
return _env("OCI_GENAI_MODEL")


@pytest.fixture(scope="session")
def oci_profile():
return _env("OCI_GENAI_PROFILE")


@pytest.fixture(scope="session")
def oci_auth_type():
return _env("OCI_GENAI_AUTH_TYPE")
2 changes: 2 additions & 0 deletions tests/integration/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
# Copyright (c) 2026 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
Loading
Loading