mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
3f38730d2c
* refactor(typing): clear basedpyright Any errors (wave 1) * refactor(typing): clear basedpyright Any errors (wave 2) * refactor(typing): clear basedpyright Any errors (wave 3) * refactor(typing): clear basedpyright Any errors (wave 4) * refactor(typing): clear basedpyright Any errors (wave 5) * refactor(typing): clear basedpyright Any errors (wave 6) * fix(typing): keep the cached-plan fallback's direct query_first calls so span naming resolves * fix(logging): keep callbacks running when litellm_params is a params model should_run_callback and the sync success and failure handlers read litellm_params through any object with a get method again, as they did before this branch, so a logging hook that returns a GenericLiteLLMParams no longer stops sync logging for that request Adds behavior tests for the changed lines codecov listed as not run * fix: read the regional uplift for every data residency again and reuse one dict TypeAdapter The typed lookup only read the eu and us multipliers, so a deployment that set regional_processing_uplift_multiplier_fedramp stopped applying it. /model/update and the model write paths now share one module-level TypeAdapter instead of building one per call. * fix: bill regional uplift multipliers float() accepts, bytes included * fix: raise the in operator's own error from one shared json_contains, and name the bad uplift type
1015 lines
34 KiB
Python
1015 lines
34 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
from typing import Final
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
|
|
from litellm.integrations.galileo import GalileoObserve
|
|
from litellm.types.llms.openai import HttpxBinaryResponseContent, ResponsesAPIResponse
|
|
from litellm.types.rerank import RerankResponse
|
|
from litellm.types.utils import (
|
|
Choices,
|
|
EmbeddingResponse,
|
|
ImageObject,
|
|
ImageResponse,
|
|
Message,
|
|
ModelResponse,
|
|
TextCompletionResponse,
|
|
TranscriptionResponse,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def galileo_v2_env(monkeypatch):
|
|
monkeypatch.setenv("GALILEO_API_KEY", "test-api-key")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "86ff8ebe-a297-4134-b167-748bdd8d2c20")
|
|
monkeypatch.setenv("GALILEO_LOG_STREAM_ID", "76c4ea50-8aa3-4771-a0d7-8567b112210f")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://api.galileo.ai")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_v2_ingest_url_and_headers(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
logger.in_memory_records = [
|
|
{
|
|
"latency_ms": 100,
|
|
"status_code": 200,
|
|
"input_text": "hi",
|
|
"output_text": "hello",
|
|
"node_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 1,
|
|
"num_output_tokens": 2,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
}
|
|
]
|
|
|
|
url, payload = logger._get_ingest_request()
|
|
assert (
|
|
url
|
|
== "https://api.galileo.ai/ingest/traces/86ff8ebe-a297-4134-b167-748bdd8d2c20"
|
|
)
|
|
assert payload["log_stream_id"] == "76c4ea50-8aa3-4771-a0d7-8567b112210f"
|
|
assert payload["is_complete"] is True
|
|
assert payload["traces"][0]["type"] == "trace"
|
|
assert payload["traces"][0]["spans"][0]["type"] == "llm"
|
|
assert payload["traces"][0]["spans"][0]["output"]["content"] == "hello"
|
|
assert payload["traces"][0]["spans"][0]["metrics"]["num_total_tokens"] == 3
|
|
assert payload["traces"][0]["metrics"]["num_input_tokens"] == 1
|
|
assert payload["traces"][0]["metrics"]["num_output_tokens"] == 2
|
|
assert payload["traces"][0]["metrics"]["num_total_tokens"] == 3
|
|
assert payload["traces"][0]["spans"][0]["trace_id"] == payload["traces"][0]["id"]
|
|
|
|
assert await logger._ensure_headers() is True
|
|
assert logger.headers["Galileo-API-Key"] == "test-api-key"
|
|
|
|
|
|
def test_galileo_token_metrics_from_record_falls_back_to_sum():
|
|
metrics = GalileoObserve._token_metrics_from_record(
|
|
{"num_input_tokens": 5, "num_output_tokens": 7}
|
|
)
|
|
assert metrics == {
|
|
"num_input_tokens": 5,
|
|
"num_output_tokens": 7,
|
|
"num_total_tokens": 12,
|
|
}
|
|
|
|
|
|
def test_galileo_token_metrics_from_record_sums_zero_total():
|
|
metrics = GalileoObserve._token_metrics_from_record(
|
|
{"num_input_tokens": 5, "num_output_tokens": 7, "num_total_tokens": 0}
|
|
)
|
|
assert metrics == {
|
|
"num_input_tokens": 5,
|
|
"num_output_tokens": 7,
|
|
"num_total_tokens": 12,
|
|
}
|
|
|
|
|
|
def test_galileo_token_metrics_from_record_includes_cost():
|
|
metrics = GalileoObserve._token_metrics_from_record(
|
|
{
|
|
"num_input_tokens": 1,
|
|
"num_output_tokens": 2,
|
|
"num_total_tokens": 3,
|
|
"cost": 0.000855,
|
|
}
|
|
)
|
|
assert metrics["cost"] == 0.000855
|
|
|
|
|
|
def test_galileo_input_text_from_messages():
|
|
assert GalileoObserve._input_text_from_messages("hello") == "hello"
|
|
assert (
|
|
GalileoObserve._input_text_from_messages(
|
|
[{"role": "user", "content": "test responses api 1"}]
|
|
)
|
|
== "test responses api 1"
|
|
)
|
|
|
|
|
|
def test_galileo_get_output_str_responses_api(galileo_v2_env):
|
|
|
|
logger = GalileoObserve()
|
|
resp_dict = {
|
|
"id": "resp_123",
|
|
"created_at": 1,
|
|
"output": [
|
|
{
|
|
"id": "msg_1",
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"status": "completed",
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": "Hi! How can I help?",
|
|
"annotations": [],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
response = ResponsesAPIResponse(**resp_dict)
|
|
result = logger.get_output_str_from_response(response, {"call_type": "aresponses"})
|
|
assert result is not None
|
|
assert '"Hi! How can I help?"' in result
|
|
assert '"type": "message"' in result
|
|
|
|
|
|
def test_galileo_v2_span_preserves_message_roles(galileo_v2_env):
|
|
record = {
|
|
"latency_ms": 1,
|
|
"status_code": 200,
|
|
"input_text": "fallback",
|
|
"output_text": "ok",
|
|
"node_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 0,
|
|
"num_output_tokens": 0,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
"messages": [
|
|
{"role": "system", "content": "be helpful"},
|
|
{"role": "user", "content": "hello"},
|
|
],
|
|
}
|
|
span = GalileoObserve._record_to_v2_span(
|
|
record, trace_id="trace-id", span_id="span-id"
|
|
)
|
|
assert span["input"] == [
|
|
{"role": "system", "content": "be helpful"},
|
|
{"role": "user", "content": "hello"},
|
|
]
|
|
|
|
|
|
def test_galileo_v2_span_unwraps_prompt_messages(galileo_v2_env):
|
|
record = {
|
|
"latency_ms": 1,
|
|
"status_code": 200,
|
|
"input_text": "fallback",
|
|
"output_text": "ok",
|
|
"node_type": "pass_through_endpoint",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 0,
|
|
"num_output_tokens": 0,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
"messages": {
|
|
"messages": [
|
|
{"role": "system", "content": "be helpful"},
|
|
{"role": "user", "content": "hello"},
|
|
]
|
|
},
|
|
}
|
|
span = GalileoObserve._record_to_v2_span(
|
|
record, trace_id="trace-id", span_id="span-id"
|
|
)
|
|
assert span["input"] == [
|
|
{"role": "system", "content": "be helpful"},
|
|
{"role": "user", "content": "hello"},
|
|
]
|
|
|
|
|
|
def test_galileo_output_text_from_model_response(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
response = ModelResponse(
|
|
choices=[
|
|
Choices(
|
|
message=Message(
|
|
content="assistant reply",
|
|
role="assistant",
|
|
annotations=[],
|
|
)
|
|
)
|
|
]
|
|
)
|
|
|
|
output = logger.get_output_str_from_response(response, {"call_type": "acompletion"})
|
|
assert output is not None
|
|
assert '"assistant reply"' in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_flush_swallows_http_errors(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
logger.in_memory_records = [
|
|
{
|
|
"latency_ms": 1,
|
|
"status_code": 200,
|
|
"input_text": "a",
|
|
"output_text": "b",
|
|
"node_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 0,
|
|
"num_output_tokens": 0,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
}
|
|
]
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.side_effect = Exception("404 Not Found")
|
|
await logger.flush_in_memory_records()
|
|
|
|
assert len(logger.in_memory_records) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_flush_clears_records_on_201(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
logger.in_memory_records = [
|
|
{
|
|
"latency_ms": 1,
|
|
"status_code": 200,
|
|
"input_text": "a",
|
|
"output_text": "b",
|
|
"node_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 0,
|
|
"num_output_tokens": 0,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
}
|
|
]
|
|
|
|
mock_response = AsyncMock()
|
|
mock_response.is_success = True
|
|
mock_response.status_code = 201
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.return_value = mock_response
|
|
await logger.flush_in_memory_records()
|
|
|
|
assert logger.in_memory_records == []
|
|
|
|
|
|
def test_galileo_normalize_base_url_none(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_BASE_URL", raising=False)
|
|
monkeypatch.delenv("GALILEO_PROJECT_ID", raising=False)
|
|
logger = GalileoObserve()
|
|
assert logger.base_url is None
|
|
assert logger._normalize_base_url(None) is None
|
|
assert logger._normalize_base_url("https://x.example/") == "https://x.example"
|
|
|
|
|
|
def test_galileo_is_configured_branches(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_BASE_URL", raising=False)
|
|
monkeypatch.delenv("GALILEO_PROJECT_ID", raising=False)
|
|
monkeypatch.delenv("GALILEO_USERNAME", raising=False)
|
|
monkeypatch.delenv("GALILEO_PASSWORD", raising=False)
|
|
|
|
no_env = GalileoObserve()
|
|
assert no_env._is_configured() is False
|
|
|
|
monkeypatch.setenv("GALILEO_API_KEY", "k")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
v2 = GalileoObserve()
|
|
assert v2._is_configured() is True
|
|
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
legacy = GalileoObserve()
|
|
assert legacy._is_configured() is True
|
|
|
|
monkeypatch.delenv("GALILEO_PASSWORD", raising=False)
|
|
no_pw = GalileoObserve()
|
|
assert no_pw._is_configured() is False
|
|
|
|
|
|
def test_galileo_input_messages_fallbacks():
|
|
assert GalileoObserve._galileo_input_messages(None, "hi") == [
|
|
{"role": "user", "content": "hi"}
|
|
]
|
|
assert GalileoObserve._galileo_input_messages(
|
|
["not-a-dict", {"content": "no role"}], "fallback"
|
|
) == [{"role": "user", "content": "fallback"}]
|
|
|
|
|
|
def test_galileo_format_created_at_converts_local_naive_to_utc():
|
|
from datetime import timedelta
|
|
|
|
ist = timezone(timedelta(hours=5, minutes=30))
|
|
|
|
with patch.object(GalileoObserve, "_local_timezone", return_value=ist):
|
|
local_naive = datetime(2026, 6, 4, 9, 44, 49)
|
|
assert GalileoObserve._format_created_at(local_naive) == "2026-06-04T04:14:49Z"
|
|
|
|
aware_utc = datetime(2026, 6, 4, 4, 14, 49, tzinfo=timezone.utc)
|
|
assert GalileoObserve._format_created_at(aware_utc) == "2026-06-04T04:14:49Z"
|
|
|
|
|
|
def test_galileo_record_to_v2_span_with_tags_and_offset():
|
|
span = GalileoObserve._record_to_v2_span(
|
|
{
|
|
"latency_ms": 5,
|
|
"status_code": 200,
|
|
"input_text": "in",
|
|
"output_text": "out",
|
|
"node_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"num_input_tokens": 1,
|
|
"num_output_tokens": 2,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
"tags": ["t1"],
|
|
},
|
|
trace_id="trace-id",
|
|
span_id="span-id",
|
|
)
|
|
assert span["tags"] == ["t1"]
|
|
assert span["created_at"].endswith("Z")
|
|
|
|
offset = GalileoObserve._record_to_v2_span(
|
|
{"created_at": "2026-05-25T12:00:00-05:00"},
|
|
trace_id="trace-id",
|
|
span_id="span-id",
|
|
)
|
|
assert offset["created_at"] == "2026-05-25T12:00:00-05:00"
|
|
|
|
|
|
def test_galileo_get_output_str_variants(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
assert logger.get_output_str_from_response(None, {}) == ""
|
|
assert (
|
|
logger.get_output_str_from_response(
|
|
EmbeddingResponse(), {"call_type": "embedding"}
|
|
)
|
|
== "embedding-output"
|
|
)
|
|
assert (
|
|
logger.get_output_str_from_response(
|
|
EmbeddingResponse(), {"call_type": "aembedding"}
|
|
)
|
|
== "embedding-output"
|
|
)
|
|
|
|
text_resp = TextCompletionResponse()
|
|
text_resp.choices = [MagicMock(text="text-completion-output")]
|
|
assert (
|
|
logger.get_output_str_from_response(text_resp, {"call_type": "text_completion"})
|
|
== "text-completion-output"
|
|
)
|
|
|
|
image_resp = ImageResponse(data=[ImageObject(url="https://x/y.png")])
|
|
assert "y.png" in logger.get_output_str_from_response(image_resp, {})
|
|
|
|
speech_resp = HttpxBinaryResponseContent(response=MagicMock())
|
|
assert (
|
|
logger.get_output_str_from_response(speech_resp, {"call_type": "aspeech"})
|
|
== "speech-output"
|
|
)
|
|
|
|
transcription_resp = TranscriptionResponse(text="hello world")
|
|
assert (
|
|
logger.get_output_str_from_response(
|
|
transcription_resp, {"call_type": "atranscription"}
|
|
)
|
|
== "hello world"
|
|
)
|
|
|
|
realtime_output = [{"type": "response", "text": "hi"}]
|
|
assert (
|
|
logger.get_output_str_from_response(
|
|
realtime_output,
|
|
{"call_type": "_arealtime", "input": {"session": "abc"}},
|
|
)
|
|
== '[{"type": "response", "text": "hi"}]'
|
|
)
|
|
|
|
pass_through_output = {"response": "passthrough-body", "status": 200}
|
|
assert (
|
|
logger.get_output_str_from_response(
|
|
pass_through_output, {"call_type": "pass_through_endpoint"}
|
|
)
|
|
== "passthrough-body"
|
|
)
|
|
|
|
model_resp = ModelResponse(
|
|
choices=[Choices(message=Message(content="chat reply", role="assistant"))]
|
|
)
|
|
assert '"chat reply"' in logger.get_output_str_from_response(
|
|
model_resp,
|
|
{"call_type": "acompletion", "messages": [{"role": "user", "content": "hi"}]},
|
|
)
|
|
|
|
assert logger.get_output_str_from_response("not-a-supported-type", {}) == ""
|
|
|
|
|
|
def test_galileo_get_input_output_error_status_message(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
input_text, output_text, _ = logger._get_galileo_input_output_content(
|
|
kwargs={"messages": [{"role": "user", "content": "fail me"}]},
|
|
response_obj=None,
|
|
level="ERROR",
|
|
status_message="provider timeout",
|
|
)
|
|
assert input_text == "fail me"
|
|
assert output_text == "provider timeout"
|
|
|
|
|
|
def test_galileo_get_output_str_rerank_response(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
rerank_response = RerankResponse(
|
|
results=[
|
|
{"index": 2, "relevance_score": 0.98},
|
|
{"index": 0, "relevance_score": 0.12},
|
|
]
|
|
)
|
|
output = logger.get_output_str_from_response(
|
|
rerank_response, {"call_type": "arerank"}
|
|
)
|
|
assert output is not None
|
|
assert '"index": 2' in output
|
|
assert '"relevance_score": 0.98' in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_log_success_embedding(galileo_v2_env):
|
|
import datetime
|
|
|
|
logger = GalileoObserve()
|
|
embedding_response = EmbeddingResponse(
|
|
data=[{"object": "embedding", "embedding": [0.1, 0.2, 0.3], "index": 0}]
|
|
)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.is_success = True
|
|
mock_response.status_code = 201
|
|
|
|
with patch.object(logger.async_httpx_handler, "post", return_value=mock_response):
|
|
await logger.async_log_success_event(
|
|
kwargs={
|
|
"call_type": "aembedding",
|
|
"model": "text-embedding-3-small",
|
|
"input": "hello world",
|
|
"standard_logging_object": {
|
|
"call_type": "aembedding",
|
|
"model": "text-embedding-3-small",
|
|
"prompt_tokens": 2,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 2,
|
|
"response_cost": 0.0,
|
|
"startTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
"endTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
},
|
|
},
|
|
response_obj=embedding_response,
|
|
start_time=datetime.datetime(2026, 5, 25, 12, 0, 0),
|
|
end_time=datetime.datetime(2026, 5, 25, 12, 0, 1),
|
|
)
|
|
|
|
assert logger.in_memory_records == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_log_success_rerank(galileo_v2_env):
|
|
import datetime
|
|
|
|
logger = GalileoObserve()
|
|
rerank_response = RerankResponse(results=[{"index": 1, "relevance_score": 0.95}])
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.is_success = True
|
|
mock_response.status_code = 201
|
|
|
|
with patch.object(logger.async_httpx_handler, "post", return_value=mock_response):
|
|
await logger.async_log_success_event(
|
|
kwargs={
|
|
"call_type": "arerank",
|
|
"model": "cohere/rerank-english-v3.0",
|
|
"query": "What is the capital of the United States?",
|
|
"documents": ["doc-a", "doc-b"],
|
|
"standard_logging_object": {
|
|
"call_type": "arerank",
|
|
"model": "cohere/rerank-english-v3.0",
|
|
"messages": "What is the capital of the United States?",
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 0,
|
|
"response_cost": 0.0,
|
|
"startTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
"endTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
},
|
|
},
|
|
response_obj=rerank_response,
|
|
start_time=datetime.datetime(2026, 5, 25, 12, 0, 0),
|
|
end_time=datetime.datetime(2026, 5, 25, 12, 0, 1),
|
|
)
|
|
|
|
assert logger.in_memory_records == []
|
|
|
|
|
|
def test_galileo_get_ingest_request_unconfigured(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_BASE_URL", raising=False)
|
|
monkeypatch.delenv("GALILEO_PROJECT_ID", raising=False)
|
|
logger = GalileoObserve()
|
|
assert logger._get_ingest_request() is None
|
|
|
|
|
|
def test_galileo_get_ingest_request_legacy(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example/")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "proj")
|
|
monkeypatch.setenv("GALILEO_LOG_STREAM_ID", "stream-id")
|
|
logger = GalileoObserve()
|
|
logger.in_memory_records = [
|
|
{
|
|
"latency_ms": 1,
|
|
"status_code": 200,
|
|
"input_text": "hi",
|
|
"output_text": "ok",
|
|
"node_type": "acompletion",
|
|
"model": "gpt",
|
|
"num_input_tokens": 1,
|
|
"num_output_tokens": 1,
|
|
"num_total_tokens": 2,
|
|
"created_at": "2026-05-25T12:00:00",
|
|
}
|
|
]
|
|
url, payload = logger._get_ingest_request()
|
|
assert url == "https://galileo.example/v2/projects/proj/traces"
|
|
assert "traces" in payload
|
|
assert payload["log_stream_id"] == "stream-id"
|
|
assert payload["traces"][0]["input"] == "hi"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_success(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
current_user_resp = MagicMock()
|
|
current_user_resp.status_code = 200
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "get", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = current_user_resp
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "healthy"
|
|
mock_get.assert_awaited_once_with(
|
|
url="https://api.galileo.ai/current_user",
|
|
headers={
|
|
"accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Galileo-API-Key": "test-api-key",
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_api_error(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
current_user_resp = MagicMock()
|
|
current_user_resp.status_code = 401
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "get", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = current_user_resp
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert "HTTP 401" in result["error_message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_missing_project_id(monkeypatch):
|
|
monkeypatch.setenv("GALILEO_API_KEY", "test-api-key")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://api.galileo.ai")
|
|
monkeypatch.delenv("GALILEO_PROJECT_ID", raising=False)
|
|
logger = GalileoObserve()
|
|
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert "GALILEO_PROJECT_ID" in result["error_message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_missing_base_url(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_BASE_URL", raising=False)
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
logger = GalileoObserve()
|
|
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert "GALILEO_BASE_URL" in result["error_message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_missing_credentials(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_USERNAME", raising=False)
|
|
monkeypatch.delenv("GALILEO_PASSWORD", raising=False)
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
logger = GalileoObserve()
|
|
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert "GALILEO_USERNAME" in result["error_message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_auth_failed(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
logger = GalileoObserve()
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.side_effect = Exception("login failed")
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert result["error_message"] == "Galileo authentication failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_health_check_request_exception(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "get", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.side_effect = Exception("connection refused")
|
|
result = await logger.async_health_check()
|
|
|
|
assert result["status"] == "unhealthy"
|
|
assert "connection refused" in result["error_message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_log_success_empty_model_response(galileo_v2_env):
|
|
import datetime
|
|
|
|
logger = GalileoObserve()
|
|
logger.batch_size = 2
|
|
empty_response = ModelResponse(choices=[])
|
|
|
|
await logger.async_log_success_event(
|
|
kwargs={
|
|
"call_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"messages": [{"role": "user", "content": "hi"}],
|
|
"standard_logging_object": {
|
|
"call_type": "acompletion",
|
|
"model": "gpt-5.2",
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 0,
|
|
"total_tokens": 1,
|
|
"response_cost": 0.0,
|
|
"startTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
"endTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
},
|
|
},
|
|
response_obj=empty_response,
|
|
start_time=datetime.datetime(2026, 5, 25, 12, 0, 0),
|
|
end_time=datetime.datetime(2026, 5, 25, 12, 0, 1),
|
|
)
|
|
|
|
assert len(logger.in_memory_records) == 1
|
|
assert logger.in_memory_records[0]["output_text"] == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_ensure_headers_v2_missing_key(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://x")
|
|
logger = GalileoObserve()
|
|
logger.use_v2_api = True
|
|
logger.api_key = None
|
|
assert await logger._ensure_headers() is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_ensure_headers_cached(galileo_v2_env):
|
|
logger = GalileoObserve()
|
|
logger.headers = {"Galileo-API-Key": "already-set"}
|
|
assert await logger._ensure_headers() is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_ensure_headers_legacy_login(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
logger = GalileoObserve()
|
|
|
|
login_resp = MagicMock()
|
|
login_resp.raise_for_status = MagicMock()
|
|
login_resp.json = MagicMock(return_value={"access_token": "tok"})
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.return_value = login_resp
|
|
assert await logger._ensure_headers() is True
|
|
|
|
assert logger.headers["Authorization"] == "Bearer tok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_ensure_headers_legacy_login_failure(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
logger = GalileoObserve()
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.side_effect = Exception("boom")
|
|
assert await logger._ensure_headers() is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_flush_noop_when_unconfigured(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.delenv("GALILEO_BASE_URL", raising=False)
|
|
monkeypatch.delenv("GALILEO_PROJECT_ID", raising=False)
|
|
logger = GalileoObserve()
|
|
logger.in_memory_records = [{"foo": "bar"}]
|
|
await logger.flush_in_memory_records()
|
|
assert logger.in_memory_records == [{"foo": "bar"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_flush_resets_headers_on_401(monkeypatch):
|
|
monkeypatch.delenv("GALILEO_API_KEY", raising=False)
|
|
monkeypatch.setenv("GALILEO_USERNAME", "u")
|
|
monkeypatch.setenv("GALILEO_PASSWORD", "pw")
|
|
monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example")
|
|
monkeypatch.setenv("GALILEO_PROJECT_ID", "p")
|
|
logger = GalileoObserve()
|
|
logger.headers = {"Authorization": "Bearer stale"}
|
|
logger.in_memory_records = [{"records": "x"}]
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.is_success = False
|
|
mock_response.status_code = 401
|
|
mock_response.text = "unauthorized"
|
|
|
|
with patch.object(
|
|
logger.async_httpx_handler, "post", new_callable=AsyncMock
|
|
) as mock_post:
|
|
mock_post.return_value = mock_response
|
|
await logger.flush_in_memory_records()
|
|
|
|
assert logger.headers is None
|
|
assert logger.in_memory_records == [{"records": "x"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_log_success_preserves_passthrough_messages(
|
|
galileo_v2_env,
|
|
):
|
|
import datetime
|
|
|
|
logger = GalileoObserve()
|
|
logger.batch_size = 2
|
|
messages = [
|
|
{"role": "system", "content": "be helpful"},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
|
|
await logger.async_log_success_event(
|
|
kwargs={
|
|
"call_type": "pass_through_endpoint",
|
|
"model": "gpt",
|
|
"messages": messages,
|
|
"standard_logging_object": {
|
|
"call_type": "pass_through_endpoint",
|
|
"model": "gpt",
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 2,
|
|
"total_tokens": 0,
|
|
"response_cost": 0.001,
|
|
"startTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
"endTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
},
|
|
},
|
|
response_obj={"response": "ok"},
|
|
start_time=datetime.datetime(2026, 5, 25, 12, 0, 0),
|
|
end_time=datetime.datetime(2026, 5, 25, 12, 0, 1),
|
|
)
|
|
|
|
assert logger.in_memory_records[0]["messages"] == messages
|
|
assert logger.in_memory_records[0]["num_total_tokens"] == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_galileo_async_log_success_appends_and_flushes(galileo_v2_env):
|
|
import datetime
|
|
|
|
logger = GalileoObserve()
|
|
response = ModelResponse(
|
|
choices=[
|
|
Choices(message=Message(content="reply", role="assistant", annotations=[]))
|
|
],
|
|
usage={"prompt_tokens": 1, "completion_tokens": 2},
|
|
)
|
|
|
|
flushed_url: dict = {}
|
|
mock_response = MagicMock()
|
|
mock_response.is_success = True
|
|
mock_response.status_code = 200
|
|
|
|
async def fake_post(**kwargs):
|
|
flushed_url["url"] = kwargs.get("url")
|
|
return mock_response
|
|
|
|
with patch.object(logger.async_httpx_handler, "post", side_effect=fake_post):
|
|
await logger.async_log_success_event(
|
|
kwargs={
|
|
"call_type": "acompletion",
|
|
"model": "gpt",
|
|
"messages": [{"role": "user", "content": "hi"}],
|
|
"standard_logging_object": {
|
|
"call_type": "acompletion",
|
|
"model": "gpt",
|
|
"messages": [{"role": "user", "content": "hi"}],
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 2,
|
|
"total_tokens": 3,
|
|
"response_cost": 0.001,
|
|
"startTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
"endTime": datetime.datetime(
|
|
2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc
|
|
).timestamp(),
|
|
},
|
|
},
|
|
response_obj=response,
|
|
start_time=datetime.datetime(2026, 5, 25, 12, 0, 0),
|
|
end_time=datetime.datetime(2026, 5, 25, 12, 0, 1),
|
|
)
|
|
|
|
assert "/ingest/traces/" in flushed_url["url"]
|
|
assert logger.in_memory_records == []
|
|
|
|
|
|
class RealtimeEvent(BaseModel):
|
|
type: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ThirdPartyEvent:
|
|
model_dump: object
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("event", "expected_output"),
|
|
[
|
|
(RealtimeEvent(type="response.done"), '[{"type": "response.done"}]'),
|
|
(ThirdPartyEvent(model_dump=lambda: {"type": "custom"}), '[{"type": "custom"}]'),
|
|
(ThirdPartyEvent(model_dump=lambda: [Decimal("2.5")]), '[["2.5"]]'),
|
|
(Decimal("1.5"), '["1.5"]'),
|
|
],
|
|
)
|
|
def test_galileo_realtime_output_serializes_events_that_json_cannot_encode(
|
|
galileo_v2_env: None, event: object, expected_output: str
|
|
) -> None:
|
|
assert GalileoObserve().get_output_str_from_response([event], {"call_type": "_arealtime"}) == expected_output
|
|
|
|
|
|
@pytest.mark.parametrize("model_dump", [None, "not callable"])
|
|
def test_galileo_realtime_output_with_a_model_dump_that_cannot_be_called_raises_a_validation_error(
|
|
galileo_v2_env: None, model_dump: object
|
|
) -> None:
|
|
with pytest.raises(ValidationError) as raised:
|
|
GalileoObserve().get_output_str_from_response(
|
|
[ThirdPartyEvent(model_dump=model_dump)], {"call_type": "_arealtime"}
|
|
)
|
|
|
|
assert "input_value" not in str(raised.value)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ThirdPartyMessage:
|
|
json: object
|
|
|
|
|
|
def _chat_response_carrying(message: object) -> ModelResponse:
|
|
response: Final = ModelResponse(choices=[Choices(message=Message(content="replaced", role="assistant"))])
|
|
response.choices[0].message = message
|
|
return response
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "expected_output"),
|
|
[
|
|
(
|
|
ThirdPartyMessage(json=lambda: '{"role":"assistant","content":"hi"}'),
|
|
'{"role": "assistant", "content": "hi"}',
|
|
),
|
|
(
|
|
ThirdPartyMessage(json=lambda: {"role": "assistant", "content": "hi"}),
|
|
'{"role": "assistant", "content": "hi"}',
|
|
),
|
|
(ThirdPartyMessage(json=lambda: '"just text"'), "just text"),
|
|
(ThirdPartyMessage(json=lambda: None), ""),
|
|
({"role": "assistant", "content": "hi"}, '{"role": "assistant", "content": "hi"}'),
|
|
("plain reply", "plain reply"),
|
|
(None, ""),
|
|
],
|
|
)
|
|
def test_galileo_output_str_of_a_chat_response_whose_message_is_not_a_litellm_message(
|
|
galileo_v2_env: None, message: object, expected_output: str
|
|
) -> None:
|
|
output: Final = GalileoObserve().get_output_str_from_response(
|
|
_chat_response_carrying(message), {"call_type": "acompletion"}
|
|
)
|
|
|
|
assert output == expected_output
|
|
|
|
|
|
GALILEO_MESSAGES: Final = [{"role": "user", "content": "weather in SF?"}]
|
|
GALILEO_TOOLS: Final = [{"type": "function", "function": {"name": "get_weather", "parameters": {}}}]
|
|
GALILEO_FUNCTIONS: Final = [{"name": "get_time", "parameters": {}}]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("optional_params", "expected_input"),
|
|
[
|
|
(
|
|
{"functions": GALILEO_FUNCTIONS, "tools": GALILEO_TOOLS, "temperature": 0.1},
|
|
{"messages": GALILEO_MESSAGES, "functions": GALILEO_FUNCTIONS, "tools": GALILEO_TOOLS},
|
|
),
|
|
(None, {"messages": GALILEO_MESSAGES}),
|
|
],
|
|
)
|
|
def test_galileo_error_span_input_carries_the_request_functions_and_tools(
|
|
optional_params: dict[str, object] | None, expected_input: dict[str, object]
|
|
) -> None:
|
|
_, output_text, logged_input = GalileoObserve()._get_galileo_input_output_content(
|
|
{"messages": GALILEO_MESSAGES, "optional_params": optional_params},
|
|
None,
|
|
level="ERROR",
|
|
status_message="upstream 500",
|
|
)
|
|
|
|
assert output_text == "upstream 500"
|
|
assert logged_input == expected_input
|