Skip to content
Snippets Groups Projects
Commit 5edb2065 authored by Alexandre Savard's avatar Alexandre Savard
Browse files

#9621: Use inheritance and overloading to derive directly from SflPhoneCtrl

parent d29b4a06
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
tools/pysflphone/sflphonectrlsimple.py → tools/pysflphone/sflphonectrl.py
+ 170
− 131
View file @ 5edb2065
......@@ -47,43 +47,46 @@ except ImportError, e:
raise SflPhoneError("No python-dbus module found")
class SflPhoneCtrlSimple(Thread):
""" Simple class for controlling SflPhoned through DBUS
If option testSuite (ts) is put to true,
simple actions are implemented on incoming call.
class SflPhoneCtrl(Thread):
""" class for controlling SflPhoned through DBUS
Classes deriving this class should reimplement signal handlers,
more especially:
onIncomingCall_cb
onCallHangup_cb
onCallRinging_cb
onCallHold_cb
onCallCurrent_cb
onCallBusy_cb
onCallFailure_cb
"""
# list of active calls (known by the client)
activeCalls = {}
def __init__(self, test=False, name=sys.argv[0]):
print "Create SFLphone instance"
Thread.__init__(self)
# current active account
def __init__(self, name=sys.argv[0]):
Thread.__init__(self)
# current active account
self.account = None
# client name
self.name = name
self.currentCallId = ""
self.isStop = False
# client registered to sflphoned ?
self.registered = False
self.register()
self.currentCallId = ""
# Glib MainLoop for processing callbacks
self.loop = MainLoop()
self.isStop = False
self.test = test
self.onIncomingCall_cb = None
self.onCallRinging_cb = None
self.onCallCurrent_cb = None
self.onCallFailure_cb = None
self.event = Event()
gobject.threads_init()
def __del__(self):
if self.registered:
self.unregister()
......@@ -91,11 +94,9 @@ class SflPhoneCtrlSimple(Thread):
def stopThread(self):
print "Stop PySFLphone"
self.isStop = True
def register(self):
if self.registered:
return
......@@ -148,14 +149,10 @@ class SflPhoneCtrlSimple(Thread):
print e
def unregister(self):
print "Unregister"
if not self.registered:
return
#raise SflPhoneError("Not registered !")
try:
self.instance.Unregister(os.getpid())
self.registered = False
......@@ -167,80 +164,115 @@ class SflPhoneCtrlSimple(Thread):
return self.registered
def getEvent(self):
return self.event
#
# Signal handling
#
def onIncomingCall_cb(self):
pass
def onCallHangup_cb(self):
pass
def wait(self):
self.event.wait()
def onCallRinging_cb(self):
pass
def isSet(self):
self.event.isSet()
def onCallHold_cb(self):
pass
def set(self):
self.event.set()
def onCallCurrent_cb(self):
pass
def clear(self):
self.event.clear()
def onCallBusy_cb(self):
pass
#
# Signal handling
#
def onCallFailure_cb(self):
pass
# On incoming call event, add the call to the list of active calls
def onIncomingCall(self, account, callid, to):
print "Incoming call: " + account + ", " + callid + ", " + to
self.activeCalls[callid] = {'Account': account, 'To': to, 'State': '' }
""" On incoming call event, add the call to the list of active calls """
self.activeCalls[callid] = {'Account': account,
'To': to,
'State': ''}
self.currentCallId = callid
self.onIncomingCall_cb()
def onCallHangUp(self, callid):
""" Remove callid from call list """
self.onCallHangup_cb()
self.currentCallId = ""
del self.activeCalls[callid]
def onCallRinging(self, callid, state):
""" Update state for this call to Ringing """
self.activeCalls[callid]['State'] = state
self.onCallRinging_cb()
def onCallHold(self, callid, state):
""" Update state for this call to Hold """
self.activeCalls[callid]['State'] = state
self.onCallHold_cb()
if(self.test):
# TODO fix this bug in daemon, cannot answer too fast
time.sleep(0.5)
if self.onIncomingCall_cb(self):
self.onIncomingCall_cb(self)
def onCallCurrent(self, callid, state):
""" Update state for this call to current """
self.activeCalls[callid]['State'] = state
self.onCallCurrent_cb()
def onCallBusy(self, callid, state):
""" Update state for this call to busy """
self.activeCalls[callid]['State'] = state
self.onCallBusy_cb()
def onCallFailure(self, callid, state):
""" Handle call failure """
self.onCallFailure_cb(self)
del self.activeCalls[callid]
# On call state changed event, set the values for new calls,
# or delete the call from the list of active calls
def onCallStateChanged(self, callid, state):
print "Call state changed: " + callid + ", " + state
""" On call state changed event, set the values for new calls,
or delete the call from the list of active calls
"""
print "On call state changed " + callid + " " + state
if callid not in self.activeCalls:
print "This call didn't exist!: " + callid + ". Adding it to the list."
callDetails = self.getCallDetails(callid)
self.activeCalls[callid] = {'Account': callDetails['ACCOUNTID'],
'To': callDetails['PEER_NUMBER'],
'State': state }
self.currentCallId = callid
if state is "HUNGUP":
try:
del self.activeCalls[callid]
except KeyError:
print "Call " + callid + " didn't exist. Cannot delete."
elif state is "RINGING":
try:
self.activeCalls[callid]['State'] = state
if self.onCallRinging_cb:
self.onCallRinging_cb(self)
except KeyError, e:
print "This call didn't exist!: " + callid + ". Adding it to the list."
callDetails = self.getCallDetails(callid)
self.activeCalls[callid] = {'Account': callDetails['ACCOUNTID'],
'To': callDetails['PEER_NUMBER'], 'State': state }
elif state in [ "CURRENT", "INCOMING", "HOLD" ]:
try:
self.activeCalls[callid]['State'] = state
if self.onCallCurrent_cb:
self.onCallCurrent_cb(self)
callDetails = self.getCallDetails(callid)
self.activeCalls[callid] = {'Account': callDetails['ACCOUNTID'],
'To': callDetails['PEER_NUMBER'], 'State': state }
except KeyError, e:
print "This call didn't exist!: " + callid + ". Adding it to the list."
elif state in [ "BUSY", "FAILURE" ]:
try:
if self.onCallFailure_cb:
self.onCallFailure_cb(self)
del self.activeCalls[callid]
except KeyError, e:
print "This call didn't exist!: " + callid
# elif state == "UNHOLD_CURRENT":
# self.activeCalls[callid]['State'] = "UNHOLD_CURRENT"
if state == "HUNGUP":
self.onCallHangUp(callid)
elif state == "RINGING":
self.onCallRinging(callid, state)
elif state == "CURRENT":
self.onCallCurrent(callid, state)
elif state == "HOLD":
self.onCallHold(callid, state)
elif state == "BUSY":
self.onCallBusy(callid, state)
elif state == "FAILURE":
self.onCallFailure(self, callid, state)
else:
print "unknown state"
#
......@@ -265,6 +297,7 @@ class SflPhoneCtrlSimple(Thread):
return self.configurationmanager.addAccount(details)
def removeAccount(self, accountID=None):
"""Remove an account from internal list"""
......@@ -273,13 +306,16 @@ class SflPhoneCtrlSimple(Thread):
self.configurationmanager.removeAccount(accountID)
def getAllAccounts(self):
"""Return a list with all accounts"""
return self.configurationmanager.getAccountList()
def getAllEnabledAccounts(self):
"""Return a list with all enabled accounts"""
accounts = self.getAllAccounts()
activeaccounts = []
for testedaccount in accounts:
......@@ -324,6 +360,7 @@ class SflPhoneCtrlSimple(Thread):
raise SPaccountError("No account matched with alias")
def setAccount(self, account):
"""Define the active account
......@@ -336,6 +373,7 @@ class SflPhoneCtrlSimple(Thread):
print account
raise SflPhoneError("Not a valid account")
def setFirstRegisteredAccount(self):
"""Find the first enabled account and define it as active"""
......@@ -344,6 +382,7 @@ class SflPhoneCtrlSimple(Thread):
raise SflPhoneError("No registered account !")
self.account = rAccounts[0]
def setFirstActiveAccount(self):
"""Find the first enabled account and define it as active"""
......@@ -373,33 +412,37 @@ class SflPhoneCtrlSimple(Thread):
"""Return True if the account is enabled. If no account is provided, active account is used"""
if account is None:
if self.account is None:
raise SflPhoneError("No provided or current account !")
if self.account is None:
raise SflPhoneError("No provided or current account !")
account = self.account
return self.getAccountDetails(account)['Account.enable'] == "TRUE"
def setAccountEnable(self, account=None, enable=False):
"""Set account enabled"""
"""Set account enabled"""
if account is None:
if self.account is None:
raise SflPhoneError("No provided or current account !")
if self.account is None:
raise SflPhoneError("No provided or current account !")
account = self.account
if enable == True:
details = self.getAccountDetails(account)
details['Account.enable'] = "true"
self.configurationmanager.setAccountDetails(account, details)
if enable == True:
details = self.getAccountDetails(account)
details['Account.enable'] = "true"
self.configurationmanager.setAccountDetails(account, details)
else:
details = self.getAccountDetails(account)
details['Account.enable'] = "false"
self.configurationmanager.setAccountDetails(account, details)
details = self.getAccountDetails(account)
details['Account.enable'] = "false"
self.configurationmanager.setAccountDetails(account, details)
def checkAccountExists(self, account=None):
""" Checks if the account exists """
if account is None:
raise SflPhoneError("No provided or current account !")
return account in self.getAllAccounts()
def getAllRegisteredAccounts(self):
"""Return a list of registered accounts"""
......@@ -410,6 +453,7 @@ class SflPhoneCtrlSimple(Thread):
return registeredAccountsList
def getAllEnabledAccounts(self):
"""Return a list of enabled accounts"""
......@@ -420,8 +464,10 @@ class SflPhoneCtrlSimple(Thread):
return enabledAccountsList
def getAllSipAccounts(self):
"""Return a list of SIP accounts"""
sipAccountsList = []
for accountName in self.getAllAccounts():
if self.getAccountDetails(accountName)['Account.type'] == "SIP":
......@@ -429,6 +475,7 @@ class SflPhoneCtrlSimple(Thread):
return sipAccountsList
def getAllIaxAccounts(self):
"""Return a list of IAX accounts"""
......@@ -439,23 +486,22 @@ class SflPhoneCtrlSimple(Thread):
return iaxAccountsList
def setAccountRegistered(self, account=None, register=False):
""" Tries to register the account """
if account is None:
if self.account is None:
raise SflPhoneError("No provided or current account !")
account = self.account
try:
if register:
self.configurationmanager.sendRegister(account, int(1))
#self.setAccount(account)
else:
self.configurationmanager.sendRegister(account, int(0))
#self.setFirstRegisteredAccount()
""" Tries to register the account """
if account is None:
if self.account is None:
raise SflPhoneError("No provided or current account !")
account = self.account
try:
if register:
self.configurationmanager.sendRegister(account, int(1))
else:
self.configurationmanager.sendRegister(account, int(0))
except SflPhoneError, e:
print e
print e
#
# Codec manager
......@@ -463,12 +509,14 @@ class SflPhoneCtrlSimple(Thread):
def getCodecList(self):
""" Return the codec list """
return self.configurationmanager.getCodecList()
def getActiveCodecList(self):
""" Return the active codec list """
return self.configurationmanager.getActiveCodecList()
return self.configurationmanager.getActiveCodecList()
#
......@@ -539,26 +587,18 @@ class SflPhoneCtrlSimple(Thread):
def HangUp(self, callid):
"""End a call identified by a CallID"""
if not self.account:
self.setFirstRegisteredAccount()
# if not self.isAccountRegistered() and self.accout is not "IP2IP":
# raise SflPhoneError("Can't hangup a call without a registered account")
if callid is None or callid == "":
pass # just to see
#raise SflPhoneError("Invalid callID")
self.callmanager.hangUp(callid)
def Transfer(self, callid, to):
"""Transfert a call identified by a CallID"""
# if not self.account:
# self.setFirstRegisteredAccount()
# if not self.isAccountRegistered():
# raise SflPhoneError("Can't transfert a call without a registered account")
if callid is None or callid == "":
raise SflPhoneError("Invalid callID")
......@@ -571,12 +611,6 @@ class SflPhoneCtrlSimple(Thread):
print "Refuse call " + callid
# if not self.account:
# self.setFirstRegisteredAccount()
# if not self.isAccountRegistered():
# raise SflPhoneError("Can't refuse a call without a registered account")
if callid is None or callid == "":
raise SflPhoneError("Invalid callID")
......@@ -585,11 +619,12 @@ class SflPhoneCtrlSimple(Thread):
def Accept(self, callid):
"""Accept an incoming call identified by a CallID"""
print "Accept call " + callid
if not self.account:
self.setFirstRegisteredAccount()
if not self.isAccountRegistered():
if not self.isAccountRegistered():
raise SflPhoneError("Can't accept a call without a registered account")
if callid is None or callid == "":
......@@ -600,6 +635,7 @@ class SflPhoneCtrlSimple(Thread):
def Hold(self, callid):
"""Hold a call identified by a CallID"""
# if not self.account:
# self.setFirstRegisteredAccount()
......@@ -614,6 +650,7 @@ class SflPhoneCtrlSimple(Thread):
def UnHold(self, callid):
"""Unhold an incoming call identified by a CallID"""
# if not self.account:
# self.setFirstRegisteredAccount()
......@@ -628,11 +665,13 @@ class SflPhoneCtrlSimple(Thread):
def Dtmf(self, key):
"""Send a DTMF"""
self.callmanager.playDTMF(key)
def GenerateCallID(self):
"""Generate Call ID"""
m = hashlib.md5()
t = long( time.time() * 1000 )
r = long( random.random()*100000000000000000L )
......@@ -649,4 +688,4 @@ class SflPhoneCtrlSimple(Thread):
context.iteration(True)
if self.isStop:
return
return
tools/pysflphone/test_sflphone_dbus_interface.py
+ 49
− 51
View file @ 5edb2065
......@@ -21,7 +21,7 @@ import time
import logging
from sippwrap import SippWrapper
from sippwrap import SippScreenStatParser
from sflphonectrlsimple import SflPhoneCtrlSimple
from sflphonectrl import SflPhoneCtrl
from nose.tools import nottest
......@@ -33,37 +33,11 @@ accountList = ["IP2IP", "Account:1332798167"]
SCENARIO_PATH = "../sippxml/"
def callHangup(sflphone):
""" On incoming call, answer the callm, then hangup """
print "Hangup Call with id " + sflphone.currentCallId
sflphone.HangUp(sflphone.currentCallId)
time.sleep(3)
print "Stopping Thread"
sflphone.stopThread()
def callIsRinging(sflphone):
""" Display messages when call is ringing """
print "The call is ringing"
def leaveThreadOnFailure(sflphone):
""" If a failure occurs duing the call, just leave the running thread """
print "Stopping Thread"
sflphone.stopThread()
class TestSFLPhoneAccountConfig:
class TestSFLPhoneAccountConfig(SflPhoneCtrl):
""" The test suite for account configuration """
def __init__(self):
self.sflphone = SflPhoneCtrlSimple(True)
SflPhoneCtrl.__init__(self)
self.logger = logging.getLogger("TestSFLPhoneAccountConfig")
filehdlr = logging.FileHandler("/tmp/sflphonedbustest.log")
......@@ -75,28 +49,28 @@ class TestSFLPhoneAccountConfig:
@nottest
def test_get_account_list(self):
self.logger.info("Test get account list")
accList = self.sflphone.getAllAccounts()
accList = self.getAllAccounts()
listIntersection = set(accList) & set(accountList)
assert len(listIntersection) == len(accountList)
@nottest
def test_account_registration(self):
self.logger.info("Test account registration")
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
accList = [x for x in self.getAllAccounts() if x != "IP2IP"]
for acc in accList:
self.logger.info("Registering account " + acc)
if self.sflphone.isAccountEnable(acc):
self.sflphone.setAccountEnable(acc, False)
if self.isAccountEnable(acc):
self.setAccountEnable(acc, False)
time.sleep(2)
# Account should not be registered
assert self.sflphone.isAccountRegistered(acc)
assert self.isAccountRegistered(acc)
self.sflphone.setAccountEnable(acc, True)
self.setAccountEnable(acc, True)
time.sleep(2)
assert self.sflphone.isAccountRegistered(acc)
assert self.isAccountRegistered(acc)
@nottest
def test_add_remove_account(self):
......@@ -105,31 +79,31 @@ class TestSFLPhoneAccountConfig:
newAccList = []
# consider only true accounts
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
accList = [x for x in self.getAllAccounts() if x != "IP2IP"]
# Store the account details localy
for acc in accList:
accountDetails[acc] = self.sflphone.getAccountDetails(acc)
accountDetails[acc] = self.getAccountDetails(acc)
# Remove all accounts from sflphone
for acc in accountDetails:
self.sflphone.removeAccount(acc)
self.removeAccount(acc)
# Recreate all accounts
for acc in accountDetails:
newAccList.append(self.sflphone.addAccount(accountDetails[acc]))
newAccList.append(self.addAccount(accountDetails[acc]))
# New accounts should be automatically registered
for acc in newAccList:
assert self.sflphone.isAccountRegistered(acc)
assert self.isAccountRegistered(acc)
class TestSFLPhoneRegisteredCalls:
class TestSFLPhoneRegisteredCalls(SflPhoneCtrl):
""" The test suite for call interaction """
def __init__(self):
self.sflphone = SflPhoneCtrlSimple(True)
SflPhoneCtrl.__init__(self)
self.logger = logging.getLogger("TestSFLPhoneRegisteredCalls")
filehdlr = logging.FileHandler("/tmp/sfltestregisteredcall.log")
......@@ -141,19 +115,41 @@ class TestSFLPhoneRegisteredCalls:
self.sippCallInstance = SippWrapper()
self.localInterface = "127.0.0.1"
self.localPort = str(5064)
self.sflphone.onCallRinging_cb = callIsRinging
self.sflphone.onCallCurrent_cb = callHangup
self.sflphone.onCallFailure_cb = leaveThreadOnFailure
# Make sure the test directory is populated with most recent log files
self.clean_log_directory()
def onCallCurrent_cb(self):
""" On incoming call, answer the callm, then hangup """
print "Hangup Call with id " + self.currentCallId
self.HangUp(self.currentCallId)
print "Stopping Thread"
self.stopThread()
def onCallRinging_cb(self):
""" Display messages when call is ringing """
print "The call is ringing"
def onCallFailure_cb(self):
""" If a failure occurs duing the call, just leave the running thread """
print "Stopping Thread"
self.stopThread()
def clean_log_directory(self):
dirlist = os.listdir("./")
files = [x for x in dirlist if "screen.log" in x]
for f in files:
os.remove(f)
def find_sipp_pid(self):
# Retreive the PID of the last
# The /proc/PID/cmdline contain the command line from
......@@ -162,6 +158,7 @@ class TestSFLPhoneRegisteredCalls:
return sippPid[0]
def parse_results(self):
dirlist = os.listdir("./")
logfile = [x for x in dirlist if "screen.log" in x]
......@@ -175,6 +172,7 @@ class TestSFLPhoneRegisteredCalls:
assert(not resultParser.isAnyFailedCall())
assert(resultParser.isAnySuccessfulCall())
def test_registered_call(self):
self.logger.info("Test Registered Call")
......@@ -209,16 +207,16 @@ class TestSFLPhoneRegisteredCalls:
sippPid = self.find_sipp_pid()
# make sure every account are enabled
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
accList = [x for x in self.getAllAccounts() if x != "IP2IP"]
for acc in accList:
if not self.sflphone.isAccountRegistered(acc):
self.sflphone.setAccountEnable(acc, True)
if not self.isAccountRegistered(acc):
self.setAccountEnable(acc, True)
# Make a call to the SIPP instance
self.sflphone.Call("300")
self.Call("300")
# Start Glib mainloop to process callbacks
self.sflphone.start()
self.start()
# Wait the sipp instance to dump log files
while os.path.exists("/proc/" + str(sippPid)):
......
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