feat(voice): add authorization to WebSocket endpoints

voice/src/vonage_voice/models/common.py

@@ -32,6 +32,21 @@ class Sip(BaseModel):
     type: Channel = Channel.SIP
 
 
+class WebsocketAuthorization(BaseModel):
+    """Authorization settings for a WebSocket endpoint.
+
+    Args:
+        type (Literal['vonage', 'custom']): The authorization mode. Use `vonage` to have
+            Vonage generate and send a JWT for you, or `custom` to provide your own
+            Authorization header value.
+        value (str, Optional): Authorization header value to send when `type` is
+            `custom`. Ignored for `vonage`.
+    """
+
+    type: Literal['vonage', 'custom']
+    value: Optional[str] = None
+
+
 class Websocket(BaseModel):
     """Model for a WebSocket connection.
 
@@ -40,13 +55,16 @@ class Websocket(BaseModel):
         content_type (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000']): The content
             type of the audio stream.
         headers (Optional[dict]): The headers to include with the WebSocket connection.
+        authorization (WebsocketAuthorization, Optional): Authorization configuration for
+            the WebSocket handshake.
     """
 
     uri: str = Field(..., min_length=1)
     content_type: Literal['audio/l16;rate=8000', 'audio/l16;rate=16000'] = Field(
         'audio/l16;rate=16000', serialization_alias='content-type'
     )
     headers: Optional[dict] = None
+    authorization: Optional[WebsocketAuthorization] = None
     type: Channel = Channel.WEBSOCKET
 
 

voice/src/vonage_voice/models/connect_endpoints.py

@@ -3,6 +3,7 @@
 from pydantic import BaseModel, Field
 from vonage_utils.types import Dtmf, PhoneNumber, SipUri
 
+from .common import WebsocketAuthorization
 from .enums import ConnectEndpointType
 
 
@@ -55,13 +56,16 @@ class WebsocketEndpoint(BaseModel):
         contentType (Literal['audio/l16;rate=8000', 'audio/l16;rate=16000']): The internet
             media type for the audio you are streaming.
         headers (Optional[dict]): The headers to include with the WebSocket connection.
+        authorization (WebsocketAuthorization, Optional): Authorization configuration for
+            the WebSocket handshake.
     """
 
     uri: str
     contentType: Literal['audio/l16;rate=16000', 'audio/l16;rate=8000'] = Field(
         None, serialization_alias='content-type'
     )
     headers: Optional[dict] = None
+    authorization: Optional[WebsocketAuthorization] = None
     type: ConnectEndpointType = ConnectEndpointType.WEBSOCKET
 
 

voice/tests/test_ncco_actions.py

@@ -105,10 +105,18 @@ def test_create_connect_endpoints():
         uri='wss://example.com',
         contentType='audio/l16;rate=8000',
         headers={'asdf': 'qwer'},
+        authorization={
+            'type': 'custom',
+            'value': 'Bearer eyJhbGciOi...',
+        },
     ).model_dump(by_alias=True) == {
         'uri': 'wss://example.com',
         'content-type': 'audio/l16;rate=8000',
         'headers': {'asdf': 'qwer'},
+        'authorization': {
+            'type': 'custom',
+            'value': 'Bearer eyJhbGciOi...',
+        },
         'type': 'websocket',
     }
 

voice/tests/test_voice.py

@@ -11,6 +11,7 @@
     ListCallsFilter,
     Sip,
     TtsStreamOptions,
+    Websocket,
 )
 from vonage_voice.errors import VoiceError
 from vonage_voice.models.ncco import Talk
@@ -112,15 +113,13 @@ def test_create_call_basic_answer_url():
     build_response(
         path, 'POST', 'https://api.nexmo.com/v1/calls', 'create_call.json', 201
     )
+    ws = Websocket(
+        uri='wss://example.com/websocket',
+        content_type='audio/l16;rate=8000',
+        headers={'key': 'value'},
+    )
     call = CreateCallRequest(
-        to=[
-            {
-                'type': 'websocket',
-                'uri': 'wss://example.com/websocket',
-                'content_type': 'audio/l16;rate=8000',
-                'headers': {'key': 'value'},
-            }
-        ],
+        to=[ws],
         answer_url=['https://example.com/answer'],
         random_from_number=True,
     )
@@ -133,6 +132,33 @@ def test_create_call_basic_answer_url():
     assert response.conversation_uuid == 'CON-2be039b2-d0a4-4274-afc8-d7b241c7c044'
 
 
+@responses.activate
+def test_create_call_websocket_authorization_custom():
+    build_response(
+        path, 'POST', 'https://api.nexmo.com/v1/calls', 'create_call.json', 201
+    )
+    ncco = [Talk(text='Hello world')]
+    ws = Websocket(
+        uri='wss://example.com/websocket',
+        content_type='audio/l16;rate=16000',
+        headers={'key': 'value'},
+        authorization={'type': 'custom', 'value': 'Bearer eyJhbGciOi...'},
+    )
+    call = CreateCallRequest(
+        ncco=ncco,
+        to=[ws],
+        random_from_number=True,
+    )
+
+    response = voice.create_call(call)
+    body = json.loads(voice.http_client.last_request.body)
+    assert body['to'][0]['authorization'] == {
+        'type': 'custom',
+        'value': 'Bearer eyJhbGciOi...',
+    }
+    assert type(response) == CreateCallResponse
+
+
 @responses.activate
 def test_create_call_answer_url_options():
     build_response(