test(ci): fix the e2e and integration reds left on rc/1.103.0 after #43382 and #43388 (#43391)

The batch cleanup leftover warning used a class defined in a test-directory module. The xdist controller cannot import it, so an uncaught leftover warning crashed the whole e2e run. It is now a plain UserWarning.

The integration harness backport missed the workers option and Gateway.request headers that the OTel conversation id tests use.

tests/e2e/batches/batch_cleanup.py

@@ -28,10 +28,6 @@ class BatchCleanupClient(Protocol):
     def cancel_batch(self, batch_id: str, *, key: str, provider: str | None = None) -> Result[BatchObject]: ...
 
 
-class BatchCleanupLeftover(UserWarning):
-    pass
-
-
 def cleanup_result[R: BaseModel](
     action: Callable[[], Result[R]], *, wait: Callable[[float], None] = sleep
 ) -> Result[R]:
@@ -68,7 +64,7 @@ def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider
     if isinstance(result, UnknownApiError) and result.status_code == 400 and FILE_IN_USE_REFUSAL in result.body:
         warnings.warn(
             f"Left file {file_id} in place: LiteLLM refused to delete it while a batch still references it",
-            BatchCleanupLeftover,
+            UserWarning,
             stacklevel=2,
         )
         return
@@ -140,7 +136,7 @@ def cleanup_batch(
             )
             warnings.warn(
                 f"Left batch {batch_id} cancelling after {BATCH_CANCEL_TIMEOUT_SECONDS}s for the provider to finish",
-                BatchCleanupLeftover,
+                UserWarning,
                 stacklevel=2,
             )
             return

tests/e2e/batches/test_batch_cleanup.py

@@ -7,7 +7,6 @@ import pytest
 from batch_cleanup import (
     BATCH_CANCEL_TIMEOUT_SECONDS,
     CLEANUP_DELAYS,
-    BatchCleanupLeftover,
     cleanup_batch,
     cleanup_file,
     cleanup_result,
@@ -141,7 +140,7 @@ class TestFileCleanup:
             calls=ExpectedCalls((f"delete None {MANAGED_FILE_ID}",)),
             files=(UnknownApiError(status_code=400, body=IN_USE_REFUSAL),),
         )
-        with pytest.warns(BatchCleanupLeftover, match=MANAGED_FILE_ID):
+        with pytest.warns(UserWarning, match=MANAGED_FILE_ID):
             cleanup_file(client, MANAGED_FILE_ID, key="test-key")
         client.calls.assert_done()
 
@@ -241,7 +240,7 @@ class TestBatchCancellation:
         key: Final = manager.key()
         manager.defer(lambda: cleanup_file(client, MANAGED_FILE_ID, key=key))
         manager.defer(lambda: cleanup_batch(client, MANAGED_BATCH_ID, key=key, clock=ticks))
-        with pytest.warns(BatchCleanupLeftover) as leftovers:
+        with pytest.warns(UserWarning, match="^Left ") as leftovers:
             manager.teardown()
         client.calls.assert_done()
         messages: Final = tuple(str(warning.message) for warning in leftovers)

tests/integration/_support/client.py

@@ -58,13 +58,18 @@ class Gateway:
         *,
         key: str | None = None,
         params: Mapping[str, str] | None = None,
+        headers: Mapping[str, str] | None = None,
     ) -> httpx.Response:
+        request_headers: Final = {
+            "Authorization": f"Bearer {self.key if key is None else key}",
+            **(headers or {}),
+        }
         return self.client.request(
             method,
             path,
             json=body,
             params=params,
-            headers={"Authorization": f"Bearer {self.key if key is None else key}"},
+            headers=request_headers,
         )
 
     def post(self, path: str, body: Mapping[str, JsonValue], *, key: str | None = None) -> dict[str, JsonValue]:

tests/integration/_support/process.py

@@ -61,9 +61,10 @@ def owned_proxy(
     *,
     config: Path | None = None,
     remove_environment: tuple[str, ...] = (),
+    workers: int = 1,
 ) -> Iterator[Gateway]:
     with owned_proxy_process(
-        gateway, directory, overrides, config=config, remove_environment=remove_environment
+        gateway, directory, overrides, config=config, remove_environment=remove_environment, workers=workers
     ) as owned:
         yield owned.gateway
 
@@ -76,6 +77,7 @@ def owned_proxy_process(
     *,
     config: Path | None = None,
     remove_environment: tuple[str, ...] = (),
+    workers: int = 1,
 ) -> Iterator[OwnedProxy]:
     with socket.socket() as reserve:
         reserve.bind(("127.0.0.1", 0))
@@ -104,7 +106,7 @@ def owned_proxy_process(
                 "--port",
                 str(port),
                 "--num_workers",
-                "1",
+                str(workers),
                 "--telemetry",
                 "False",
                 "--use_prisma_db_push",