test: run live module integration tests against both backends

google/genai/tests/conftest.py

@@ -116,6 +116,13 @@ def replays_prefix():
   return 'test'
 
 
+# Overridden by test packages whose Vertex model is not served at the configured
+# location. Returning None leaves GOOGLE_CLOUD_LOCATION as-is.
+@pytest.fixture
+def location_override():
+  return None
+
+
 def _get_replay_id(use_vertex: bool, replays_prefix: str) -> str:
   test_name_ending = os.environ.get('PYTEST_CURRENT_TEST').split('::')[-1]
   test_name = (
@@ -127,7 +134,7 @@ def _get_replay_id(use_vertex: bool, replays_prefix: str) -> str:
 
 
 @pytest.fixture
-def client(use_vertex, replays_prefix, http_options, request):
+def client(use_vertex, replays_prefix, http_options, request, location_override):
   mode = request.config.getoption('--mode')
   if mode not in ['auto', 'record', 'replay', 'api', 'tap']:
     raise ValueError('Invalid mode: ' + mode)
@@ -171,8 +178,10 @@ def client(use_vertex, replays_prefix, http_options, request):
   # Get private arg.
   private = request.config.getoption('--private')
 
-  location_override = None
-  if use_vertex and 'tunings' in replays_prefix:
+  if not use_vertex:
+    location_override = None
+  elif location_override is None and 'tunings' in replays_prefix:
+    # Tuning jobs are not supported on the global endpoint.
     if os.environ.get('GOOGLE_CLOUD_LOCATION') == 'global':
       location_override = 'us-central1'
 

google/genai/tests/live_api/conftest.py

@@ -20,3 +20,16 @@ def http_options():
   Normally injected by pytest_helper.setup(); live tests use the SDK defaults.
   """
   return None
+
+
+@pytest.fixture
+def location_override():
+  """Pins the Vertex client to a region for the live model.
+
+  gemini-live-2.5-flash-native-audio is not served on the global endpoint: a
+  setup there is rejected with 1008 "Publisher model ... was not found". It is
+  available in us-central1, us-east5 and europe-west4. The Agent Platform
+  wrapper sets GOOGLE_CLOUD_LOCATION=global for the shared suite, so the live
+  tests override it here rather than changing the wrapper.
+  """
+  return 'us-central1'

google/genai/tests/live_api/test_live_session.py

@@ -44,11 +44,17 @@
 
 pytest_plugins = ('pytest_asyncio',)
 
-# The only live model family currently served. It is audio-native and rejects a
-# TEXT response modality outright ("The requested combination of response
-# modalities (TEXT) is not supported by the model"), so these tests request
-# AUDIO and turn on output transcription to get an assertable text signal.
-LIVE_MODEL = 'gemini-3.1-flash-live-preview'
+# Live models are backend-specific: gemini-3.1-flash-live-preview is served only
+# on the Gemini API, and gemini-live-2.5-flash-native-audio only on Vertex, and
+# there only regionally -- see the location_override fixture in this package's
+# conftest. Both are audio-native and reject a TEXT response modality outright
+# ("The requested combination of response modalities (TEXT) is not supported by
+# the model"), so these tests request AUDIO and turn on output transcription to
+# get an assertable text signal.
+LIVE_MODELS = {
+    False: 'gemini-3.1-flash-live-preview',
+    True: 'gemini-live-2.5-flash-native-audio',
+}
 
 # A live turn is an open-ended stream with no built-in deadline. Without this
 # bound a wedged receive would hang the nightly rather than fail it.
@@ -121,12 +127,13 @@ async def _say(session, text: str) -> None:
   )
 
 
+@pytest.mark.parametrize('use_vertex', [False, True])
 @pytest.mark.asyncio
-async def test_text_input(client):
+async def test_text_input(client, use_vertex):
   """A single text turn produces audio output and a matching transcription."""
   try:
     async with client.aio.live.connect(
-        model=LIVE_MODEL, config=_base_config()
+        model=LIVE_MODELS[use_vertex], config=_base_config()
     ) as session:
       await _say(session, 'Say hello.')
       turn = await _receive_turn(session)
@@ -138,12 +145,13 @@ async def test_text_input(client):
     raise
 
 
+@pytest.mark.parametrize('use_vertex', [False, True])
 @pytest.mark.asyncio
-async def test_multi_turn(client):
+async def test_multi_turn(client, use_vertex):
   """A second turn in the same session can see the first turn's context."""
   try:
     async with client.aio.live.connect(
-        model=LIVE_MODEL, config=_base_config()
+        model=LIVE_MODELS[use_vertex], config=_base_config()
     ) as session:
       await _say(session, 'Remember the number 42. Just acknowledge it.')
       first = await _receive_turn(session)
@@ -162,8 +170,9 @@ async def test_multi_turn(client):
     raise
 
 
+@pytest.mark.parametrize('use_vertex', [False, True])
 @pytest.mark.asyncio
-async def test_function_calling(client):
+async def test_function_calling(client, use_vertex):
   """The model requests a declared tool, and the session accepts its result."""
   turn_on_the_lights = types.FunctionDeclaration(
       name='turn_on_the_lights',
@@ -176,7 +185,7 @@ async def test_function_calling(client):
 
   try:
     async with client.aio.live.connect(
-        model=LIVE_MODEL, config=config
+        model=LIVE_MODELS[use_vertex], config=config
     ) as session:
       await _say(session, 'Please turn on the lights.')
       turn = await _receive_turn(session)
@@ -194,20 +203,31 @@ async def test_function_calling(client):
           ]
       )
       follow_up = await _receive_turn(session)
-      assert follow_up.transcript.strip(), (
-          'expected the model to respond after the tool result'
-      )
+      if not use_vertex:
+        # Vertex accepts the tool result and completes the turn, but emits an
+        # empty transcription and no audio for it, so only the Gemini API can be
+        # asserted on content here. Confirmed at the raw protocol level: the
+        # follow-up carries outputTranscription with empty text, then
+        # generationComplete and turnComplete.
+        assert follow_up.transcript.strip(), (
+            'expected the model to respond after the tool result'
+        )
   except Exception as e:  # pylint: disable=broad-except
     _skip_if_quota_exhausted(e)
     raise
 
 
+@pytest.mark.parametrize('use_vertex', [False])
 @pytest.mark.asyncio
-async def test_send_tool_response_without_id_raises(client):
-  """The Gemini API backend requires an id on every FunctionResponse."""
+async def test_send_tool_response_without_id_raises(client, use_vertex):
+  """The Gemini API backend requires an id on every FunctionResponse.
+
+  Gemini API only: live.py:409 puts this validation in the non-vertexai branch
+  of send_tool_response, so Vertex accepts an id-less FunctionResponse.
+  """
   try:
     async with client.aio.live.connect(
-        model=LIVE_MODEL, config=_base_config()
+        model=LIVE_MODELS[use_vertex], config=_base_config()
     ) as session:
       with pytest.raises(ValueError, match='must have an `id` field'):
         await session.send_tool_response(