Commit 5e23b563 authored by Ghislain MARY's avatar Ghislain MARY
Browse files

Start adding registration unit tests for the Python wrapper.

parent aa05370d
from datetime import timedelta, datetime
from nose.tools import assert_equals
import linphone
import logging
import os
import time
test_username = "liblinphone_tester"
test_password = "secret"
test_route = "sip2.linphone.org"
def log_handler(level, msg):
method = getattr(logging, level)
if not msg.strip().startswith('[PYLINPHONE]'):
msg = '[CORE] ' + msg
method(msg)
def setup_logging(filename):
format = "%(asctime)s.%(msecs)03d %(levelname)s: %(message)s"
datefmt = "%H:%M:%S"
logging.basicConfig(filename=filename, level=logging.INFO, format=format, datefmt=datefmt)
linphone.set_log_handler(log_handler)
def create_address(domain):
addr = linphone.Address.new(None)
assert addr != None
addr.username = test_username
assert_equals(addr.username, test_username)
if domain is None:
domain = test_route
addr.domain = domain
assert_equals(addr.domain, domain)
addr.display_name = None
addr.display_name = "Mr Tester"
assert_equals(addr.display_name, "Mr Tester")
return addr
class CoreManagerStats:
def __init__(self):
self.reset()
def reset(self):
self.number_of_LinphoneRegistrationNone = 0
self.number_of_LinphoneRegistrationProgress = 0
self.number_of_LinphoneRegistrationOk = 0
self.number_of_LinphoneRegistrationCleared = 0
self.number_of_LinphoneRegistrationFailed = 0
self.number_of_auth_info_requested = 0
self.number_of_LinphoneCallIncomingReceived = 0
self.number_of_LinphoneCallOutgoingInit = 0
self.number_of_LinphoneCallOutgoingProgress = 0
self.number_of_LinphoneCallOutgoingRinging = 0
self.number_of_LinphoneCallOutgoingEarlyMedia = 0
self.number_of_LinphoneCallConnected = 0
self.number_of_LinphoneCallStreamsRunning = 0
self.number_of_LinphoneCallPausing = 0
self.number_of_LinphoneCallPaused = 0
self.number_of_LinphoneCallResuming = 0
self.number_of_LinphoneCallRefered = 0
self.number_of_LinphoneCallError = 0
self.number_of_LinphoneCallEnd = 0
self.number_of_LinphoneCallPausedByRemote = 0
self.number_of_LinphoneCallUpdatedByRemote = 0
self.number_of_LinphoneCallIncomingEarlyMedia = 0
self.number_of_LinphoneCallUpdating = 0
self.number_of_LinphoneCallReleased = 0
self.number_of_LinphoneTransferCallOutgoingInit = 0
self.number_of_LinphoneTransferCallOutgoingProgress = 0
self.number_of_LinphoneTransferCallOutgoingRinging = 0
self.number_of_LinphoneTransferCallOutgoingEarlyMedia = 0
self.number_of_LinphoneTransferCallConnected = 0
self.number_of_LinphoneTransferCallStreamsRunning = 0
self.number_of_LinphoneTransferCallError = 0
self.number_of_LinphoneMessageReceived = 0
self.number_of_LinphoneMessageReceivedWithFile = 0
self.number_of_LinphoneMessageReceivedLegacy = 0
self.number_of_LinphoneMessageExtBodyReceived = 0
self.number_of_LinphoneMessageInProgress = 0
self.number_of_LinphoneMessageDelivered = 0
self.number_of_LinphoneMessageNotDelivered = 0
self.number_of_LinphoneIsComposingActiveReceived = 0
self.number_of_LinphoneIsComposingIdleReceived = 0
self.progress_of_LinphoneFileTransfer = 0
self.number_of_IframeDecoded = 0
self.number_of_NewSubscriptionRequest =0
self.number_of_NotifyReceived = 0
self.number_of_LinphonePresenceActivityOffline = 0
self.number_of_LinphonePresenceActivityOnline = 0
self.number_of_LinphonePresenceActivityAppointment = 0
self.number_of_LinphonePresenceActivityAway = 0
self.number_of_LinphonePresenceActivityBreakfast = 0
self.number_of_LinphonePresenceActivityBusy = 0
self.number_of_LinphonePresenceActivityDinner = 0
self.number_of_LinphonePresenceActivityHoliday = 0
self.number_of_LinphonePresenceActivityInTransit = 0
self.number_of_LinphonePresenceActivityLookingForWork = 0
self.number_of_LinphonePresenceActivityLunch = 0
self.number_of_LinphonePresenceActivityMeal = 0
self.number_of_LinphonePresenceActivityMeeting = 0
self.number_of_LinphonePresenceActivityOnThePhone = 0
self.number_of_LinphonePresenceActivityOther = 0
self.number_of_LinphonePresenceActivityPerformance = 0
self.number_of_LinphonePresenceActivityPermanentAbsence = 0
self.number_of_LinphonePresenceActivityPlaying = 0
self.number_of_LinphonePresenceActivityPresentation = 0
self.number_of_LinphonePresenceActivityShopping = 0
self.number_of_LinphonePresenceActivitySleeping = 0
self.number_of_LinphonePresenceActivitySpectator = 0
self.number_of_LinphonePresenceActivitySteering = 0
self.number_of_LinphonePresenceActivityTravel = 0
self.number_of_LinphonePresenceActivityTV = 0
self.number_of_LinphonePresenceActivityUnknown = 0
self.number_of_LinphonePresenceActivityVacation = 0
self.number_of_LinphonePresenceActivityWorking = 0
self.number_of_LinphonePresenceActivityWorship = 0
self.number_of_inforeceived = 0
self.number_of_LinphoneSubscriptionIncomingReceived = 0
self.number_of_LinphoneSubscriptionOutgoingInit = 0
self.number_of_LinphoneSubscriptionPending = 0
self.number_of_LinphoneSubscriptionActive = 0
self.number_of_LinphoneSubscriptionTerminated = 0
self.number_of_LinphoneSubscriptionError = 0
self.number_of_LinphoneSubscriptionExpiring = 0
self.number_of_LinphonePublishProgress = 0
self.number_of_LinphonePublishOk = 0
self.number_of_LinphonePublishExpiring = 0
self.number_of_LinphonePublishError = 0
self.number_of_LinphonePublishCleared = 0
self.number_of_LinphoneConfiguringSkipped = 0
self.number_of_LinphoneConfiguringFailed = 0
self.number_of_LinphoneConfiguringSuccessful = 0
self.number_of_LinphoneCallEncryptedOn = 0
self.number_of_LinphoneCallEncryptedOff = 0
class CoreManager:
core_map = {}
@classmethod
def registration_state_changed(cls, lc, cfg, state, message):
logging.info("New registration state {state} for user id [{identity}] at proxy [{addr}]".format(
state=linphone.RegistrationState.string(state), identity=cfg.identity, addr=cfg.server_addr))
manager = CoreManager.core_map[lc]
if state == linphone.RegistrationState.RegistrationNone:
manager.stats.number_of_LinphoneRegistrationNone += 1
elif state == linphone.RegistrationState.RegistrationProgress:
manager.stats.number_of_LinphoneRegistrationProgress += 1
elif state == linphone.RegistrationState.RegistrationOk:
manager.stats.number_of_LinphoneRegistrationOk += 1
elif state == linphone.RegistrationState.RegistrationCleared:
manager.stats.number_of_LinphoneRegistrationCleared += 1
elif state == linphone.RegistrationState.RegistrationFailed:
manager.stats.number_of_LinphoneRegistrationFailed += 1
else:
raise Exception("Unexpected registration state")
@classmethod
def auth_info_requested(cls, lc, realm, username, domain):
logging.info("Auth info requested for user id [{username}] at realm [{realm}]".format(
username=username, realm=realm))
manager = CoreManager.core_map[lc]
manager.stats.number_of_auth_info_requested +=1
def __init__(self, rc_file = None, check_for_proxies = True, vtable = {}):
if not vtable.has_key('registration_state_changed'):
vtable['registration_state_changed'] = CoreManager.registration_state_changed
if not vtable.has_key('auth_info_requested'):
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('file_transfer_recv'):
#vtable['file_transfer_recv'] = CoreManager.file_transfer_recv
#if not vtable.has_key('file_transfer_send'):
#vtable['file_transfer_send'] = CoreManager.file_transfer_send
#if not vtable.has_key('file_transfer_progress_indication'):
#vtable['file_transfer_progress_indication'] = CoreManager.file_transfer_progress_indication
#if not vtable.has_key('is_composing_received'):
#vtable['is_composing_received'] = CoreManager.is_composing_received
#if not vtable.has_key('new_subscription_requested'):
#vtable['new_subscription_requested'] = CoreManager.new_subscription_requested
#if not vtable.has_key('notify_presence_received'):
#vtable['notify_presence_received'] = CoreManager.notify_presence_received
#if not vtable.has_key('transfer_state_changed'):
#vtable['transfer_state_changed'] = CoreManager.transfer_state_changed
#if not vtable.has_key('info_received'):
#vtable['info_received'] = CoreManager.info_received
#if not vtable.has_key('subscription_state_changed'):
#vtable['subscription_state_changed'] = CoreManager.subscription_state_changed
#if not vtable.has_key('notify_received'):
#vtable['notify_received'] = CoreManager.notify_received
#if not vtable.has_key('publish_state_changed'):
#vtable['publish_state_changed'] = CoreManager.publish_state_changed
#if not vtable.has_key('configuring_status'):
#vtable['configuring_status'] = CoreManager.configuring_status
#if not vtable.has_key('call_encryption_changed'):
#vtable['call_encryption_changed'] = CoreManager.call_encryption_changed
self.identity = None
self.stats = CoreManagerStats()
rc_path = None
tester_resources_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../tester/"))
if rc_file is not None:
rc_path = os.path.join('rcfiles', rc_file)
self.lc = self.configure_lc_from(vtable, tester_resources_path, rc_path)
CoreManager.core_map[self.lc] = self
if check_for_proxies and rc_file is not None:
proxy_count = len(self.lc.proxy_config_list)
else:
proxy_count = 0
if proxy_count:
self.wait_for_until(self.lc, None, lambda manager: manager.stats.number_of_LinphoneRegistrationOk == proxy_count, 5000 * proxy_count)
assert_equals(self.stats.number_of_LinphoneRegistrationOk, proxy_count)
self.enable_audio_codec("PCMU", 8000)
# TODO: Need to wrap getter of default proxy
#if self.lc.default_proxy is not None:
# self.identity = linphone.Address.new(self.lc.default_proxy.identity)
# self.identity.clean()
def stop(self):
del CoreManager.core_map[self.lc]
self.lc = None
def __del__(self):
self.stop()
def configure_lc_from(self, vtable, resources_path, rc_path):
filepath = None
if rc_path is not None:
filepath = os.path.join(resources_path, rc_path)
assert_equals(os.path.isfile(filepath), True)
lc = linphone.Core.new(vtable, None, filepath)
lc.root_ca = os.path.join(resources_path, 'certificates', 'cn', 'cafile.pem')
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')
return lc
def wait_for_until(self, lc_1, lc_2, func, timeout):
lcs = []
if lc_1 is not None:
lcs.append(lc_1)
if lc_2 is not None:
lcs.append(lc_2)
return self.wait_for_list(lcs, func, timeout)
def wait_for_list(self, lcs, func, timeout):
start = datetime.now()
end = start + timedelta(milliseconds = timeout)
res = func(self)
while not res and datetime.now() < end:
for lc in lcs:
lc.iterate()
time.sleep(0.02)
res = func(self)
return res
def enable_audio_codec(self, mime, rate):
codecs = self.lc.audio_codecs
for codec in codecs:
self.lc.enable_payload_type(codec, False)
codec = self.lc.find_payload_type(mime, rate, 1)
if codec is not None:
self.lc.enable_payload_type(codec, True)
from nose.tools import assert_equals
import linphone
import linphonetester
import os
import time
class RegisterCoreManager(linphonetester.CoreManager):
@classmethod
def auth_info_requested(cls, lc, realm, username, domain):
linphonetester.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
def __init__(self, with_auth = False):
vtable = {}
if with_auth:
vtable['auth_info_requested'] = RegisterCoreManager.auth_info_requested
linphonetester.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)
proxy_cfg.identity = from_address.as_string()
server_addr = from_address.domain
proxy_cfg.register_enabled = True
proxy_cfg.expires = 1
if route is None:
proxy_cfg.server_addr = server_addr
else:
proxy_cfg.route = route
proxy_cfg.server_addr = route
self.lc.add_proxy_config(proxy_cfg)
self.lc.default_proxy = proxy_cfg
#linphone_core_set_sip_transports(lc,&transport);
retry = 0
expected_count = 1
if refresh:
expected_count += 1
max_retry = 110
if expected_final_state == linphone.RegistrationState.RegistrationProgress:
max_retry += 200
while self.stats.number_of_LinphoneRegistrationOk < expected_count and retry < max_retry:
retry += 1
self.lc.iterate()
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
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):
break
time.sleep(0.1)
assert_equals(proxy_cfg.state, expected_final_state)
assert_equals(self.stats.number_of_LinphoneRegistrationNone, 0)
assert self.stats.number_of_LinphoneRegistrationProgress >= 1
if expected_final_state == linphone.RegistrationState.RegistrationOk:
assert_equals(self.stats.number_of_LinphoneRegistrationOk, expected_count)
expected_failed = 0
if late_auth_info:
expected_failed = 1
assert_equals(self.stats.number_of_LinphoneRegistrationFailed, expected_failed)
else:
assert_equals(self.stats.number_of_LinphoneRegistrationCleared, 0)
self.stop()
assert_equals(self.stats.number_of_LinphoneRegistrationCleared, 1)
class TestRegister:
@classmethod
def setup_class(cls):
base, ext = os.path.splitext(os.path.basename(__file__))
linphonetester.setup_logging(base + '.log')
def test_simple_register(self):
cm = RegisterCoreManager()
cm.register_with_refresh(False, None, None)
assert_equals(cm.stats.number_of_auth_info_requested, 0)
from nose.tools import assert_equals
import linphone
test_username = "liblinphone_tester"
test_route = "sip2.linphone.org"
def create_address(domain):
addr = linphone.Address.new(None)
assert addr != None
addr.username = test_username
assert_equals(addr.username, test_username)
if domain is not None:
domain = test_route
addr.domain = domain
assert_equals(addr.domain, domain)
addr.display_name = None
addr.display_name = "Mr Tester"
assert_equals(addr.display_name, "Mr Tester")
return addr
import linphonetester
class TestSetup:
def test_address(self):
create_address(None)
linphonetester.create_address(None)
def test_core_init(self):
lc = linphone.Core.new({}, None, None)
......
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