test writing logs to GCS bucket

litellm/tests/adroit-crow-413218-bc47f303efc9.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "adroit-crow-413218",
+  "private_key_id": "",
+  "private_key": "",
+  "client_email": "test-adroit-crow@adroit-crow-413218.iam.gserviceaccount.com",
+  "client_id": "104886546564708740969",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/test-adroit-crow%40adroit-crow-413218.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

litellm/tests/test_gcs_bucket.py

@@ -5,7 +5,9 @@ import sys
 sys.path.insert(0, os.path.abspath("../.."))
 
 import asyncio
+import json
 import logging
+import tempfile
 import uuid
 
 import pytest
@@ -18,12 +20,53 @@ from litellm.integrations.gcs_bucket import GCSBucketLogger
 verbose_logger.setLevel(logging.DEBUG)
 
 
-gcs_logger = GCSBucketLogger()
-print("GCSBucketLogger", gcs_logger)
+def load_vertex_ai_credentials():
+    # Define the path to the vertex_key.json file
+    print("loading vertex ai credentials")
+    filepath = os.path.dirname(os.path.abspath(__file__))
+    vertex_key_path = filepath + "/adroit-crow-413218-bc47f303efc9.json"
+
+    # Read the existing content of the file or create an empty dictionary
+    try:
+        with open(vertex_key_path, "r") as file:
+            # Read the file content
+            print("Read vertexai file path")
+            content = file.read()
+
+            # If the file is empty or not valid JSON, create an empty dictionary
+            if not content or not content.strip():
+                service_account_key_data = {}
+            else:
+                # Attempt to load the existing JSON content
+                file.seek(0)
+                service_account_key_data = json.load(file)
+    except FileNotFoundError:
+        # If the file doesn't exist, create an empty dictionary
+        service_account_key_data = {}
+
+    # Update the service_account_key_data with environment variables
+    private_key_id = os.environ.get("GCS_PRIVATE_KEY_ID", "")
+    private_key = os.environ.get("GCS_PRIVATE_KEY", "")
+    private_key = private_key.replace("\\n", "\n")
+    service_account_key_data["private_key_id"] = private_key_id
+    service_account_key_data["private_key"] = private_key
+
+    # Create a temporary file
+    with tempfile.NamedTemporaryFile(mode="w+", delete=False) as temp_file:
+        # Write the updated content to the temporary files
+        json.dump(service_account_key_data, temp_file, indent=2)
+
+    # Export the temporary file as GOOGLE_APPLICATION_CREDENTIALS
+    os.environ["GCS_PATH_SERVICE_ACCOUNT"] = os.path.abspath(temp_file.name)
+    print("created gcs path service account=", os.environ["GCS_PATH_SERVICE_ACCOUNT"])
 
 
 @pytest.mark.asyncio
 async def test_basic_gcs_logger():
+    load_vertex_ai_credentials()
+    gcs_logger = GCSBucketLogger()
+    print("GCSBucketLogger", gcs_logger)
+
     litellm.callbacks = [gcs_logger]
     response = await litellm.acompletion(
         model="gpt-3.5-turbo",