fix(tests): accept either initial event type from xAI realtime

xAI's Grok Voice Agent API used to emit 'conversation.created' as the
first event over the WebSocket. It has since shipped a fully
OpenAI-compatible 'session.created' event (and may still emit the
legacy 'conversation.created' on some routes), which breaks the
strict-equality assertion in the realtime e2e test:

    AssertionError: Expected conversation.created, got session.created

This is an upstream behavior change, not a regression in our code.
Loosen the base realtime test so get_initial_event_type() may return a
tuple of acceptable event types, and have the xAI subclass accept both
'conversation.created' and 'session.created'. The OpenAI subclasses
keep their single-string contract unchanged.
This commit is contained in:
mateo-berri 2026-05-21 01:13:24 +00:00
parent 198d6cc960
commit 944008b52b
No known key found for this signature in database
2 changed files with 24 additions and 10 deletions

View file

@ -10,7 +10,7 @@ import json
import os
import sys
from abc import ABC, abstractmethod
from typing import Optional
from typing import Optional, Tuple, Union
import pytest
import websockets
@ -153,8 +153,14 @@ class BaseRealtimeTest(ABC):
pass
@abstractmethod
def get_initial_event_type(self) -> str:
"""Return the expected initial event type (e.g., 'session.created' or 'conversation.created')"""
def get_initial_event_type(self) -> Union[str, Tuple[str, ...]]:
"""Return the expected initial event type(s).
May return a single event type (e.g. ``'session.created'``) or a tuple
of acceptable types when the upstream provider can legitimately emit
more than one initial event (e.g. xAI's Grok Voice Agent has shipped
both ``conversation.created`` and ``session.created``).
"""
pass
def get_skip_reason(self) -> str:
@ -229,9 +235,14 @@ class BaseRealtimeTest(ABC):
# Verify initial event
initial_event = websocket_client.messages_received[0]
expected_event_type = self.get_initial_event_type()
if isinstance(expected_event_type, str):
allowed_event_types: Tuple[str, ...] = (expected_event_type,)
else:
allowed_event_types = tuple(expected_event_type)
assert (
initial_event["type"] == self.get_initial_event_type()
), f"Expected {self.get_initial_event_type()}, got {initial_event.get('type')}"
initial_event["type"] in allowed_event_types
), f"Expected one of {allowed_event_types}, got {initial_event.get('type')}"
@pytest.mark.asyncio
async def test_realtime_with_query_params(self):

View file

@ -7,6 +7,7 @@ Uses the base test class to ensure consistent behavior across providers.
import os
import sys
from typing import Tuple
import pytest
@ -19,10 +20,12 @@ class TestXAIRealtime(BaseRealtimeTest):
"""
E2E tests for xAI Realtime API.
xAI's Grok Voice Agent API is OpenAI-compatible but uses:
- Different initial event: "conversation.created" instead of "session.created"
- Different endpoint: wss://api.x.ai/v1/realtime
xAI's Grok Voice Agent API is OpenAI-compatible:
- Endpoint: wss://api.x.ai/v1/realtime
- Model: grok-4-1-fast-non-reasoning
- Initial event: historically "conversation.created"; xAI has since shipped
"session.created" (matching OpenAI). Accept either to avoid spurious
failures whenever xAI flips the wire format.
"""
def get_model(self) -> str:
@ -31,5 +34,5 @@ class TestXAIRealtime(BaseRealtimeTest):
def get_api_key_env_var(self) -> str:
return "XAI_API_KEY"
def get_initial_event_type(self) -> str:
return "conversation.created"
def get_initial_event_type(self) -> Tuple[str, ...]:
return ("conversation.created", "session.created")