Commit f943678e authored by Alexandre Savard's avatar Alexandre Savard
Browse files

#9621 Make registered account call as functional tests

parent a4525f8d
......@@ -75,10 +75,13 @@ class SflPhoneCtrlSimple(Thread):
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):
......@@ -90,7 +93,7 @@ class SflPhoneCtrlSimple(Thread):
def stopThread(self):
print "Stop PySFLphone"
self.isStop = True
def register(self):
......@@ -192,31 +195,46 @@ class SflPhoneCtrlSimple(Thread):
if(self.test):
# TODO fix this bug in daemon, cannot answer too fast
time.sleep(0.5)
if self.onIncomingCall_cb(self) is not None:
if self.onIncomingCall_cb(self):
self.onIncomingCall_cb(self)
# 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
if state == "HUNGUP":
self.currentCallId = callid
if state is "HUNGUP":
try:
del self.activeCalls[callid]
except KeyError:
print "Call " + callid + " didn't exist. Cannot delete."
elif state in [ "RINGING", "CURRENT", "INCOMING", "HOLD" ]:
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
......@@ -237,7 +255,7 @@ class SflPhoneCtrlSimple(Thread):
Required parameters are type, alias, hostname, username and password
input details
"""
if details is None:
......@@ -381,7 +399,7 @@ class SflPhoneCtrlSimple(Thread):
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"""
......@@ -496,7 +514,7 @@ class SflPhoneCtrlSimple(Thread):
if dest is None or dest == "":
raise SflPhoneError("Invalid call destination")
# Set the account to be used for this call
if dest.find('sip:') is 0 or dest.find('sips:') is 0:
print "Ip 2 IP call"
......@@ -508,7 +526,7 @@ class SflPhoneCtrlSimple(Thread):
raise SflPhoneError("Can't place a call without a registered account")
# Generate a call ID for this call
callid = self.GenerateCallID()
callid = self.GenerateCallID()
# Add the call to the list of active calls and set status to SENT
self.activeCalls[callid] = {'Account': self.account, 'To': dest, 'State': 'SENT' }
......@@ -576,7 +594,7 @@ class SflPhoneCtrlSimple(Thread):
if callid is None or callid == "":
raise SflPhoneError("Invalid callID")
self.callmanager.accept(callid)
......
#!/usr/bin/env python
#
# Copyright (C) 2009 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import os
class SippWrapper:
""" Wrapper taht allow for managing sipp command line easily """
def __init__(self):
self.commandLine = "sipp"
self.remoteServer = ""
self.remotePort = ""
self.localInterface = ""
self.localPort = ""
self.customScenarioFile = ""
self.isUserAgenClient = True
self.launchInBackground = False
self.numberOfCall = 0
self.numberOfSimultaneousCall = 0
self.enableTraceMsg = False
self.enableTraceShormsg = False
self.enableTraceScreen = False
self.enableTraceError = False
self.enableTraceStat = False
self.enableTraceCounts = False
self.enableTraceRtt = False
self.enableTraceLogs = False
def buildCommandLine(self):
""" Fill the command line arguments based on specified parameters """
if not self.remotePort and not self.remoteServer:
self.isUserAgentClient = False
elif self.remotePort and not self.remoteServer:
print "Error cannot have remote port specified with no server"
return
if self.remoteServer:
self.commandLine += " " + self.remoteServer
if self.remotePort:
self.commandLine += ":" + self.remotePort
if self.localInterface:
self.commandLine += " -i " + self.localInterface
if self.localPort:
self.commandLine += " -p " + self.localPort
if self.customScenarioFile:
self.commandLine += " -sf " + self.customScenarioFile
elif self.isUserAgentClient is True:
self.commandLine += " -sn uac"
elif self.isUserAgentClient is False:
self.commandLine += " -sn uas"
if self.launchInBackground:
self.commandLine += " -bg"
if self.numberOfCall:
self.commandLine += " -m " + str(self.numberOfCall)
if self.numberOfSimultaneousCall:
self.commandLine += " -l " + str(self.numberOfSimultaneousCall)
if self.enableTraceMsg:
self.commandLine += " -trace_msg"
if self.enableTraceShormsg:
self.commandLine += " -trace_shortmsg"
if self.enableTraceScreen:
self.commandLine += " -trace_screen"
if self.enableTraceError:
self.commandLine += " -trace_err"
if self.enableTraceStat:
self.commandLine += " -trace_stat"
if self.enableTraceCounts:
self.commandLine += " -trace_counts"
if self.enableTraceRtt:
self.commandLine += " -trace_rtt"
if self.enableTraceLogs:
self.commandLine += " -trace_logs"
def launch(self):
""" Launch the sipp instance using the specified arguments """
self.buildCommandLine()
print self.commandLine
return os.system(self.commandLine)
class SippScreenStatParser:
""" Class that parse statistic reported by a sipp instance
report some of the most important value """
def __init__(self, filename):
print "Opening " + filename
self.logfile = open(filename, "r").readlines()
print self.logfile[39]
print self.logfile[40]
def isAnyFailedCall(self):
""" Look for any failed call
Return true if there are failed call, false elsewhere """
# TODO: Find a better way to determine which line to consider
if "Failed call" not in self.logfile[40]:
print "Error: Could not find 'Failed call' statistics"
# We consider this as a failure
return True
return "1" in self.logfile[40]
def isAnySuccessfulCall(self):
""" Look for any successful call
Return true if there are successful call, false elsewhere """
# TODO: Find a better way to determine which line to consider
if "Successful call" not in self.logfile[39]:
print "Error: Could not find 'Successful call' statistics"
return False
return "1" in self.logfile[39]
def test_result_parsing():
dirlist = os.listdir("./")
logfile = [x for x in dirlist if "screen.log" in x]
testResult = SippScreenStatParser(logfile[0])
assert(not testResult.isAnyFailedCall())
assert(testResult.isAnySuccessfulCall())
......@@ -16,8 +16,11 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
import os
import time
import logging
from sippwrap import SippWrapper
from sippwrap import SippScreenStatParser
from sflphonectrlsimple import SflPhoneCtrlSimple
from nose.tools import nottest
......@@ -26,14 +29,42 @@ from nose.tools import nottest
### function starting with 'test' are executed.
###
# Open sflphone and connect to sflphoned through dbus
sflphone = SflPhoneCtrlSimple(True)
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:
""" The test suite for account configuration """
def __init__(self):
self.sflphone = SflPhoneCtrlSimple(True)
self.logger = logging.getLogger("TestSFLPhoneAccountConfig")
filehdlr = logging.FileHandler("/tmp/sflphonedbustest.log")
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
......@@ -44,28 +75,28 @@ class TestSFLPhoneAccountConfig:
@nottest
def test_get_account_list(self):
self.logger.info("Test get account list")
accList = sflphone.getAllAccounts()
accList = self.sflphone.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 sflphone.getAllAccounts() if x != "IP2IP"]
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
for acc in accList:
self.logger.info("Registering account " + acc)
if sflphone.isAccountEnable(acc):
sflphone.setAccountEnable(acc, False)
if self.sflphone.isAccountEnable(acc):
self.sflphone.setAccountEnable(acc, False)
time.sleep(2)
# Account should not be registered
assert sflphone.isAccountRegistered(acc)
assert self.sflphone.isAccountRegistered(acc)
sflphone.setAccountEnable(acc, True)
self.sflphone.setAccountEnable(acc, True)
time.sleep(2)
assert sflphone.isAccountRegistered(acc)
assert self.sflphone.isAccountRegistered(acc)
@nottest
def test_add_remove_account(self):
......@@ -74,20 +105,123 @@ class TestSFLPhoneAccountConfig:
newAccList = []
# consider only true accounts
accList = [x for x in sflphone.getAllAccounts() if x != "IP2IP"]
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
# Store the account details localy
for acc in accList:
accountDetails[acc] = sflphone.getAccountDetails(acc)
accountDetails[acc] = self.sflphone.getAccountDetails(acc)
# Remove all accounts from sflphone
for acc in accountDetails:
sflphone.removeAccount(acc)
self.sflphone.removeAccount(acc)
# Recreate all accounts
for acc in accountDetails:
newAccList.append(sflphone.addAccount(accountDetails[acc]))
newAccList.append(self.sflphone.addAccount(accountDetails[acc]))
# New accounts should be automatically registered
for acc in newAccList:
assert sflphone.isAccountRegistered(acc)
assert self.sflphone.isAccountRegistered(acc)
class TestSFLPhoneRegisteredCalls:
""" The test suite for call interaction """
def __init__(self):
self.sflphone = SflPhoneCtrlSimple(True)
self.logger = logging.getLogger("TestSFLPhoneRegisteredCalls")
filehdlr = logging.FileHandler("/tmp/sfltestregisteredcall.log")
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
filehdlr.setFormatter(formatter)
self.logger.addHandler(filehdlr)
self.logger.setLevel(logging.INFO)
self.sippRegistrationInstance = SippWrapper()
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 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
pids = [int(x) for x in os.listdir("/proc") if x.isdigit()]
sippPid = [pid for pid in pids if "sipp" in open("/proc/" + str(pid) + "/cmdline").readline()]
return sippPid[0]
def parse_results(self):
dirlist = os.listdir("./")
logfile = [x for x in dirlist if "screen.log" in x]
print logfile
fullpath = os.path.dirname(os.path.realpath(__file__)) + "/"
# there should be only one screen.log file (see clean_log_directory)
resultParser = SippScreenStatParser(fullpath + logfile[0])
assert(not resultParser.isAnyFailedCall())
assert(resultParser.isAnySuccessfulCall())
def test_registered_call(self):
self.logger.info("Test Registered Call")
# launch the sipp instance in background
# sipp 127.0.0.1:5060 -sf uac_register_no_cvs.xml -i 127.0.0.1 -p 5062
self.sippRegistrationInstance.remoteServer = "127.0.0.1"
self.sippRegistrationInstance.remotePort = str(5062)
self.sippRegistrationInstance.localInterface = self.localInterface
self.sippRegistrationInstance.localPort = self.localPort
self.sippRegistrationInstance.customScenarioFile = SCENARIO_PATH + "uac_register_no_cvs.xml"
self.sippRegistrationInstance.launchInBackground = True
self.sippRegistrationInstance.numberOfCall = 1
self.sippRegistrationInstance.numberOfSimultaneousCall = 1
self.sippRegistrationInstance.launch()
# wait for this instance of sipp to complete registration
sippPid = self.find_sipp_pid()
while os.path.exists("/proc/" + str(sippPid)):
time.sleep(1)
# sipp -sn uas -p 5062 -i 127.0.0.1
self.sippCallInstance.localInterface = self.localInterface
self.sippCallInstance.localPort = self.localPort
self.sippCallInstance.launchInBackground = True
self.sippCallInstance.numberOfCall = 1
self.sippCallInstance.numberOfSimultaneousCall = 1
self.sippCallInstance.enableTraceScreen = True
self.sippCallInstance.launch()
sippPid = self.find_sipp_pid()
# make sure every account are enabled
accList = [x for x in self.sflphone.getAllAccounts() if x != "IP2IP"]
for acc in accList:
if not self.sflphone.isAccountRegistered(acc):
self.sflphone.setAccountEnable(acc, True)
# Make a call to the SIPP instance
self.sflphone.Call("300")
# Start Glib mainloop to process callbacks
self.sflphone.start()
# Wait the sipp instance to dump log files
while os.path.exists("/proc/" + str(sippPid)):
time.sleep(1)
self.parse_results()
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