test(copilot): route direct-token fixtures through authenticated peers (#50307)

Co-authored-by: mateo <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/integration/_support/device_login.py

@@ -378,14 +378,17 @@ def _tunnel(
 
 
 @contextmanager
-def device_login_peers(directory: Path) -> Generator[Peers]:
+def device_login_peers(
+    directory: Path, provider: Callable[[Request], Reply] | None = None
+) -> Generator[Peers]:
     cert, key = write_self_signed_cert(directory, (*AUTH_HOSTS, COPILOT_API_HOST))
     switches: Final = Switches(threading.Event(), threading.Event(), threading.Event(), threading.Event())
     authorities: Final[SimpleQueue[str]] = SimpleQueue()
     hangups: Final[SimpleQueue[Hangup]] = SimpleQueue()
+    respond: Final = provider if provider is not None else _provider_api(rv.ResponsesVendor())
     with (
-        wire_server(_provider_api(rv.ResponsesVendor())) as api,
-        wire_server(_provider_api(rv.ResponsesVendor()), tls=server_context(cert, key)) as copilot_api,
+        wire_server(respond) as api,
+        wire_server(respond, tls=server_context(cert, key)) as copilot_api,
         wire_server(_auth_hosts(api.url, switches.grant), tls=server_context(cert, key)) as auth,
         _tunnel(auth, copilot_api, switches, authorities, hangups) as tunnel,
     ):

tests/integration/messages_endpoint/providers/anthropic/test_anthropic_safeguards_beta_wire.py

@@ -15,6 +15,7 @@ import openai
 import pytest
 from integration._support.client import Gateway, eventually, gateway_from_environment, object_value
 from integration._support.database import read_rows
+from integration._support.device_login import device_login_peers, write_copilot_access
 from integration._support.process import owned_proxy
 from integration._support.vertex import service_account_json
 from integration._support.wire import Reply, Request, Wire, wire_server
@@ -665,18 +666,18 @@ def test_github_copilot_messages_forward_safeguards_without_deriving_the_beta(
     marker: Final = uuid.uuid4().hex
     token_dir: Final = tmp_path / "copilot"
     token_dir.mkdir()
-    with wire_server(anthropic_peer) as wire:
-        (token_dir / "api-key.json").write_text(
-            json.dumps({"token": "synthetic-copilot-token", "expires_at": 4102444800, "endpoints": {"api": wire.url}})
-        )
+    with device_login_peers(tmp_path, provider=anthropic_peer) as peers:
+        write_copilot_access(token_dir, "synthetic-copilot-token")
         with (
-            owned_proxy(gateway, tmp_path, {"GITHUB_COPILOT_TOKEN_DIR": str(token_dir)}) as owned,
+            owned_proxy(
+                gateway, tmp_path, {**peers.environment(), "GITHUB_COPILOT_TOKEN_DIR": str(token_dir)}
+            ) as owned,
             owned.scenario() as scenario,
         ):
-            model: Final = scenario.model(model="github_copilot/claude-sonnet-4-5", api_base=wire.url)
+            model: Final = scenario.model(model="github_copilot/claude-sonnet-4-5", api_base=peers.api.url)
             response: Final = _post(owned, messages_body(model, marker), beta=_CLIENT_BETAS_WITH_DANGEROUS)
             _assert_answered(response, marker)
-            observed: Final = _only(wire)
+            observed: Final = _only(peers.api.secure)
             _expect_messages(
                 observed, target="/v1/messages", auth=("authorization", "Bearer synthetic-copilot-token"), beta=None
             )

tests/integration/providers/test_responses_minted_reasoning_replay_chaos.py

@@ -18,6 +18,7 @@ import pytest
 import websockets
 import yaml
 from integration._support import claude_code as cc
+from integration._support import device_login as dl
 from integration._support import responses_vendor as rv
 from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment
 from integration._support.database import read_rows
@@ -365,6 +366,7 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_dropping_the_minted_i
 @dataclass(frozen=True, slots=True)
 class _Rig:
     wire: Wire
+    copilot_wire: Wire
     proxy: OwnedProxy
     cert: Path
     key: Path
@@ -378,12 +380,12 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
     chatgpt: Final = directory / "chatgpt"
     copilot.mkdir()
     chatgpt.mkdir()
-    with gateway_from_environment() as gateway, wire_server(rv.ResponsesVendor().respond) as wire:
-        (copilot / "api-key.json").write_text(
-            json.dumps(
-                {"token": "synthetic-copilot-token", "expires_at": time.time() + 3600, "endpoints": {"api": wire.url}}
-            )
-        )
+    with (
+        gateway_from_environment() as gateway,
+        wire_server(rv.ResponsesVendor().respond) as wire,
+        dl.device_login_peers(directory, provider=rv.ResponsesVendor().respond) as peers,
+    ):
+        dl.write_copilot_access(copilot, "synthetic-copilot-token")
         (chatgpt / "auth.json").write_text(
             json.dumps(
                 {
@@ -394,15 +396,15 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
             )
         )
         overrides: Final = {
+            **peers.environment(),
             "GITHUB_COPILOT_TOKEN_DIR": str(copilot),
             "CHATGPT_TOKEN_DIR": str(chatgpt),
             "CHATGPT_API_BASE": wire.url,
-            "SSL_CERT_FILE": str(cert),
             "HTTP_PROXY": wire.url,
             "NO_PROXY": "127.0.0.1,localhost",
         }
         with owned_proxy_process(gateway, directory, overrides, workers=2) as owned:
-            yield _Rig(wire, owned, cert, key)
+            yield _Rig(wire, peers.api.secure, owned, cert, key)
 
 
 def _replay(gateway: Gateway, model: str, history: list[dict[str, JsonValue]], stream: bool) -> httpx.Response:
@@ -434,13 +436,14 @@ def test_login_backed_and_foundry_deployments_forward_the_minted_item_unchanged(
     history: Final = rv.agents_sdk_history(marker, minted)
     api_base: Final = _FOUNDRY_BASE if deployment.label.startswith("azure_ai") else rig.wire.url
     rig.wire.drain()
+    rig.copilot_wire.drain()
     with rig.proxy.gateway.scenario() as scenario:
         parameters: Final[dict[str, JsonValue]] = {"model": deployment.model, "api_base": api_base}
         model: Final = scenario.model(
             **parameters, **({} if deployment.api_key is None else {"api_key": deployment.api_key})
         )
         response: Final = _replay(rig.proxy.gateway, model, history, stream)
-        received: Final = rig.wire.drain()
+        received: Final = (*rig.wire.drain(), *rig.copilot_wire.drain())
         assert len(received) == 1, [(request.method, request.target) for request in received]
         target: Final = urlsplit(received[0].target)
         assert target.path.endswith("/responses"), received[0].target