test(galileo): cover async_health_check failure paths and empty model response

Add regression tests for the Galileo health check unhealthy branches
(missing project id, missing base url, missing credentials, auth
failure, and request exception) and for logging a model response with
no choices, which now queues an empty output instead of being skipped.
This commit is contained in:
mateo-berri 2026-06-08 20:42:52 +00:00 • committed by Claude
parent f7a04d4ee1
commit b2bff79109
No known key found for this signature in database

View file

@ -611,6 +611,119 @@ async def test_galileo_async_health_check_api_error(galileo_v2_env):
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)