Fix: Resolve flakiness in three integration tests (#17594)

Fixed three flaky tests that were intermittently failing in CI:

1. test_no_duplicate_spend_logs (test_litellm/responses/test_no_duplicate_spend_logs.py)
   Problem: Used await asyncio.sleep(1) to wait for async logging completion,
            which created race conditions. The async logging worker queues tasks
            in the background, and sleep() doesn't guarantee completion.

   Fix: Replaced sleep() with GLOBAL_LOGGING_WORKER.flush() which properly waits
        for the logging queue to empty, ensuring all async logging tasks complete
        before assertions run.

2. test_log_langfuse_v2_handles_null_usage_values (test_litellm/integrations/test_langfuse.py)
   Problem: Used datetime.datetime.now() twice for start_time and end_time, which
            could cause timing inconsistencies between test runs, especially in
            CI environments with variable execution speeds.

   Fix: Use fixed timestamps instead of datetime.now() to ensure consistent timing
        across all test runs, eliminating timing-related flakiness.

3. test_watsonx_gpt_oss_prompt_transformation (test_litellm/llms/watsonx/test_watsonx.py)
   Problem: Directly accessed mock_post.call_args without checking if it exists,
            which could be None if the mock wasn't called or if an exception
            occurred before the POST request. The test catches exceptions and
            continues, making this a potential failure point.

   Fix: Added proper assertions and use call_args_list[0] for safer access:
        - Assert that call_args_list has at least one call
        - Assert that call_args is not None
        - Assert that 'data' key exists in kwargs
        This ensures the test fails with clear error messages rather than
        intermittent AttributeError exceptions.

All fixes maintain the original test intent while making them deterministic
and reliable in CI environments.

tests/test_litellm/integrations/test_langfuse.py

@@ -279,14 +279,16 @@ class TestLangfuseUsageDetails(unittest.TestCase):
                 "response_cost": 0.0,
             }
 
+            # Use fixed timestamps to avoid timing-related flakiness
+            fixed_time = datetime.datetime(2024, 1, 1, 12, 0, 0)
             # Call the method under test
             self.logger._log_langfuse_v2(
                 user_id="test-user",
                 metadata={},
                 litellm_params=kwargs["litellm_params"],
                 output={"role": "assistant", "content": "Response"},
-                start_time=datetime.datetime.now(),
-                end_time=datetime.datetime.now(),
+                start_time=fixed_time,
+                end_time=fixed_time + datetime.timedelta(seconds=1),
                 kwargs=kwargs,
                 optional_params=kwargs["optional_params"],
                 input={"messages": kwargs["messages"]},

tests/test_litellm/llms/watsonx/test_watsonx.py

@@ -279,7 +279,11 @@ async def test_watsonx_gpt_oss_prompt_transformation(monkeypatch):
     ), f"POST should have been called at least once, got {mock_post.call_count}"
 
     # Get the request body from the first call
-    call_args = mock_post.call_args
+    # Use call_args_list to be more robust - get the first call's arguments
+    assert len(mock_post.call_args_list) > 0, "mock_post should have at least one call"
+    call_args = mock_post.call_args_list[0]
+    assert call_args is not None, "call_args should not be None"
+    assert "data" in call_args.kwargs, "call_args.kwargs should contain 'data'"
     json_data = json.loads(call_args.kwargs["data"])
 
     print(f"\n{'='*80}")

tests/test_litellm/responses/test_no_duplicate_spend_logs.py

@@ -83,9 +83,10 @@ async def test_no_duplicate_spend_logs():
             mock_response="Hello! I'm doing well."  # Use mock to avoid real API call
         )
 
-        # Give async logging time to complete
-        import asyncio
-        await asyncio.sleep(1)
+        # Wait for async logging to complete using the logging worker's flush method
+        # This is more reliable than sleep() which can cause race conditions
+        from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+        await GLOBAL_LOGGING_WORKER.flush()
 
         # Verify that log_success_event was called exactly once
         assert spend_logger.log_count == 1, (