Skip to content

GitLab

Commit 4157c07f authored by Ghislain MARY's avatar Ghislain MARY
Browse files

Use weakref from Core to CoreManager in the Python module tester.

parent 90ee807c
Hide whitespace changes
Inline Side-by-side
tools/python/unittests/linphonetester.py
View file @ 4157c07f
......@@ -6,6 +6,7 @@ import logging
import os
import sys
import time
import weakref
test_domain = "sipopen.example.org"
......@@ -93,13 +94,13 @@ class AccountManager:
@classmethod
def account_created_on_server_cb(cls, lc, cfg, state, message):
if state == linphone.RegistrationState.Ok:
lc.user_data.created = True
lc.user_data().created = True
elif state == linphone.RegistrationState.Cleared:
lc.user_data.done = True
lc.user_data().done = True
@classmethod
def account_created_auth_requested_cb(cls, lc, realm, username, domain):
lc.user_data.auth_requested = True
lc.user_data().auth_requested = True
def check_account(self, cfg):
create_account = False
......@@ -142,7 +143,7 @@ class AccountManager:
cfg.server_addr = server_addr.as_string()
cfg.expires = 3600
lc.add_proxy_config(cfg)
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data.auth_requested == True, 10000) != True:
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data().auth_requested == True, 10000) != True:
linphonetester_logger.critical("[TESTER] Account for {identity} could not be created on server.".format(identity=refcfg.identity))
sys.exit(-1)
cfg.edit()
......@@ -150,11 +151,11 @@ class AccountManager:
cfg.done()
ai = linphone.AuthInfo.new(account.modified_identity.username, None, account.password, None, None, account.modified_identity.domain)
lc.add_auth_info(ai)
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data.created == True, 3000) != True:
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data().created == True, 3000) != True:
linphonetester_logger.critical("[TESTER] Account for {identity} is not working on server.".format(identity=refcfg.identity))
sys.exit(-1)
lc.remove_proxy_config(cfg)
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data.done == True, 3000) != True:
if AccountManager.wait_for_until(lc, None, lambda lc: lc.user_data().done == True, 3000) != True:
linphonetester_logger.critical("[TESTER] Account creation could not clean the registration context.")
sys.exit(-1)
......@@ -287,7 +288,7 @@ class CoreManager:
lc.ring = os.path.join(resources_path, 'sounds', 'oldphone.wav')
lc.ringback = os.path.join(resources_path, 'sounds', 'ringback.wav')
lc.static_picture = os.path.join(resources_path, 'images', 'nowebcamCIF.jpg')
lc.user_data = user_data
lc.user_data = weakref.ref(user_data)
return lc
@classmethod
......@@ -389,7 +390,7 @@ class CoreManager:
@classmethod
def registration_state_changed(cls, lc, cfg, state, message):
manager = lc.user_data
manager = lc.user_data()
linphonetester_logger.info("[TESTER] New registration state {state} for user id [{identity}] at proxy [{addr}]".format(
state=linphone.RegistrationState.string(state), identity=cfg.identity, addr=cfg.server_addr))
if state == linphone.RegistrationState.None:
......@@ -407,14 +408,14 @@ class CoreManager:
@classmethod
def auth_info_requested(cls, lc, realm, username, domain):
manager = lc.user_data
manager = lc.user_data()
linphonetester_logger.info("[TESTER] Auth info requested for user id [{username}] at realm [{realm}]".format(
username=username, realm=realm))
manager.stats.number_of_auth_info_requested +=1
@classmethod
def call_state_changed(cls, lc, call, state, msg):
manager = lc.user_data
manager = lc.user_data()
to_address = call.call_log.to_address.as_string()
from_address = call.call_log.from_address.as_string()
direction = "Outgoing"
......@@ -463,7 +464,7 @@ class CoreManager:
@classmethod
def message_received(cls, lc, room, message):
manager = lc.user_data
manager = lc.user_data()
from_str = message.from_address.as_string()
text_str = message.text
external_body_url = message.external_body_url
......@@ -478,7 +479,7 @@ class CoreManager:
@classmethod
def new_subscription_requested(cls, lc, lf, url):
manager = lc.user_data
manager = lc.user_data()
linphonetester_logger.info("[TESTER] New subscription request: from [{from_str}], url [{url}]".format(
from_str=lf.address.as_string(), url=url))
manager.stats.number_of_NewSubscriptionRequest += 1
......@@ -486,7 +487,7 @@ class CoreManager:
@classmethod
def notify_presence_received(cls, lc, lf):
manager = lc.user_data
manager = lc.user_data()
linphonetester_logger.info("[TESTER] New notify request: from [{from_str}]".format(
from_str=lf.address.as_string()))
manager.stats.number_of_NotifyReceived += 1
......@@ -600,9 +601,6 @@ class CoreManager:
self.identity = linphone.Address.new(self.lc.default_proxy_config.identity)
self.identity.clean()
def stop(self):
self.lc = None
def enable_audio_codec(self, mime, rate):
codecs = self.lc.audio_codecs
for codec in codecs:
......
tools/python/unittests/test_call.py
View file @ 4157c07f
......@@ -22,8 +22,6 @@ class TestCall:
out_call_log = pauline.lc.call_logs[0]
assert out_call_log is not None
assert_equals(out_call_log.status, linphone.CallStatus.Aborted)
marie.stop()
pauline.stop()
def test_declined_call(self):
marie = CoreManager('marie_rc')
......@@ -40,8 +38,6 @@ class TestCall:
assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1)
assert_equals(in_call.reason, linphone.Reason.Declined)
assert_equals(out_call.reason, linphone.Reason.Declined)
marie.stop()
pauline.stop()
def test_cancelled_call(self):
marie = CoreManager('marie_rc')
......@@ -53,8 +49,6 @@ class TestCall:
assert_equals(pauline.stats.number_of_LinphoneCallEnd, 1)
assert_equals(marie.stats.number_of_LinphoneCallIncomingReceived, 0)
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 = CoreManager('marie_rc')
......@@ -72,8 +66,6 @@ class TestCall:
# Now the CANCEL should have been sent and the the call at marie's side should terminate
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 = CoreManager('marie_rc')
......@@ -84,8 +76,6 @@ class TestCall:
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 = CoreManager('marie_rc')
......@@ -100,8 +90,6 @@ class TestCall:
assert_equals(out_call.reason, linphone.Reason.NotAcceptable)
assert_equals(marie.stats.number_of_LinphoneCallIncomingReceived, 0)
assert_equals(marie.stats.number_of_LinphoneCallReleased, 0)
marie.stop()
pauline.stop()
def test_simple_call(self):
marie = CoreManager('marie_rc')
......@@ -109,5 +97,3 @@ class TestCall:
assert_equals(CoreManager.call(pauline, marie), True)
#liblinphone_tester_check_rtcp(marie,pauline);
CoreManager.end_call(marie, pauline)
marie.stop()
pauline.stop()
tools/python/unittests/test_message.py
View file @ 4157c07f
......@@ -12,7 +12,7 @@ class TestMessage:
@classmethod
def msg_state_changed(cls, msg, state):
stats = msg.chat_room.core.user_data.stats
stats = msg.chat_room.core.user_data().stats
linphonetester_logger.info("[TESTER] Message [{text}] [{state}]".format(text=msg.text, state=linphone.ChatMessageState.string(state)))
if state == linphone.ChatMessageState.Delivered:
stats.number_of_LinphoneMessageDelivered += 1
......@@ -27,7 +27,7 @@ class TestMessage:
@classmethod
def file_transfer_progress_indication(cls, msg, content, offset, total):
stats = msg.chat_room.core.user_data.stats
stats = msg.chat_room.core.user_data().stats
progress = int((offset * 100) / total)
direction = 'received'
tofrom = 'from'
......@@ -57,7 +57,7 @@ class TestMessage:
@classmethod
def file_transfer_recv(cls, msg, content, buf):
receive_filepath = msg.user_data
stats = msg.chat_room.core.user_data.stats
stats = msg.chat_room.core.user_data().stats
if buf.empty: # Transfer complete
stats.number_of_LinphoneMessageExtBodyReceived += 1
else: # Store content
......@@ -77,7 +77,7 @@ class TestMessage:
@classmethod
def memory_file_transfer_recv(cls, msg, content, buf):
stats = msg.chat_room.core.user_data.stats
stats = msg.chat_room.core.user_data().stats
if buf.empty: # Transfer complete
stats.number_of_LinphoneMessageExtBodyReceived += 1
else: # Store content
......@@ -98,8 +98,6 @@ class TestMessage:
chat_room.send_chat_message(msg)
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')
......@@ -112,8 +110,6 @@ class TestMessage:
chat_room.send_chat_message(msg)
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_file_transfer_message(self):
marie = CoreManager('marie_rc')
......@@ -147,8 +143,6 @@ class TestMessage:
assert_equals(filecmp.cmp(send_filepath, receive_filepath, shallow=False), True)
if os.path.exists(receive_filepath):
os.remove(receive_filepath)
marie.stop()
pauline.stop()
def test_small_file_transfer_message(self):
send_buf = "small file"
......@@ -185,8 +179,6 @@ class TestMessage:
assert_equals(pauline.stats.number_of_LinphoneMessageDelivered, 1)
assert_equals(marie.stats.number_of_LinphoneMessageExtBodyReceived, 1)
assert_equals(send_buf, marie.stats.last_received_chat_message.user_data)
marie.stop()
pauline.stop()
def test_file_transfer_message_upload_cancelled(self):
send_buf = "big file"
......@@ -218,5 +210,3 @@ class TestMessage:
assert_equals(CoreManager.wait_for(pauline, marie, lambda pauline, marie: pauline.stats.number_of_LinphoneMessageNotDelivered == 1), True)
assert_equals(pauline.stats.number_of_LinphoneMessageNotDelivered, 1)
assert_equals(marie.stats.number_of_LinphoneMessageExtBodyReceived, 0)
marie.stop()
pauline.stop()
tools/python/unittests/test_presence.py
View file @ 4157c07f
......@@ -35,5 +35,3 @@ class TestPresence:
marie = PresenceCoreManager('marie')
pauline = PresenceCoreManager('pauline')
assert_equals(self.subscribe_to_callee_presence(marie, pauline), True)
marie.stop()
pauline.stop()
tools/python/unittests/test_register.py
View file @ 4157c07f
......@@ -70,7 +70,6 @@ class RegisterCoreManager(CoreManager):
def register_with_refresh(self, refresh, domain, route, late_auth_info = False, transport = linphone.SipTransports(5070, 5070, 5071, 0), expected_final_state = linphone.RegistrationState.Ok):
self.register_with_refresh_base(refresh, domain, route, late_auth_info, expected_final_state = expected_final_state)
self.stop()
# Not testable as the callbacks can not be called once the core destruction has started
#assert_equals(self.stats.number_of_LinphoneRegistrationCleared, 1)
......@@ -81,7 +80,6 @@ class TestRegister:
cm = RegisterCoreManager()
cm.register_with_refresh(False, None, None)
assert_equals(cm.stats.number_of_auth_info_requested, 0)
cm.stop()
def test_simple_unregister(self):
cm = RegisterCoreManager()
......@@ -94,22 +92,18 @@ class TestRegister:
pc.register_enabled = False
pc.done()
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationCleared == 1), True)
cm.stop()
def test_simple_tcp_register(self):
cm = RegisterCoreManager()
cm.register_with_refresh(False, test_domain, "sip:{route};transport=tcp".format(route=test_route))
cm.stop()
def test_simple_tcp_register_compatibility_mode(self):
cm = RegisterCoreManager()
cm.register_with_refresh(False, test_domain, "sip:{route}".format(route=test_route), transport=linphone.SipTransports(0, 5070, 0, 0))
cm.stop()
def test_simple_tls_register(self):
cm = RegisterCoreManager()
cm.register_with_refresh(False, test_domain, "sip:{route};transport=tls".format(route=test_route))
cm.stop()
def test_tls_register_with_alt_name(self):
cm = CoreManager('pauline_alt_rc', False)
......@@ -117,7 +111,6 @@ class TestRegister:
cm.lc.refresh_registers()
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 1), True)
assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 0)
cm.stop()
def test_tls_wildcard_register(self):
cm = CoreManager('pauline_wild_rc', False)
......@@ -125,7 +118,6 @@ class TestRegister:
cm.lc.refresh_registers()
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 2), True)
assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 0)
cm.stop()
def test_tls_certificate_failure(self):
cm = CoreManager('pauline_rc', False)
......@@ -139,7 +131,6 @@ class TestRegister:
cm.lc.refresh_registers()
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == 1), True)
assert_equals(cm.stats.number_of_LinphoneRegistrationFailed, 2)
cm.stop()
def test_tls_with_non_tls_server(self):
cm = CoreManager('marie_rc', False)
......@@ -153,7 +144,6 @@ class TestRegister:
pc.server_addr = "sip:{domain}:{port};transport=tls".format(domain=addr.domain, port=port)
pc.done()
assert_equals(CoreManager.wait_for_until(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == 1, 5000), True)
cm.stop()
def test_simple_authenticated_register(self):
cm = RegisterCoreManager()
......@@ -161,35 +151,29 @@ class TestRegister:
cm.lc.add_auth_info(info)
cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route))
assert_equals(cm.stats.number_of_auth_info_requested, 0)
cm.stop()
def test_digest_auth_without_initial_credentials(self):
cm = RegisterCoreManager(with_auth=True)
cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route))
assert_equals(cm.stats.number_of_auth_info_requested, 1)
cm.stop()
def test_authenticated_register_with_late_credentials(self):
cm = RegisterCoreManager()
cm.register_with_refresh(False, auth_domain, "sip:{route}".format(route=test_route), True, linphone.SipTransports(5070, 5070, 5071, 0))
assert_equals(cm.stats.number_of_auth_info_requested, 1)
cm.stop()
def test_simple_register_with_refresh(self):
cm = RegisterCoreManager()
cm.register_with_refresh(True, None, None)
assert_equals(cm.stats.number_of_auth_info_requested, 0)
cm.stop()
def test_simple_auth_register_with_refresh(self):
cm = RegisterCoreManager(with_auth=True)
cm.register_with_refresh(True, auth_domain, "sip:{route}".format(route=test_route))
assert_equals(cm.stats.number_of_auth_info_requested, 1)
cm.stop()
def test_multiple_accounts(self):
cm = CoreManager('multi_account_rc', True)
cm.stop()
CoreManager('multi_account_rc', True)
def test_transport_change(self):
cm = CoreManager('multi_account_rc', True)
......@@ -202,4 +186,3 @@ class TestRegister:
cm.lc.sip_transports = tr
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationOk == (register_ok + number_of_udp_proxies)), True)
assert_equals(CoreManager.wait_for(cm, cm, lambda cm1, cm2: cm1.stats.number_of_LinphoneRegistrationFailed == (total_number_of_proxies - number_of_udp_proxies)), True)
cm.stop()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment