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.

tests/local_testing/test_opik.py

@@ -16,6 +16,8 @@ verbose_logger.setLevel(logging.DEBUG)
 litellm.set_verbose = True
 import time
 
+INTERVAL_TOO_LONG_TO_FIRE_DURING_THIS_TEST = 3600
+
 
 @pytest.mark.asyncio
 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
     - 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():