Added live test Media streaming and Transcription (#35960)

* Added live test Media streaming and Transcription

* updating the live tests for media streaming and transcription

---------

Co-authored-by: Vinothini Dharmaraj <[email protected]>

Author: v-durgeshs

sdk/communication/azure-communication-callautomation/azure/communication/callautomation/_call_automation_client.py

@@ -224,8 +224,8 @@ def create_call(
             targets = [serialize_identifier(p) for p in target_participant]
         except TypeError:
             targets = [serialize_identifier(target_participant)]
-        media_config = media_streaming.to_generated() if media_streaming else None
-        transcription_config = transcription.to_generated() if transcription else None
+        media_config = media_streaming._to_generated() if media_streaming else None # pylint:disable=protected-access
+        transcription_config = transcription._to_generated() if transcription else None # pylint:disable=protected-access
         create_call_request = CreateCallRequest(
             targets=targets,
             callback_uri=callback_url,
@@ -337,9 +337,9 @@ def answer_call(
             call_intelligence_options=call_intelligence_options,
             answered_by=serialize_communication_user_identifier(
                 self.source) if self.source else None,
-            media_streaming_options=media_streaming._to_generated(
-            ) if media_streaming else None,
-            transcription_options=transcription._to_generated()
+            media_streaming_options=media_streaming._to_generated() # pylint:disable=protected-access
+            if media_streaming else None,
+            transcription_options=transcription._to_generated() # pylint:disable=protected-access
             if transcription else None,
             operation_context=operation_context
         )

sdk/communication/azure-communication-callautomation/azure/communication/callautomation/aio/_call_automation_client_async.py

@@ -210,8 +210,8 @@ async def create_call(
             targets = [serialize_identifier(p) for p in target_participant]
         except TypeError:
             targets = [serialize_identifier(target_participant)]
-        media_config = media_streaming.to_generated() if media_streaming else None
-        transcription_config = transcription.to_generated() if transcription else None
+        media_config = media_streaming._to_generated() if media_streaming else None # pylint:disable=protected-access
+        transcription_config = transcription._to_generated() if transcription else None # pylint:disable=protected-access
         create_call_request = CreateCallRequest(
             targets=targets,
             callback_uri=callback_url,
@@ -320,9 +320,9 @@ async def answer_call(
         answer_call_request = AnswerCallRequest(
             incoming_call_context=incoming_call_context,
             callback_uri=callback_url,
-            media_streaming_options=media_streaming.to_generated(
+            media_streaming_options=media_streaming._to_generated( # pylint:disable=protected-access
             ) if media_streaming else None,
-            transcription_options=transcription.to_generated()
+            transcription_options=transcription._to_generated() # pylint:disable=protected-access
             if transcription else None,
             answered_by=serialize_communication_user_identifier(
                 self.source) if self.source else None,

sdk/communication/azure-communication-callautomation/tests/callautomation_test_case.py

@@ -37,12 +37,16 @@ def setup_class(cls):
             cls.servicebus_connection_str = os.environ.get('SERVICEBUS_STRING')
             cls.dispatcher_endpoint = os.environ.get('DISPATCHER_ENDPOINT')
             cls.file_source_url = os.environ.get('FILE_SOURCE_URL')
+            cls.cognitive_service_endpoint = os.environ.get('COGNITIVE_SERVICE_ENDPOINT')
+            cls.transport_url = os.environ.get('TRANSPORT_URL')
         else:
             print("Recorded Test")
             cls.connection_str = "endpoint=https://someEndpoint/;accesskey=someAccessKeyw=="
             cls.servicebus_connection_str =  "Endpoint=sb://someEndpoint/;SharedAccessKeyName=somekey;SharedAccessKey=someAccessKey="
             cls.dispatcher_endpoint = "https://REDACTED.azurewebsites.net"
             cls.file_source_url = "https://REDACTED/prompt.wav"
+            cls.cognitive_service_endpoint = "https://REDACTED.cognitiveservices.azure.com"
+            cls.transport_url ="wss://REDACTED"
 
         cls.dispatcher_callback = cls.dispatcher_endpoint + '/api/servicebuscallback/events'
         cls.identity_client = CommunicationIdentityClient.from_connection_string(cls.connection_str)
@@ -266,4 +270,55 @@ def terminate_call(self, unique_id) -> None:
                     raise ValueError("Receiver CallDisconnected event is None")
         finally:
             while unique_id in self.wait_for_event_flags: self.wait_for_event_flags.remove(unique_id)
-            pass
+            pass
+        
+    def establish_callconnection_voip_with_streaming_options(self, caller, target, options, is_transcription) -> tuple:
+        call_automation_client_caller = CallAutomationClient.from_connection_string(self.connection_str, source=caller) # for creating call
+        call_automation_client_target = CallAutomationClient.from_connection_string(self.connection_str, source=target) # answering call, all other actions
+
+        unique_id = self._unique_key_gen(caller, target)
+        if is_live():
+            dispatcher_url = f"{self.dispatcher_endpoint}/api/servicebuscallback/subscribe?q={unique_id}"
+            response = requests.post(dispatcher_url)
+
+            if response is None:
+                raise ValueError("Response cannot be None")
+
+            print(f"Subscription to dispatcher of {unique_id}: {response.status_code}")
+
+            self.wait_for_event_flags.append(unique_id)
+            thread = threading.Thread(target=self._message_awaiter, args=(unique_id,))
+            thread.start()
+
+        # create a call with options either media streaming or transcription.
+        create_call_result = call_automation_client_caller.create_call(
+            target_participant=target, 
+            callback_url=(self.dispatcher_callback + "?q={}".format(unique_id)),
+            media_streaming=options if not is_transcription else None,
+            transcription=options if is_transcription else None,
+            cognitive_services_endpoint=self.cognitive_service_endpoint if is_transcription else None
+            )
+
+        if create_call_result is None:
+            raise ValueError("Invalid create_call_result")
+
+        caller_connection_id = create_call_result.call_connection_id
+        if caller_connection_id is None:
+            raise ValueError("Caller connection ID is None")
+
+        # wait for incomingCallContext
+        incoming_call_event = self.check_for_event('IncomingCall', unique_id, timedelta(seconds=30))
+        if incoming_call_event is None:
+            raise ValueError("incoming_call_event is None")
+        incoming_call_context = incoming_call_event["incomingCallContext"]
+
+        # answer the call
+        answer_call_result = call_automation_client_target.answer_call(incoming_call_context=incoming_call_context, callback_url=self.dispatcher_callback)
+        if answer_call_result is None:
+            raise ValueError("Invalid answer_call result")
+
+        call_connection_caller = CallConnectionClient.from_connection_string(self.connection_str, caller_connection_id)
+        call_connection_target = CallConnectionClient.from_connection_string(self.connection_str, answer_call_result.call_connection_id)
+        self.open_call_connections[unique_id] = call_connection_caller
+
+        return unique_id, call_connection_caller, call_connection_target

sdk/communication/azure-communication-callautomation/tests/test_e2e_media_client.py

@@ -12,7 +12,13 @@
 from azure.communication.callautomation import (
     FileSource,
     DtmfTone,
-    PhoneNumberIdentifier
+    PhoneNumberIdentifier,
+    MediaStreamingOptions,
+    MediaStreamingContentType,
+    MediaStreamingTransportType,
+    MediaStreamingAudioChannelType,
+    TranscriptionOptions,
+    TranscriptionTransportType
 )
 from callautomation_test_case import CallAutomationRecordedTestCase
 from azure.communication.callautomation._shared.models import identifier_from_raw_id
@@ -178,4 +184,139 @@ def test_add_and_hold_unhold_participant_in_a_call(self):
             raise ValueError("Failed to unhold participant")
 
         self.terminate_call(unique_id)
-        return
+        return
+    
+    @recorded_by_proxy 
+    def test_start_stop_media_streaming_in_a_call(self):
+        
+        # try to establish the call
+        caller = self.identity_client.create_user()
+        target = self.identity_client.create_user()
+
+        media_streaming_options=MediaStreamingOptions(
+            transport_url=self.transport_url,
+            transport_type=MediaStreamingTransportType.WEBSOCKET,
+            content_type=MediaStreamingContentType.AUDIO,
+            audio_channel_type=MediaStreamingAudioChannelType.MIXED,
+            start_media_streaming=False
+            )        
+
+        unique_id, call_connection, _ = self.establish_callconnection_voip_with_streaming_options(caller, target, media_streaming_options, False)
+
+        # check returned events
+        connected_event = self.check_for_event('CallConnected', call_connection._call_connection_id, timedelta(seconds=15))
+        participant_updated_event = self.check_for_event('ParticipantsUpdated', call_connection._call_connection_id, timedelta(seconds=15))
+
+        if connected_event is None:
+            raise ValueError("Caller CallConnected event is None")
+        if participant_updated_event is None:
+            raise ValueError("Caller ParticipantsUpdated event is None")
+        
+        # start media streaming.
+        call_connection.start_media_streaming()
+        
+        # check for MediaStreamingStarted event
+        media_streaming_started = self.check_for_event('MediaStreamingStarted', call_connection._call_connection_id, timedelta(seconds=30))
+        if media_streaming_started is None:
+            raise ValueError("MediaStreamingStarted event is None")
+
+        time.sleep(3)
+        
+        # check for media streaming subscription from call connection properties for media streaming started event
+        call_connection_properties=call_connection.get_call_properties()
+        if call_connection_properties is None:
+            raise ValueError("call_connection_properties is None")
+        if call_connection_properties.media_streaming_subscription is None:
+            raise ValueError("call_connection_properties.media_streaming_subscription is None")
+        if call_connection_properties.media_streaming_subscription.state!='active':
+            raise ValueError("media streaming state is invalid for MediaStreamingStarted event")
+
+        # stop media streaming.
+        call_connection.stop_media_streaming()
+
+        # check for MediaStreamingStopped event
+        media_streaming_stopped = self.check_for_event('MediaStreamingStopped', call_connection._call_connection_id, timedelta(seconds=30))
+        if media_streaming_stopped is None:
+            raise ValueError("MediaStreamingStopped event is None")
+        
+        # check for media streaming subscription from call connection properties for media streaming stopped event
+        call_connection_properties=call_connection.get_call_properties()
+        if call_connection_properties is None:
+            raise ValueError("call_connection_properties is None")
+        if call_connection_properties.media_streaming_subscription is None:
+            raise ValueError("call_connection_properties.media_streaming_subscription is None")
+        if call_connection_properties.media_streaming_subscription.state!='inactive':
+            raise ValueError("media streaming state is invalid for MediaStreamingStopped event")
+
+        self.terminate_call(unique_id)
+        return
+    
+    @recorded_by_proxy
+    def test_start_stop_transcription_in_call(self):
+        # try to establish the call
+        caller = self.identity_client.create_user()
+        target = self.identity_client.create_user()
+
+        transcription_options=TranscriptionOptions(
+            transport_url=self.transport_url,
+            transport_type=TranscriptionTransportType.WEBSOCKET,
+            locale="en-US",
+            start_transcription=False)
+
+        unique_id, call_connection, _ = self.establish_callconnection_voip_with_streaming_options(caller, target, transcription_options, True)
+
+        # check returned events
+        connected_event = self.check_for_event('CallConnected', call_connection._call_connection_id, timedelta(seconds=15))
+        participant_updated_event = self.check_for_event('ParticipantsUpdated', call_connection._call_connection_id, timedelta(seconds=15))
+
+        if connected_event is None:
+            raise ValueError("Caller CallConnected event is None")
+        if participant_updated_event is None:
+            raise ValueError("Caller ParticipantsUpdated event is None")
+        
+        # start transcription
+        call_connection.start_transcription(locale="en-ca")
+        
+        # check for TranscriptionStarted event
+        transcription_started = self.check_for_event('TranscriptionStarted', call_connection._call_connection_id, timedelta(seconds=30))
+        if transcription_started is None:
+            raise ValueError("TranscriptionStarted event is None")
+
+        # check for transcription subscription from call connection properties for transcription started event
+        call_connection_properties=call_connection.get_call_properties()
+        if call_connection_properties is None:
+            raise ValueError("call_connection_properties is None")
+        if call_connection_properties.transcription_subscription is None:
+            raise ValueError("call_connection_properties.transcription_subscription is None")
+        if call_connection_properties.transcription_subscription.state!='active':
+            raise ValueError("transcription subscription state is invalid for TranscriptionStarted event")
+
+        time.sleep(3)
+        call_connection.update_transcription(locale="en-gb")
+        
+        # check for TranscriptionUpdated event
+        transcription_updated = self.check_for_event('TranscriptionUpdated', call_connection._call_connection_id, timedelta(seconds=30))
+        if transcription_updated is None:
+            raise ValueError("TranscriptionUpdated event is None")
+        
+        time.sleep(3)
+
+        # stop transcription
+        call_connection.stop_transcription()
+        
+        # check for TranscriptionStopped event
+        transcription_stopped = self.check_for_event('TranscriptionStopped', call_connection._call_connection_id, timedelta(seconds=30))
+        if transcription_stopped is None:
+            raise ValueError("TranscriptionStopped event is None")
+        
+        # check for transcription subscription from call connection properties for transcription stopped event
+        call_connection_properties=call_connection.get_call_properties()
+        if call_connection_properties is None:
+            raise ValueError("call_connection_properties is None")
+        if call_connection_properties.transcription_subscription is None:
+            raise ValueError("call_connection_properties.transcription_subscription is None")
+        if call_connection_properties.transcription_subscription.state!='inactive':
+            raise ValueError("transcription subscription state is invalid for TranscriptionStopped event")
+        
+        self.terminate_call(unique_id)
+        return