test(e2e): gemini native — assert outcomes in tests (not fixtures), auth/model-path check, #121317 xfail

tests/e2e/core/providers/test_native_gemini_compaction.py

@@ -87,11 +87,10 @@ def run(tmp_path_factory: pytest.TempPathFactory) -> Run:
         Text(ANSWER_2, prompt_tokens=3_000),
     ]
     with GeminiFake(root / "fake", script, route=_summary_route) as fake:
+        # Outcomes are asserted by the tests (a rejected request must name the broken contract).
         first = nh.run_chat(home, "Read big1.txt and big2.txt.", env=fake.child_env())
-        assert first.returncode == 0, first.describe()
         second = nh.run_chat(home, "Now read big3.txt, then echo the marker.", env=fake.child_env(),
                              resume=nh.latest_session(home))
-        assert second.returncode == 0, second.describe()
     return Run(home, fake, [first, second])
 
 
@@ -100,6 +99,7 @@ def test_compaction_happens_on_the_wire(run: Run) -> None:
     # Without compaction the next request = previous request + (functionCall, functionResponse).
     assert len(after.contents) < len(before.contents) + 2, (len(before.contents), len(after.contents))
     assert SUMMARY_MARK in after.all_text(), "compaction summary never reached the next request"
+    assert all(t.returncode == 0 for t in run.turns), [t.describe() for t in run.turns]
     assert ANSWER_2 in run.turns[1].stdout, run.turns[1].describe()
 
 

tests/e2e/core/providers/test_native_gemini_tools.py

@@ -18,7 +18,17 @@ from dataclasses import dataclass
 import pytest
 
 from tests.e2e.core.providers import _native_helpers as nh
-from tests.fakes.providers.gemini_native import HERMES_ENV, Call, Calls, GeminiFake, Recorded, Text, hermes_model
+from tests.fakes.providers.gemini_native import (
+    API_KEY,
+    HERMES_ENV,
+    MODEL_ID,
+    Call,
+    Calls,
+    GeminiFake,
+    Recorded,
+    Text,
+    hermes_model,
+)
 
 SEED = "GEMINI-SEED-CANARY-5521"
 ECHO = "GEMINI-RESUME-ECHO-8813"
@@ -50,11 +60,11 @@ def run(tmp_path_factory: pytest.TempPathFactory) -> Run:
         Text(ANSWER_2),
     ]
     with GeminiFake(root / "fake", script) as fake:
+        # No asserts here: a failing turn must show up as the specific test assertion (a rejected
+        # request, a missing signature), not as a fixture error that hides which contract broke.
         first = nh.run_chat(home, "Read seed.txt and tell me what it says.", env=fake.child_env())
-        assert first.returncode == 0, first.describe()
         sid = nh.latest_session(home)
         second = nh.run_chat(home, "Now echo the marker in the terminal.", env=fake.child_env(), resume=sid)
-        assert second.returncode == 0, second.describe()
     return Run(home, fake, [first, second], sid)
 
 
@@ -81,7 +91,7 @@ def test_function_call_round_trip_pairs_response_and_persists(run: Run) -> None:
     assert SEED in json.dumps(resp["response"]), resp
     part = _call_parts(follow_up)[call_id]
     assert part.get("thoughtSignature") == run.fake.call_signatures[call_id], part
-    assert ANSWER_1 in run.turns[0].stdout, run.turns[0].describe()
+    assert run.turns[0].returncode == 0 and ANSWER_1 in run.turns[0].stdout, run.turns[0].describe()
 
     rows = nh.messages(run.home, run.session_id)
     ids = [tc.get("id") for r in rows if r["role"] == "assistant" for tc in nh.tool_calls_of(r)]
@@ -107,13 +117,17 @@ def test_resume_replays_thought_signatures_verbatim(run: Run) -> None:
     final = run.main()[3]
     assert _call_parts(final)[second_id].get("thoughtSignature") == run.fake.call_signatures[second_id]
     assert ECHO in json.dumps(_responses(final).get(second_id))
-    assert ANSWER_2 in run.turns[1].stdout, run.turns[1].describe()
+    assert run.turns[1].returncode == 0 and ANSWER_2 in run.turns[1].stdout, run.turns[1].describe()
     nh.assert_no_duplicate_assistant_text(nh.messages(run.home), ANSWER_2)
 
 
-def test_no_egress_beyond_the_google_host(run: Run) -> None:
-    """Every generate call carried the API key; no other request reached a Google path."""
-    for rec in run.fake.generate_calls():
-        assert rec.headers.get("x-goog-api-key") or rec.query.get("key"), rec.headers
-    other = [r.path for r in run.fake.requests if not r.rpc]
-    assert other == [], other
+def test_every_request_authenticates_and_targets_the_configured_model(run: Run) -> None:
+    """The key from ``.env`` travels as ``x-goog-api-key`` (or ``key=``) on every generate call, to
+    ``/v1beta/models/<configured model>``; nothing else was requested from the Google host."""
+    calls = run.fake.generate_calls()
+    assert calls
+    for rec in calls:
+        key = rec.headers.get("x-goog-api-key") or (rec.query.get("key") or [""])[0]
+        assert key == API_KEY, f"generate call without the configured API key: {rec.headers}"
+        assert (rec.version, rec.model) == ("v1beta", MODEL_ID), rec.path
+    assert [r.path for r in run.fake.requests if not r.rpc] == []