mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
test(opik): stop the batching test racing its own 5-second flush timer
test_opik_logging_http_request asserted "nothing has been POSTed yet" roughly one second into a window governed by OpikLogger's 5-second periodic flush. On a loaded CI worker the five preceding acompletion calls eat that budget, the periodic flush fires, and the assertion flips. Reproduced with no product changes at all: letting 5.5 seconds pass before the assertion drains the queue and sets mock_post.called, which is exactly the failure CircleCI reports. The test now pins flush_interval past anything the test can reach, so the two batching assertions measure batching instead of wall clock, and drives the flush path explicitly at the end rather than sleeping the interval. That last phase used to be near-vacuous, since the size-triggered flush had already emptied the queue. Assertions now match only calls to Opik's own /traces/batch and /spans/batch. get_async_httpx_client caches one client per special provider, so the mock is process-wide and any other logging callback's POST would otherwise count. Dropped the teardown that closed that shared client, which broke every later test in the same worker that logs through it, and the try/except that turned assertion failures into a pytest.fail with no traceback. Mutation checked: flushing on every event and never flushing on size both fail the test.
This commit is contained in:
parent
a47c2f76ab
commit
21092d633b
1 changed files with 45 additions and 53 deletions
|
|
@ -16,6 +16,8 @@ verbose_logger.setLevel(logging.DEBUG)
|
||||||
litellm.set_verbose = True
|
litellm.set_verbose = True
|
||||||
import time
|
import time
|
||||||
|
|
||||||
|
INTERVAL_TOO_LONG_TO_FIRE_DURING_THIS_TEST = 3600
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_opik_logging_http_request():
|
async def test_opik_logging_http_request():
|
||||||
|
|
@ -23,70 +25,60 @@ async def test_opik_logging_http_request():
|
||||||
- Test that HTTP requests are made to Opik
|
- Test that HTTP requests are made to Opik
|
||||||
- Traces and spans are batched correctly
|
- Traces and spans are batched correctly
|
||||||
"""
|
"""
|
||||||
try:
|
from litellm.integrations.opik.opik import OpikLogger
|
||||||
from litellm.integrations.opik.opik import OpikLogger
|
|
||||||
|
|
||||||
os.environ["OPIK_URL_OVERRIDE"] = "https://fake.comet.com/opik/api"
|
os.environ["OPIK_URL_OVERRIDE"] = "https://fake.comet.com/opik/api"
|
||||||
os.environ["OPIK_API_KEY"] = "anything"
|
os.environ["OPIK_API_KEY"] = "anything"
|
||||||
os.environ["OPIK_WORKSPACE"] = "anything"
|
os.environ["OPIK_WORKSPACE"] = "anything"
|
||||||
|
|
||||||
# Initialize OpikLogger
|
test_opik_logger = OpikLogger()
|
||||||
test_opik_logger = OpikLogger()
|
test_opik_logger.flush_interval = INTERVAL_TOO_LONG_TO_FIRE_DURING_THIS_TEST
|
||||||
|
test_opik_logger.batch_size = 12
|
||||||
|
|
||||||
litellm.callbacks = [test_opik_logger]
|
litellm.callbacks = [test_opik_logger]
|
||||||
test_opik_logger.batch_size = 12
|
|
||||||
litellm.set_verbose = True
|
|
||||||
|
|
||||||
# Create a mock for the async_client's post method
|
mock_post = AsyncMock(return_value=Mock(status_code=202, text="Accepted"))
|
||||||
mock_post = AsyncMock()
|
test_opik_logger.async_httpx_client.post = mock_post
|
||||||
mock_post.return_value.status_code = 202
|
|
||||||
mock_post.return_value.text = "Accepted"
|
|
||||||
test_opik_logger.async_httpx_client.post = mock_post
|
|
||||||
|
|
||||||
# Make multiple calls to ensure we don't hit the batch size
|
def opik_batch_calls():
|
||||||
for _ in range(5):
|
return [
|
||||||
response = await litellm.acompletion(
|
call
|
||||||
model="gpt-3.5-turbo",
|
for call in mock_post.call_args_list
|
||||||
messages=[{"role": "user", "content": "Test message"}],
|
if "/traces/batch" in str(call) or "/spans/batch" in str(call)
|
||||||
max_tokens=10,
|
]
|
||||||
temperature=0.2,
|
|
||||||
mock_response="This is a mock response",
|
|
||||||
)
|
|
||||||
await asyncio.sleep(1)
|
|
||||||
|
|
||||||
# Check batching of events and that the queue contains 5 trace events and 5 span events
|
for _ in range(5):
|
||||||
assert (
|
await litellm.acompletion(
|
||||||
mock_post.called == False
|
model="gpt-3.5-turbo",
|
||||||
), "HTTP request was made but events should have been batched"
|
messages=[{"role": "user", "content": "Test message"}],
|
||||||
assert len(test_opik_logger.log_queue) == 10
|
max_tokens=10,
|
||||||
|
temperature=0.2,
|
||||||
|
mock_response="This is a mock response",
|
||||||
|
)
|
||||||
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
# Now make calls to exceed the batch size
|
assert opik_batch_calls() == [], "events below batch_size must stay queued"
|
||||||
for _ in range(3):
|
assert len(test_opik_logger.log_queue) == 10
|
||||||
response = await litellm.acompletion(
|
|
||||||
model="gpt-3.5-turbo",
|
|
||||||
messages=[{"role": "user", "content": "Test message"}],
|
|
||||||
max_tokens=10,
|
|
||||||
temperature=0.2,
|
|
||||||
mock_response="This is a mock response",
|
|
||||||
)
|
|
||||||
|
|
||||||
# Wait a short time for any asynchronous operations to complete
|
for _ in range(3):
|
||||||
await asyncio.sleep(1)
|
await litellm.acompletion(
|
||||||
|
model="gpt-3.5-turbo",
|
||||||
|
messages=[{"role": "user", "content": "Test message"}],
|
||||||
|
max_tokens=10,
|
||||||
|
temperature=0.2,
|
||||||
|
mock_response="This is a mock response",
|
||||||
|
)
|
||||||
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
# Check that the queue was flushed after exceeding batch size
|
assert opik_batch_calls(), "crossing batch_size must flush the queue"
|
||||||
assert len(test_opik_logger.log_queue) < test_opik_logger.batch_size
|
events_left_over_after_the_size_triggered_flush = len(test_opik_logger.log_queue)
|
||||||
|
assert 0 < events_left_over_after_the_size_triggered_flush < test_opik_logger.batch_size
|
||||||
|
|
||||||
# Check that the data has been sent when it goes above the flush interval
|
calls_before_periodic_flush = len(opik_batch_calls())
|
||||||
await asyncio.sleep(test_opik_logger.flush_interval)
|
await test_opik_logger.flush_queue()
|
||||||
assert len(test_opik_logger.log_queue) == 0
|
|
||||||
|
|
||||||
# Clean up
|
assert len(opik_batch_calls()) > calls_before_periodic_flush
|
||||||
for cb in litellm.callbacks:
|
assert len(test_opik_logger.log_queue) == 0
|
||||||
if isinstance(cb, OpikLogger):
|
|
||||||
await cb.async_httpx_client.client.aclose()
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
pytest.fail(f"Error occurred: {e}")
|
|
||||||
|
|
||||||
|
|
||||||
def test_sync_opik_logging_http_request():
|
def test_sync_opik_logging_http_request():
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue