diff --git a/tools/python/unittests/linphonetester.py b/tools/python/unittests/linphonetester.py index f60d943a4..14c1dbde5 100644 --- a/tools/python/unittests/linphonetester.py +++ b/tools/python/unittests/linphonetester.py @@ -1,5 +1,6 @@ from datetime import timedelta, datetime from nose.tools import assert_equals +from copy import deepcopy import linphone import logging import os @@ -180,6 +181,78 @@ class CoreManager: def wait_for(cls, manager1, manager2, func): return cls.wait_for_until(manager1, manager2, func, 10000) + @classmethod + def call(cls, caller_manager, callee_manager, caller_params = None, callee_params = None, build_callee_params = False): + initial_caller_stats = deepcopy(caller_manager.stats) + initial_callee_stats = deepcopy(callee_manager.stats) + + # Use playfile for callee to avoid locking on capture card + callee_manager.lc.use_files = True + callee_manager.lc.play_file = os.path.join(tester_resources_path, 'sounds', 'hello8000.wav') + + if caller_params is None: + call = caller_manager.lc.invite_address(callee_manager.identity) + else: + call = caller_manager.lc.invite_address_with_params(callee_manager.identity, caller_params) + assert call is not None + + assert_equals(CoreManager.wait_for(callee_manager, caller_manager, + lambda callee_manager, caller_manager: callee_manager.stats.number_of_LinphoneCallIncomingReceived == initial_callee_stats.number_of_LinphoneCallIncomingReceived + 1), True) + assert_equals(callee_manager.lc.incoming_invite_pending, True) + assert_equals(caller_manager.stats.number_of_LinphoneCallOutgoingProgress, initial_caller_stats.number_of_LinphoneCallOutgoingProgress + 1) + + retry = 0 + while (caller_manager.stats.number_of_LinphoneCallOutgoingRinging != initial_caller_stats.number_of_LinphoneCallOutgoingRinging + 1) and \ + (caller_manager.stats.number_of_LinphoneCallOutgoingEarlyMedia != initial_caller_stats.number_of_LinphoneCallOutgoingEarlyMedia + 1) and \ + retry < 20: + retry += 1 + caller_manager.lc.iterate() + callee_manager.lc.iterate() + time.sleep(0.1) + assert ((caller_manager.stats.number_of_LinphoneCallOutgoingRinging == initial_caller_stats.number_of_LinphoneCallOutgoingRinging + 1) or \ + (caller_manager.stats.number_of_LinphoneCallOutgoingEarlyMedia == initial_caller_stats.number_of_LinphoneCallOutgoingEarlyMedia + 1)) == True + + assert callee_manager.lc.current_call_remote_address is not None + if caller_manager.lc.current_call is None or callee_manager.lc.current_call is None or callee_manager.lc.current_call_remote_address is None: + return False + callee_from_address = caller_manager.identity.clone() + callee_from_address.port = 0 # Remove port because port is never present in from header + assert_equals(callee_from_address.weak_equal(callee_manager.lc.current_call_remote_address), True) + + if callee_params is not None: + callee_manager.lc.accept_call_with_params(callee_manager.lc.current_call, callee_params) + elif build_callee_params: + default_params = callee_manager.lc.create_call_params(callee_manager.lc.current_call) + callee_manager.lc.accept_call_with_params(callee_manager.lc.current_call, default_params) + else: + callee_manager.lc.accept_call(callee_manager.lc.current_call) + assert_equals(CoreManager.wait_for(callee_manager, caller_manager, + lambda callee_manager, caller_manager: (callee_manager.stats.number_of_LinphoneCallConnected == initial_callee_stats.number_of_LinphoneCallConnected + 1) and \ + (caller_manager.stats.number_of_LinphoneCallConnected == initial_caller_stats.number_of_LinphoneCallConnected + 1)), True) + # Just to sleep + result = CoreManager.wait_for(callee_manager, caller_manager, + lambda callee_manager, caller_manager: (callee_manager.stats.number_of_LinphoneCallStreamsRunning == initial_callee_stats.number_of_LinphoneCallStreamsRunning + 1) and \ + (caller_manager.stats.number_of_LinphoneCallStreamsRunning == initial_caller_stats.number_of_LinphoneCallStreamsRunning + 1)) + + if caller_manager.lc.media_encryption != linphone.MediaEncryption.MediaEncryptionNone and callee_manager.lc.media_encryption != linphone.MediaEncryption.MediaEncryptionNone: + # Wait for encryption to be on, in case of zrtp, it can take a few seconds + if caller_manager.lc.media_encryption == linphone.MediaEncryption.MediaEncryptionZRTP: + CoreManager.wait_for(callee_manager, caller_manager, + lambda callee_manager, caller_manager: caller_manager.stats.number_of_LinphoneCallEncryptedOn == initial_caller_stats.number_of_LinphoneCallEncryptedOn + 1) + if callee_manager.lc.media_encryption == linphone.MediaEncryption.MediaEncryptionZRTP: + CoreManager.wait_for(callee_manager, caller_manager, + lambda callee_manager, caller_manager: callee_manager.stats.number_of_LinphoneCallEncryptedOn == initial_callee_stats.number_of_LinphoneCallEncryptedOn + 1) + assert_equals(callee_manager.lc.current_call.current_params.media_encryption, caller_manager.lc.media_encryption) + assert_equals(caller_manager.lc.current_call.current_params.media_encryption, callee_manager.lc.media_encryption) + + return result + + @classmethod + def end_call(cls, caller_manager, callee_manager): + caller_manager.lc.terminate_all_calls() + assert_equals(CoreManager.wait_for(caller_manager, callee_manager, + lambda caller_manager, callee_manager: caller_manager.stats.number_of_LinphoneCallEnd == 1 and callee_manager.stats.number_of_LinphoneCallEnd == 1), True) + @classmethod def registration_state_changed(cls, lc, cfg, state, message): manager = lc.user_data @@ -257,6 +330,20 @@ class CoreManager: else: raise Exception("Unexpected call state") + @classmethod + def message_received(cls, lc, room, message): + manager = lc.user_data + from_str = message.from_address.as_string() + text_str = message.text + external_body_url = message.external_body_url + if manager.logger is not None: + manager.logger.info("[TESTER] Message from [{from_str}] is [{text_str}], external URL [{external_body_url}]".format( + from_str=from_str, text_str=text_str, external_body_url=external_body_url)) + manager.stats.number_of_LinphoneMessageReceived += 1 + + if message.external_body_url is not None: + manager.stats.number_of_LinphoneMessageExtBodyReceived += 1 + def __init__(self, rc_file = None, check_for_proxies = True, vtable = {}, logger=None): self.logger = logger if not vtable.has_key('registration_state_changed'): @@ -265,10 +352,8 @@ class CoreManager: vtable['auth_info_requested'] = CoreManager.auth_info_requested if not vtable.has_key('call_state_changed'): vtable['call_state_changed'] = CoreManager.call_state_changed - #if not vtable.has_key('text_received'): - #vtable['text_received'] = CoreManager.text_received - #if not vtable.has_key('message_received'): - #vtable['message_received'] = CoreManager.message_received + if not vtable.has_key('message_received'): + vtable['message_received'] = CoreManager.message_received #if not vtable.has_key('file_transfer_recv'): #vtable['file_transfer_recv'] = CoreManager.file_transfer_recv #if not vtable.has_key('file_transfer_send'): diff --git a/tools/python/unittests/test_call.py b/tools/python/unittests/test_call.py index 6d7ca0f4a..93dfd1175 100644 --- a/tools/python/unittests/test_call.py +++ b/tools/python/unittests/test_call.py @@ -1,7 +1,6 @@ from nose.tools import assert_equals -from copy import deepcopy import linphone -import linphonetester +from linphonetester import * import os import time @@ -11,86 +10,16 @@ class TestCall: @classmethod def setup_class(cls): base, ext = os.path.splitext(os.path.basename(__file__)) - cls.logger = linphonetester.Logger(base + '.log') - - def call(self, caller_manager, callee_manager, caller_params = None, callee_params = None, build_callee_params = False): - initial_caller_stats = deepcopy(caller_manager.stats) - initial_callee_stats = deepcopy(callee_manager.stats) - - # Use playfile for callee to avoid locking on capture card - callee_manager.lc.use_files = True - callee_manager.lc.play_file = os.path.join(linphonetester.tester_resources_path, 'sounds', 'hello8000.wav') - - if caller_params is None: - call = caller_manager.lc.invite_address(callee_manager.identity) - else: - call = caller_manager.lc.invite_address_with_params(callee_manager.identity, caller_params) - assert call is not None - - assert_equals(linphonetester.CoreManager.wait_for(callee_manager, caller_manager, - lambda callee_manager, caller_manager: callee_manager.stats.number_of_LinphoneCallIncomingReceived == initial_callee_stats.number_of_LinphoneCallIncomingReceived + 1), True) - assert_equals(callee_manager.lc.incoming_invite_pending, True) - assert_equals(caller_manager.stats.number_of_LinphoneCallOutgoingProgress, initial_caller_stats.number_of_LinphoneCallOutgoingProgress + 1) - - retry = 0 - while (caller_manager.stats.number_of_LinphoneCallOutgoingRinging != initial_caller_stats.number_of_LinphoneCallOutgoingRinging + 1) and \ - (caller_manager.stats.number_of_LinphoneCallOutgoingEarlyMedia != initial_caller_stats.number_of_LinphoneCallOutgoingEarlyMedia + 1) and \ - retry < 20: - retry += 1 - caller_manager.lc.iterate() - callee_manager.lc.iterate() - time.sleep(0.1) - assert ((caller_manager.stats.number_of_LinphoneCallOutgoingRinging == initial_caller_stats.number_of_LinphoneCallOutgoingRinging + 1) or \ - (caller_manager.stats.number_of_LinphoneCallOutgoingEarlyMedia == initial_caller_stats.number_of_LinphoneCallOutgoingEarlyMedia + 1)) == True - - assert callee_manager.lc.current_call_remote_address is not None - if caller_manager.lc.current_call is None or callee_manager.lc.current_call is None or callee_manager.lc.current_call_remote_address is None: - return False - callee_from_address = caller_manager.identity.clone() - callee_from_address.port = 0 # Remove port because port is never present in from header - assert_equals(callee_from_address.weak_equal(callee_manager.lc.current_call_remote_address), True) - - if callee_params is not None: - callee_manager.lc.accept_call_with_params(callee_manager.lc.current_call, callee_params) - elif build_callee_params: - default_params = callee_manager.lc.create_call_params(callee_manager.lc.current_call) - callee_manager.lc.accept_call_with_params(callee_manager.lc.current_call, default_params) - else: - callee_manager.lc.accept_call(callee_manager.lc.current_call) - assert_equals(linphonetester.CoreManager.wait_for(callee_manager, caller_manager, - lambda callee_manager, caller_manager: (callee_manager.stats.number_of_LinphoneCallConnected == initial_callee_stats.number_of_LinphoneCallConnected + 1) and \ - (caller_manager.stats.number_of_LinphoneCallConnected == initial_caller_stats.number_of_LinphoneCallConnected + 1)), True) - # Just to sleep - result = linphonetester.CoreManager.wait_for(callee_manager, caller_manager, - lambda callee_manager, caller_manager: (callee_manager.stats.number_of_LinphoneCallStreamsRunning == initial_callee_stats.number_of_LinphoneCallStreamsRunning + 1) and \ - (caller_manager.stats.number_of_LinphoneCallStreamsRunning == initial_caller_stats.number_of_LinphoneCallStreamsRunning + 1)) - - if caller_manager.lc.media_encryption != linphone.MediaEncryption.MediaEncryptionNone and callee_manager.lc.media_encryption != linphone.MediaEncryption.MediaEncryptionNone: - # Wait for encryption to be on, in case of zrtp, it can take a few seconds - if caller_manager.lc.media_encryption == linphone.MediaEncryption.MediaEncryptionZRTP: - linphonetester.CoreManager.wait_for(callee_manager, caller_manager, - lambda callee_manager, caller_manager: caller_manager.stats.number_of_LinphoneCallEncryptedOn == initial_caller_stats.number_of_LinphoneCallEncryptedOn + 1) - if callee_manager.lc.media_encryption == linphone.MediaEncryption.MediaEncryptionZRTP: - linphonetester.CoreManager.wait_for(callee_manager, caller_manager, - lambda callee_manager, caller_manager: callee_manager.stats.number_of_LinphoneCallEncryptedOn == initial_callee_stats.number_of_LinphoneCallEncryptedOn + 1) - assert_equals(callee_manager.lc.current_call.current_params.media_encryption, caller_manager.lc.media_encryption) - assert_equals(caller_manager.lc.current_call.current_params.media_encryption, callee_manager.lc.media_encryption) - - return result - - def end_call(self, caller_manager, callee_manager): - caller_manager.lc.terminate_all_calls() - assert_equals(linphonetester.CoreManager.wait_for(caller_manager, callee_manager, - lambda caller_manager, callee_manager: caller_manager.stats.number_of_LinphoneCallEnd == 1 and callee_manager.stats.number_of_LinphoneCallEnd == 1), True) + cls.logger = Logger(base + '.log') def test_early_declined_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) marie.lc.max_calls = 0 out_call = pauline.lc.invite('marie') # Wait until flexisip transfers the busy... - assert_equals(linphonetester.CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallError == 1, 33000), True) + assert_equals(CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallError == 1, 33000), True) assert_equals(pauline.stats.number_of_LinphoneCallError, 1) assert_equals(out_call.reason, linphone.Reason.ReasonBusy) if len(pauline.lc.call_logs) > 0: @@ -101,16 +30,16 @@ class TestCall: pauline.stop() def test_declined_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) out_call = pauline.lc.invite_address(marie.identity) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) in_call = marie.lc.current_call assert in_call is not None if in_call is not None: marie.lc.terminate_call(in_call) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallReleased == 1), True) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallReleased == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) assert_equals(marie.stats.number_of_LinphoneCallEnd, 1) assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1) assert_equals(in_call.reason, linphone.Reason.ReasonDeclined) @@ -119,69 +48,69 @@ class TestCall: pauline.stop() def test_cancelled_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) out_call = pauline.lc.invite('marie') - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) pauline.lc.terminate_call(out_call) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallEnd == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallEnd == 1), True) assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1) assert_equals(marie.stats.number_of_LinphoneCallIncomingReceived, 0) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) marie.stop() pauline.stop() def test_early_cancelled_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('empty_rc', check_for_proxies=False, logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('empty_rc', check_for_proxies=False, logger=TestCall.logger) out_call = pauline.lc.invite_address(marie.identity) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) pauline.lc.terminate_call(out_call) # Since everything is executed in a row, no response can be received from the server, thus the CANCEL cannot be sent. # It will ring at Marie's side. - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallEnd == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallEnd == 1), True) assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) # Now the CANCEL should have been sent and the the call at marie's side should terminate - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallEnd == 1), True) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallEnd == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallReleased == 1), True) marie.stop() pauline.stop() def test_cancelled_ringing_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) out_call = pauline.lc.invite('marie') - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneCallIncomingReceived == 1), True) pauline.lc.terminate_call(out_call) - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: (pauline.stats.number_of_LinphoneCallReleased == 1) and (marie.stats.number_of_LinphoneCallReleased == 1)), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: (pauline.stats.number_of_LinphoneCallReleased == 1) and (marie.stats.number_of_LinphoneCallReleased == 1)), True) assert_equals(marie.stats.number_of_LinphoneCallEnd, 1) assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1) marie.stop() pauline.stop() def test_call_failed_because_of_codecs(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) marie.disable_all_audio_codecs_except_one('pcmu') pauline.disable_all_audio_codecs_except_one('pcma') out_call = pauline.lc.invite('marie') - assert_equals(linphonetester.CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallOutgoingInit == 1), True) # flexisip will retain the 488 until the "urgent reply" timeout arrives. - assert_equals(linphonetester.CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallError == 1, 6000), True) + assert_equals(CoreManager.wait_for_until(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneCallError == 1, 6000), True) assert_equals(out_call.reason, linphone.Reason.ReasonNotAcceptable) assert_equals(marie.stats.number_of_LinphoneCallIncomingReceived, 0) marie.stop() pauline.stop() def test_simple_call(self): - marie = linphonetester.CoreManager('marie_rc', logger=TestCall.logger) - pauline = linphonetester.CoreManager('pauline_rc', logger=TestCall.logger) - assert_equals(self.call(pauline, marie), True) + marie = CoreManager('marie_rc', logger=TestCall.logger) + pauline = CoreManager('pauline_rc', logger=TestCall.logger) + assert_equals(CoreManager.call(pauline, marie), True) #liblinphone_tester_check_rtcp(marie,pauline); - self.end_call(marie, pauline) + CoreManager.end_call(marie, pauline) marie.stop() pauline.stop() diff --git a/tools/python/unittests/test_message.py b/tools/python/unittests/test_message.py new file mode 100644 index 000000000..56a7e9e17 --- /dev/null +++ b/tools/python/unittests/test_message.py @@ -0,0 +1,46 @@ +from nose.tools import assert_equals +from copy import deepcopy +import linphone +from linphonetester import * +import os +import time + + +class TestMessage: + + @classmethod + def setup_class(cls): + base, ext = os.path.splitext(os.path.basename(__file__)) + cls.logger = Logger(base + '.log') + + def wait_for_server_to_purge_messages(self, manager1, manager2): + # Wait a little bit just to have time to purge message stored in the server + CoreManager.wait_for_until(manager1, manager2, lambda manager1, manager2: False, 100) + manager1.stats.reset() + manager2.stats.reset() + + def test_text_message(self): + marie = CoreManager('marie_rc', logger=TestMessage.logger) + pauline = CoreManager('pauline_rc', logger=TestMessage.logger) + chat_room = pauline.lc.get_chat_room(marie.identity) + self.wait_for_server_to_purge_messages(marie, pauline) + msg = chat_room.create_message("Bla bla bla bla") + chat_room.send_message2(msg, None, None) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneMessageReceived == 1), True) + assert marie.lc.get_chat_room(pauline.identity) is not None + marie.stop() + pauline.stop() + + def test_text_message_within_dialog(self): + marie = CoreManager('marie_rc', logger=TestMessage.logger) + pauline = CoreManager('pauline_rc', logger=TestMessage.logger) + pauline.lc.config.set_int('sip', 'chat_use_call_dialogs', 1) + chat_room = pauline.lc.get_chat_room(marie.identity) + self.wait_for_server_to_purge_messages(marie, pauline) + assert_equals(CoreManager.call(marie, pauline), True) + msg = chat_room.create_message("Bla bla bla bla") + chat_room.send_message2(msg, None, None) + assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: marie.stats.number_of_LinphoneMessageReceived == 1), True) + assert marie.lc.get_chat_room(pauline.identity) is not None + marie.stop() + pauline.stop() diff --git a/tools/python/unittests/test_register.py b/tools/python/unittests/test_register.py index ba34a42a2..ac81f6f01 100644 --- a/tools/python/unittests/test_register.py +++ b/tools/python/unittests/test_register.py @@ -1,14 +1,14 @@ from nose.tools import assert_equals import linphone -import linphonetester +from linphonetester import * import os import time -class RegisterCoreManager(linphonetester.CoreManager): +class RegisterCoreManager(CoreManager): @classmethod def auth_info_requested(cls, lc, realm, username, domain): - linphonetester.CoreManager.auth_info_requested(cls, lc, realm, username, domain) + CoreManager.auth_info_requested(cls, lc, realm, username, domain) info = linphone.AuthInfo.new(test_username, None, test_password, None, realm, domain) # Create authentication structure from identity lc.add_auth_info(info) # Add authentication info to LinphoneCore @@ -16,13 +16,13 @@ class RegisterCoreManager(linphonetester.CoreManager): vtable = {} if with_auth: vtable['auth_info_requested'] = RegisterCoreManager.auth_info_requested - linphonetester.CoreManager.__init__(self, vtable=vtable) + CoreManager.__init__(self, vtable=vtable) def register_with_refresh(self, refresh, domain, route, late_auth_info = False, expected_final_state = linphone.RegistrationState.RegistrationOk): assert self.lc is not None self.stats.reset() proxy_cfg = self.lc.create_proxy_config() - from_address = linphonetester.create_address(domain) + from_address = create_address(domain) proxy_cfg.identity = from_address.as_string() server_addr = from_address.domain proxy_cfg.register_enabled = True @@ -50,7 +50,7 @@ class RegisterCoreManager(linphonetester.CoreManager): if self.stats.number_of_auth_info_requested > 0 and proxy_cfg.state == linphone.RegistrationState.RegistrationFailed and late_auth_info: if len(self.lc.auth_info_list) == 0: assert_equals(proxy_cfg.error, linphone.Reason.ReasonUnauthorized) - info = linphone.AuthInfo.new(linphonetester.test_username, None, linphonetester.test_password, None, None, None) # Create authentication structure from identity + info = linphone.AuthInfo.new(test_username, None, test_password, None, None, None) # Create authentication structure from identity self.lc.add_auth_info(info) if proxy_cfg.error == linphone.Reason.ReasonForbidden or \ (self.stats.number_of_auth_info_requested > 2 and proxy_cfg.error == linphone.Reason.ReasonUnauthorized): @@ -78,7 +78,7 @@ class TestRegister: @classmethod def setup_class(cls): base, ext = os.path.splitext(os.path.basename(__file__)) - cls.logger = linphonetester.Logger(base + '.log') + cls.logger = Logger(base + '.log') def test_simple_register(self): cm = RegisterCoreManager() diff --git a/tools/python/unittests/test_setup.py b/tools/python/unittests/test_setup.py index d0a7c6b73..ff710217f 100644 --- a/tools/python/unittests/test_setup.py +++ b/tools/python/unittests/test_setup.py @@ -1,6 +1,6 @@ from nose.tools import assert_equals import linphone -import linphonetester +from linphonetester import * import os class TestSetup: @@ -8,10 +8,10 @@ class TestSetup: @classmethod def setup_class(cls): base, ext = os.path.splitext(os.path.basename(__file__)) - cls.logger = linphonetester.Logger(base + '.log') + cls.logger = Logger(base + '.log') def test_address(self): - linphonetester.create_address(None) + create_address(None) def test_core_init(self): lc = linphone.Core.new({}, None, None)