# ------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # -------------------------------------------------------------------------- import json import threading import time import os import inspect import asyncio from datetime import datetime, timedelta from typing import Dict, Any, List, Optional import requests from azure.servicebus import ServiceBusClient from azure.identity import AzureCliCredential from devtools_testutils import AzureRecordedTestCase, is_live from devtools_testutils.helpers import get_test_id from azure.communication.callautomation import CommunicationIdentifierKind, CommunicationIdentifier from azure.communication.callautomation.aio import CallAutomationClient, CallConnectionClient from azure.communication.callautomation._shared.models import identifier_from_raw_id from azure.communication.identity.aio import CommunicationIdentityClient from azure.communication.phonenumbers import PhoneNumbersClient class CallAutomationRecordedTestCaseAsync(AzureRecordedTestCase): @classmethod def setup_class(cls): if is_live(): print("Live Test") cls.connection_str = os.environ.get("COMMUNICATION_LIVETEST_STATIC_CONNECTION_STRING") cls.servicebus_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_str = "redacted.servicebus.windows.net" cls.dispatcher_endpoint = "https://REDACTED.azurewebsites.net" cls.file_source_url = "https://REDACTED/prompt.wav" cls.cognitive_service_endpoint = "https://sanitized/" cls.transport_url = "wss://sanitized/ws" cls.credential = AzureCliCredential() cls.dispatcher_callback = cls.dispatcher_endpoint + "/api/servicebuscallback/events" # Initialize async clients as None - they will be created per test cls.identity_client = None cls.phonenumber_client = PhoneNumbersClient.from_connection_string(cls.connection_str) cls.service_bus_client = ServiceBusClient( fully_qualified_namespace=cls.servicebus_str, credential=cls.credential ) cls.wait_for_event_flags = [] cls.event_store: Dict[str, Dict[str, Any]] = {} cls.event_to_save: Dict[str, Dict[str, Any]] = {} cls.open_call_connections: Dict[str, CallConnectionClient] = {} @classmethod def teardown_class(cls): cls.wait_for_event_flags.clear() # Clear open connections - they should be cleaned up in individual tests cls.open_call_connections.clear() def setup_method(self, method): try: self.test_name = get_test_id().split("/")[-1] except (AttributeError, TypeError): # Fallback when get_test_id() returns None or fails self.test_name = getattr(method, "__name__", "unknown_test") self.event_store: Dict[str, Dict[str, Any]] = {} self.event_to_save: Dict[str, Dict[str, Any]] = {} self._prepare_events_recording() # Create identity client for this test self.identity_client = CommunicationIdentityClient.from_connection_string(self.connection_str) def teardown_method(self, method): # Close the identity client properly async def close_identity_client(): if hasattr(self, "identity_client") and self.identity_client: try: await self.identity_client.close() except Exception as e: print(f"Error closing identity client in teardown_method: {e}") try: loop = asyncio.get_event_loop() if not loop.is_closed(): loop.run_until_complete(close_identity_client()) except RuntimeError: # Event loop might be closed, create a new one loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(close_identity_client()) finally: loop.close() except Exception as e: print(f"Error in teardown_method: {e}") self._record_method_events() self.event_store: Dict[str, Dict[str, Any]] = {} self.event_to_save: Dict[str, Dict[str, Any]] = {} @staticmethod def _format_string(s) -> str: s1 = f"{s[:12]}-{s[12:16]}-{s[16:20]}-{s[20:24]}-{s[24:36]}" s2 = f"{s[36:44]}-{s[44:48]}-{s[48:52]}-{s[52:56]}-{s[56:]}" return f"{s1}_{s2}" @staticmethod def _parse_ids_from_identifier(identifier: CommunicationIdentifier) -> str: if identifier is None: raise ValueError("Identifier cannot be None") elif identifier.kind == CommunicationIdentifierKind.COMMUNICATION_USER: return CallAutomationRecordedTestCaseAsync._format_string("".join(filter(str.isalnum, identifier.raw_id))) elif identifier.kind == CommunicationIdentifierKind.PHONE_NUMBER: return identifier.raw_id else: raise ValueError("Identifier type not supported") @staticmethod def _event_key_gen(event_type: str) -> str: return event_type @staticmethod def _unique_key_gen( caller_identifier: CommunicationIdentifier, receiver_identifier: CommunicationIdentifier ) -> str: return CallAutomationRecordedTestCaseAsync._parse_ids_from_identifier( caller_identifier ) + CallAutomationRecordedTestCaseAsync._parse_ids_from_identifier(receiver_identifier) def _get_test_event_file_name(self): script_dir = os.path.dirname(os.path.realpath(__file__)) file_path = os.path.join(script_dir, "events", f"{self.test_name}.event.json") return file_path def _message_awaiter(self, unique_id) -> None: service_bus_receiver = self.service_bus_client.get_queue_receiver(queue_name=unique_id) while unique_id in self.wait_for_event_flags: received_messages = service_bus_receiver.receive_messages(max_wait_time=20) for msg in received_messages: body_bytes = b"".join(msg.body) body_str = body_bytes.decode("utf-8") mapper = json.loads(body_str) if "incomingCallContext" in mapper: caller = identifier_from_raw_id(mapper["from"]["rawId"]) receiver = identifier_from_raw_id(mapper["to"]["rawId"]) unique_id = self._unique_key_gen(caller, receiver) key = self._event_key_gen("IncomingCall") print("EventRegistration(IncomingCall):" + key) self.event_store[key] = mapper self.event_to_save[key] = mapper else: if isinstance(mapper, list): mapper = mapper[0] if mapper["type"]: key = self._event_key_gen(mapper["type"].split(".")[-1]) print("EventRegistration:" + key) self.event_store[key] = mapper self.event_to_save[key] = mapper service_bus_receiver.complete_message(msg) time.sleep(1) return def _prepare_events_recording(self) -> None: # only load during playback try: if not is_live(): file_path = self._get_test_event_file_name() with open(file_path, "r") as json_file: self.event_store = json.load(json_file) except Exception as e: # Don't fail test setup for file I/O issues print(f"Warning: Event loading failed: {e}") self.event_store = {} def _record_method_events(self) -> None: # only save during live try: if is_live(): file_path = self._get_test_event_file_name() keys_to_redact = ["incomingCallContext", "callerDisplayName"] redacted_dict = self.redact_by_key(self.event_to_save, keys_to_redact) with open(file_path, "w") as json_file: json.dump(redacted_dict, json_file) except Exception as e: # Don't fail the test for file I/O issues print(f"Warning: Event recording failed: {e}") def check_for_event(self, event_type: str, call_connection_id: str, wait_time: timedelta) -> Any: key = self._event_key_gen(event_type) time_out_time = datetime.now() + wait_time while datetime.now() < time_out_time: popped_event = self.event_store.pop(key, None) if popped_event is not None: print(f"Matching Event Found [{key}]") return popped_event time.sleep(1) return None async def establish_callconnection_voip( self, caller, target, *, cognitive_service_enabled: Optional[bool] = False ) -> tuple: return await self._establish_callconnection(caller, target, cognitive_service_enabled=cognitive_service_enabled) async def establish_callconnection_pstn(self, caller, target) -> tuple: return await self._establish_callconnection(caller, target, is_pstn=True) async def establish_callconnection_voip_with_streaming_options( self, caller, target, options, is_transcription ) -> tuple: return await self._establish_callconnection(caller, target, options=options, is_transcription=is_transcription) async def establish_callconnection_voip_connect_call(self, caller, target) -> tuple: return await self._establish_callconnection(caller, target, connect_call=True) async def _establish_callconnection( self, caller, target, cognitive_service_enabled: Optional[bool] = False, is_pstn: bool = False, options=None, is_transcription: bool = False, connect_call: bool = False, ) -> tuple: call_automation_client_caller = self._create_call_automation_client(caller) call_automation_client_target = self._create_call_automation_client(target) unique_id = self._unique_key_gen(caller, target) if is_live(): self._subscribe_to_dispatcher(unique_id) callback_url = f"{self.dispatcher_callback}?q={unique_id}" create_call_result = await self._create_call( call_automation_client_caller, caller, target, unique_id, cognitive_service_enabled, is_pstn, options, is_transcription, ) caller_connection_id = self._validate_create_call_result(create_call_result) incoming_call_context = self._wait_for_incoming_call(unique_id) answer_call_result = await self._answer_call(call_automation_client_target, incoming_call_context) call_connection_caller = self._create_call_connection_client(caller_connection_id) call_connection_target = self._create_call_connection_client(answer_call_result.call_connection_id) self.open_call_connections[unique_id] = call_connection_caller if connect_call: return ( unique_id, call_connection_caller, call_connection_target, call_automation_client_caller, callback_url, ) return unique_id, call_connection_caller, call_connection_target def _create_call_automation_client(self, source): return CallAutomationClient.from_connection_string(self.connection_str, source=source) def _subscribe_to_dispatcher(self, unique_id): 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() async def _create_call( self, client, caller, target, unique_id, cognitive_service_enabled, is_pstn, options, is_transcription ): if is_pstn: return await client.create_call( target_participant=target, source_caller_id_number=caller, callback_url=f"{self.dispatcher_callback}?q={unique_id}", ) elif options: return await client.create_call( target_participant=target, callback_url=f"{self.dispatcher_callback}?q={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, ) else: return await client.create_call( target_participant=target, callback_url=f"{self.dispatcher_callback}?q={unique_id}", cognitive_services_endpoint=self.cognitive_service_endpoint if cognitive_service_enabled else None, ) def _validate_create_call_result(self, result): if result is None: raise ValueError("Invalid create_call_result") caller_connection_id = result.call_connection_id if caller_connection_id is None: raise ValueError("Caller connection ID is None") return caller_connection_id def _wait_for_incoming_call(self, unique_id): 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") return incoming_call_event["incomingCallContext"] async def _answer_call(self, client, context): result = await client.answer_call(incoming_call_context=context, callback_url=self.dispatcher_callback) if result is None: raise ValueError("Invalid answer_call result") return result def _create_call_connection_client(self, connection_id): return CallConnectionClient.from_connection_string(self.connection_str, connection_id) async def terminate_call(self, unique_id) -> None: try: call_connection = self.open_call_connections.pop(unique_id, None) if call_connection is not None: await call_connection.hang_up(is_for_everyone=True) disconnected_event = self.check_for_event( "CallDisconnected", call_connection._call_connection_id, timedelta(seconds=15) ) if disconnected_event is 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 def redact_by_key(self, data: Dict[str, Dict[str, any]], keys_to_redact: List[str]) -> Dict[str, Dict[str, any]]: for _, inner_dict in data.items(): for key in keys_to_redact: if key in inner_dict: inner_dict[key] = "REDACTED" return data