scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 35
litellm/tests/integration/test_oci_integration.py
yuneng-jiang 6a0d03914c
test: drop the cwd-relative sys.path.insert calls from the test suite (#37802) 
* test: drop the cwd-relative sys.path.insert calls from the test suite

TQ003 stands at 1,077 across 1,058 files, and 1,015 of them are the same shape:
sys.path.insert(0, os.path.abspath("../..")) and its deeper siblings. The
argument resolves against the working directory rather than the file, so from
the repo root, where every job runs pytest, it inserts the directory two levels
above the checkout. It has never pointed at litellm. The package is installed
into the environment anyway, which is what actually makes the import work, and
what the rule's message has said all along.

Removing them leaves 1,634 imports of sys and os with no remaining reference,
and those go too, except where another test module imports the name back out of
the file. The rest of TQ003 is 62 call sites that resolve against __file__ or a
variable, which are a different question and are left alone.

Collection is identical either way: 45,871 tests and the same 51 pre-existing
collection errors before and after, and ruff reports no new undefined name.

* test: drop the duplicate imports the sys.path sweep exposed to F811

* test(pre-call-utils): restore the os import the new bedrock tests need
2026-08-22 09:25:58 -07:00

667 lines
22 KiB
Python
Raw Permalink Blame History

"""
OCI Generative AI — end-to-end integration tests.
These tests make REAL calls to OCI. They are skipped automatically when the
standard ~/.oci/config is absent or when OCI_TEST_COMPARTMENT_ID is not set.
Prerequisites
-------------
- ~/.oci/config with a valid [DEFAULT] profile
- Private key referenced by key_file in that profile
- Sufficient IAM policies to call the Generative AI inference service
Environment variables (all optional — fall back to ~/.oci/config values):
OCI_TEST_REGION OCI region (default: us-chicago-1)
OCI_TEST_COMPARTMENT_ID compartment OCID (default: tenancy root from config)
Run only these tests:
pytest tests/integration/test_oci_integration.py -v
"""
import math
import os
from typing import NamedTuple, Optional
import pytest
# ---------------------------------------------------------------------------
# Fixtures / helpers
# ---------------------------------------------------------------------------
OCI_CONFIG_FILE = os.path.expanduser("~/.oci/config")
_OCI_AVAILABLE = os.path.isfile(OCI_CONFIG_FILE)
pytestmark = pytest.mark.skipif(
not _OCI_AVAILABLE,
reason="~/.oci/config not found — skipping OCI integration tests",
)
def _load_oci_config():
"""Load OCI config from the profile named by ``OCI_CONFIG_PROFILE`` env var,
falling back to ``[DEFAULT]``. Lets CI/local runs target a specific profile
without needing a ``[DEFAULT]`` section in ``~/.oci/config``."""
oci = pytest.importorskip("oci")
profile = os.environ.get("OCI_CONFIG_PROFILE", "DEFAULT")
return oci.config.from_file(profile_name=profile)
@pytest.fixture(scope="module")
def oci_signer():
"""Return an oci.Signer (or SecurityTokenSigner for session-token profiles)
built from ~/.oci/config — profile chosen via OCI_CONFIG_PROFILE."""
oci = pytest.importorskip("oci")
config = _load_oci_config()
# Session-token profiles carry a `security_token_file` instead of a user
# OCID; build the corresponding signer in that case.
if "security_token_file" in config:
with open(os.path.expanduser(config["security_token_file"])) as f:
token = f.read().strip()
private_key = oci.signer.load_private_key_from_file(
config["key_file"], config.get("pass_phrase")
)
return oci.auth.signers.SecurityTokenSigner(token, private_key)
return oci.Signer(
tenancy=config["tenancy"],
user=config["user"],
fingerprint=config["fingerprint"],
private_key_file_location=config["key_file"],
)
@pytest.fixture(scope="module")
def oci_params(oci_signer) -> dict:
"""Common OCI call-time parameters shared by all tests."""
config = _load_oci_config()
compartment_id = os.environ.get("OCI_TEST_COMPARTMENT_ID", config["tenancy"])
region = os.environ.get("OCI_TEST_REGION", "us-chicago-1")
return {
"oci_signer": oci_signer,
"oci_compartment_id": compartment_id,
"oci_region": region,
}
# ---------------------------------------------------------------------------
# Model registry
#
# Each entry drives the runtime pivot inside OCI's own transformation layer —
# the tests themselves are format-agnostic. Per-model quirks are captured in
# the config fields below rather than in separate test classes.
# ---------------------------------------------------------------------------
class _M(NamedTuple):
"""Per-model test configuration."""
model: str
max_tokens: int
# Reasoning models (Gemini 2.5, Grok mini) may return None content when the
# reasoning budget is exhausted before the answer token budget starts.
reasoning: bool = False
# tool_choice value to send; None means omit the parameter entirely.
tool_choice: Optional[str] = "auto"
# Whether to include the model in tool-use parametrize list.
supports_tool_use: bool = True
# All chat models under test.
CHAT_MODELS = [
pytest.param(_M("meta.llama-3.3-70b-instruct", 64), id="meta"),
pytest.param(_M("google.gemini-2.5-flash", 200, reasoning=True), id="google"),
pytest.param(_M("xai.grok-3-mini", 100, reasoning=True), id="xai"),
pytest.param(_M("cohere.command-latest", 64, tool_choice=None), id="cohere"),
]
# Subset of models that reliably support tool use in OCI.
# xAI Grok mini is omitted — OCI does not expose tool-use for it yet.
TOOL_USE_MODELS = [
pytest.param(_M("meta.llama-3.3-70b-instruct", 100), id="meta"),
pytest.param(_M("cohere.command-latest", 200, tool_choice=None), id="cohere"),
pytest.param(_M("google.gemini-2.5-flash", 200, reasoning=True), id="google"),
]
# Simple weather tool used by all tool-use tests.
_WEATHER_TOOL = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a city.",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string", "description": "The city name."}},
"required": ["city"],
},
},
}
# ---------------------------------------------------------------------------
# Sync chat tests — model list drives the pivot, not separate test classes
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("m", CHAT_MODELS)
def test_basic_completion(m: _M, oci_params):
import litellm
resp = litellm.completion(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "Reply with only the word: pong"}],
max_tokens=m.max_tokens,
**oci_params,
)
assert resp.choices[0].finish_reason is not None
assert resp.usage.prompt_tokens > 0
if not m.reasoning:
assert resp.choices[0].message.content is not None
@pytest.mark.parametrize("m", CHAT_MODELS)
def test_usage_populated(m: _M, oci_params):
import litellm
resp = litellm.completion(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "What is 2+2?"}],
max_tokens=m.max_tokens,
**oci_params,
)
assert resp.usage.prompt_tokens > 0
assert resp.usage.total_tokens >= resp.usage.prompt_tokens
@pytest.mark.parametrize("m", CHAT_MODELS)
def test_system_message(m: _M, oci_params):
import litellm
resp = litellm.completion(
model=f"oci/{m.model}",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Say hello."},
],
max_tokens=m.max_tokens,
**oci_params,
)
assert resp.choices[0].finish_reason is not None
@pytest.mark.parametrize("m", CHAT_MODELS)
def test_streaming(m: _M, oci_params):
import litellm
chunks = list(
litellm.completion(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "Count to 3."}],
max_tokens=m.max_tokens,
stream=True,
**oci_params,
)
)
assert len(chunks) > 0
# Reasoning models may stream only reasoning tokens and return empty content.
if not m.reasoning:
content = "".join(c.choices[0].delta.content or "" for c in chunks if c.choices)
assert len(content) > 0
@pytest.mark.parametrize(
"model",
["cohere.command-latest", "cohere.command-r-plus-08-2024"],
)
def test_cohere_streaming_no_doubling(model, oci_params):
"""Regression: OCI Cohere's terminal SSE event re-sends the full assembled
response in `text` alongside a populated `chatHistory`. Emitting that text
as another delta would concatenate the whole response onto the
already-streamed output (e.g. "How can I help?How can I help?").
Reported by @gotsysdba on PR #25177. Fix: drop terminal text when
`chatHistory` is present in `handle_cohere_stream_chunk`.
"""
import litellm
streamed = "".join(
(c.choices[0].delta.content or "")
for c in litellm.completion(
model=f"oci/{model}",
messages=[{"role": "user", "content": "Hello!"}],
max_tokens=64,
stream=True,
**oci_params,
)
if c.choices
).strip()
assert streamed, "expected non-empty streamed content"
# Compare against a non-streamed call. With the doubling bug the streamed
# assembly is ~2x the real response; without it the two are the same order
# of magnitude (the model is non-deterministic, so allow generous slack).
non_streamed = (
litellm.completion(
model=f"oci/{model}",
messages=[{"role": "user", "content": "Hello!"}],
max_tokens=64,
**oci_params,
)
.choices[0]
.message.content
or ""
).strip()
assert len(streamed) < 2 * len(non_streamed) + 10, (
f"streamed output appears doubled — "
f"streamed={len(streamed)} chars vs non_streamed={len(non_streamed)} chars\n"
f"streamed: {streamed!r}\n"
f"non_streamed: {non_streamed!r}"
)
# Stronger signal: the very start of the response should not appear twice.
head = streamed[:12]
assert streamed.count(head) == 1, (
f"streamed output contains its own prefix {head!r} more than once — "
f"likely the terminal chunk re-emitted the full response.\n"
f"streamed: {streamed!r}"
)
@pytest.mark.parametrize("m", CHAT_MODELS)
def test_multi_turn(m: _M, oci_params):
import litellm
resp = litellm.completion(
model=f"oci/{m.model}",
messages=[
{"role": "user", "content": "My name is Alice."},
{"role": "assistant", "content": "Nice to meet you, Alice!"},
{"role": "user", "content": "What is my name?"},
],
max_tokens=m.max_tokens,
**oci_params,
)
# Reasoning models may have None content; skip text assertion for them.
content = resp.choices[0].message.content or ""
if not m.reasoning:
assert "Alice" in content
# ---------------------------------------------------------------------------
# Async chat tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize("m", CHAT_MODELS)
async def test_async_completion(m: _M, oci_params):
import litellm
resp = await litellm.acompletion(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "Reply with only the word: pong"}],
max_tokens=m.max_tokens,
**oci_params,
)
assert resp.choices[0].finish_reason is not None
assert resp.usage.total_tokens > 0
if not m.reasoning:
assert resp.choices[0].message.content is not None
@pytest.mark.asyncio
@pytest.mark.parametrize("m", CHAT_MODELS)
async def test_async_streaming(m: _M, oci_params):
import litellm
chunks = []
async for chunk in await litellm.acompletion(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "Count to 3."}],
max_tokens=m.max_tokens,
stream=True,
**oci_params,
):
chunks.append(chunk)
assert len(chunks) > 0
if not m.reasoning:
content = "".join(c.choices[0].delta.content or "" for c in chunks if c.choices)
assert len(content) > 0
# ---------------------------------------------------------------------------
# Tool-use tests
# ---------------------------------------------------------------------------
def _assert_tool_call(resp, expected_tool: str = "get_weather"):
"""Assert the response contains the expected tool call (or a plain stop)."""
choice = resp.choices[0]
assert choice.finish_reason in ("tool_calls", "stop")
if choice.finish_reason == "tool_calls":
assert choice.message.tool_calls is not None
assert len(choice.message.tool_calls) > 0
assert choice.message.tool_calls[0].function.name == expected_tool
@pytest.mark.parametrize("m", TOOL_USE_MODELS)
def test_tool_use(m: _M, oci_params):
import litellm
call_kwargs = dict(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "What is the weather in Paris?"}],
tools=[_WEATHER_TOOL],
max_tokens=m.max_tokens,
**oci_params,
)
if m.tool_choice is not None:
call_kwargs["tool_choice"] = m.tool_choice
resp = litellm.completion(**call_kwargs)
_assert_tool_call(resp)
@pytest.mark.asyncio
@pytest.mark.parametrize("m", TOOL_USE_MODELS)
async def test_async_tool_use(m: _M, oci_params):
import litellm
call_kwargs = dict(
model=f"oci/{m.model}",
messages=[{"role": "user", "content": "What is the weather in Berlin?"}],
tools=[_WEATHER_TOOL],
max_tokens=m.max_tokens,
**oci_params,
)
if m.tool_choice is not None:
call_kwargs["tool_choice"] = m.tool_choice
resp = await litellm.acompletion(**call_kwargs)
_assert_tool_call(resp)
# ---------------------------------------------------------------------------
# Reasoning-effort tests (reasoning models only)
# ---------------------------------------------------------------------------
# Reasoning model that accepts the `reasoningEffort` parameter on OCI.
# Not every reasoning model does — xai.grok-4-fast-reasoning, for example,
# rejects it with a 400.
_REASONING_MODEL = "xai.grok-3-mini"
@pytest.mark.parametrize("effort", ["low", "medium", "high"])
def test_reasoning_effort_lowercase_accepted(effort, oci_params):
"""OpenAI clients send lowercase reasoning_effort; OCI requires uppercase.
The transform layer should uppercase it transparently."""
import litellm
resp = litellm.completion(
model=f"oci/{_REASONING_MODEL}",
messages=[{"role": "user", "content": "What is 2+2? One word."}],
max_tokens=200,
reasoning_effort=effort,
**oci_params,
)
assert resp.choices[0].finish_reason is not None
assert resp.usage.prompt_tokens > 0
def test_reasoning_effort_disable_mapped_to_none(oci_params):
"""OpenAI's 'disable' maps to OCI's 'NONE'. Without this mapping the
request 400s."""
import litellm
resp = litellm.completion(
model=f"oci/{_REASONING_MODEL}",
messages=[{"role": "user", "content": "What is 2+2? One word."}],
max_tokens=200,
reasoning_effort="disable",
**oci_params,
)
assert resp.choices[0].finish_reason is not None
def test_reasoning_tokens_in_usage(oci_params):
"""OCI returns completionTokensDetails.reasoningTokens on reasoning models;
LiteLLM should surface it on Usage.completion_tokens_details."""
import litellm
resp = litellm.completion(
model=f"oci/{_REASONING_MODEL}",
messages=[{"role": "user", "content": "What is 2+2? One word."}],
max_tokens=200,
reasoning_effort="low",
**oci_params,
)
assert resp.usage.completion_tokens_details is not None
assert resp.usage.completion_tokens_details.reasoning_tokens is not None
assert resp.usage.completion_tokens_details.reasoning_tokens > 0
# ---------------------------------------------------------------------------
# Embedding tests
# ---------------------------------------------------------------------------
class TestOCIEmbeddings:
def test_english_v3_basic(self, oci_params):
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=["Hello world"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 1
assert len(resp.data[0]["embedding"]) == 1024
assert resp.usage.prompt_tokens > 0
def test_english_v3_batch(self, oci_params):
import litellm
texts = [
"The quick brown fox",
"jumps over the lazy dog",
"Paris is the capital of France",
]
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=texts,
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 3
for i, item in enumerate(resp.data):
assert item["index"] == i
assert len(item["embedding"]) == 1024
def test_multilingual_v3(self, oci_params):
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-multilingual-v3.0",
input=["Bonjour le monde", "Hola mundo"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 2
assert len(resp.data[0]["embedding"]) == 1024
def test_search_query_input_type(self, oci_params):
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=["What is the capital of France?"],
input_type="SEARCH_QUERY",
**oci_params,
)
assert len(resp.data[0]["embedding"]) == 1024
def test_semantic_similarity(self, oci_params):
"""Semantically similar texts should have higher cosine similarity."""
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=[
"The cat sat on the mat",
"A feline rested on the rug",
"The stock market crashed today",
],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
def cosine(a, b):
dot = sum(x * y for x, y in zip(a, b))
mag_a = math.sqrt(sum(x**2 for x in a))
mag_b = math.sqrt(sum(x**2 for x in b))
return dot / (mag_a * mag_b)
cat1 = resp.data[0]["embedding"]
cat2 = resp.data[1]["embedding"]
stock = resp.data[2]["embedding"]
sim_cats = cosine(cat1, cat2)
sim_diff = cosine(cat1, stock)
assert (
sim_cats > sim_diff
), f"Expected similar sentences to score higher ({sim_cats:.3f} vs {sim_diff:.3f})"
def test_embed_v4(self, oci_params):
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-v4.0",
input=["Hello world"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 1
assert len(resp.data[0]["embedding"]) == 1536
def test_usage_tokens(self, oci_params):
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=["short text", "another short text"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert resp.usage.prompt_tokens > 0
assert resp.usage.total_tokens == resp.usage.prompt_tokens
# ---------------------------------------------------------------------------
# Async embedding tests
# ---------------------------------------------------------------------------
class TestOCIAsyncEmbeddings:
@pytest.mark.asyncio
async def test_async_embedding_basic(self, oci_params):
import litellm
resp = await litellm.aembedding(
model="oci/cohere.embed-english-v3.0",
input=["Hello world"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 1
assert len(resp.data[0]["embedding"]) == 1024
assert resp.usage.prompt_tokens > 0
@pytest.mark.asyncio
async def test_async_embedding_batch(self, oci_params):
import litellm
texts = ["The quick brown fox", "jumps over the lazy dog"]
resp = await litellm.aembedding(
model="oci/cohere.embed-english-v3.0",
input=texts,
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data) == 2
assert all(len(item["embedding"]) == 1024 for item in resp.data)
@pytest.mark.asyncio
async def test_async_embedding_multilingual(self, oci_params):
import litellm
resp = await litellm.aembedding(
model="oci/cohere.embed-multilingual-v3.0",
input=["Bonjour le monde"],
input_type="SEARCH_DOCUMENT",
**oci_params,
)
assert len(resp.data[0]["embedding"]) == 1024
# ---------------------------------------------------------------------------
# Env-var credential path
# ---------------------------------------------------------------------------
class TestOCIEnvVarCredentials:
"""Verify that OCI_* env vars are picked up without explicit params."""
def test_completion_via_env_vars(self, monkeypatch):
"""Completion works when credentials are set through environment variables."""
pytest.importorskip("oci")
config = _load_oci_config()
key_path = os.path.expanduser(config["key_file"])
with open(key_path) as f:
key_pem = f.read()
monkeypatch.setenv("OCI_REGION", "us-chicago-1")
monkeypatch.setenv("OCI_USER", config["user"])
monkeypatch.setenv("OCI_FINGERPRINT", config["fingerprint"])
monkeypatch.setenv("OCI_TENANCY", config["tenancy"])
monkeypatch.setenv("OCI_KEY", key_pem)
monkeypatch.setenv("OCI_COMPARTMENT_ID", config["tenancy"])
import litellm
resp = litellm.completion(
model="oci/meta.llama-3.3-70b-instruct",
messages=[{"role": "user", "content": "Reply with only the word: pong"}],
max_tokens=10,
)
assert resp.choices[0].message.content is not None
def test_embedding_via_env_vars(self, monkeypatch):
pytest.importorskip("oci")
config = _load_oci_config()
key_path = os.path.expanduser(config["key_file"])
with open(key_path) as f:
key_pem = f.read()
monkeypatch.setenv("OCI_REGION", "us-chicago-1")
monkeypatch.setenv("OCI_USER", config["user"])
monkeypatch.setenv("OCI_FINGERPRINT", config["fingerprint"])
monkeypatch.setenv("OCI_TENANCY", config["tenancy"])
monkeypatch.setenv("OCI_KEY", key_pem)
monkeypatch.setenv("OCI_COMPARTMENT_ID", config["tenancy"])
import litellm
resp = litellm.embedding(
model="oci/cohere.embed-english-v3.0",
input=["hello"],
input_type="SEARCH_DOCUMENT",
)
assert len(resp.data[0]["embedding"]) == 1024
Reference in a new issue View git blame Copy permalink