[svn-commits] mjordan: testsuite/asterisk/trunk r5037 - in /asterisk/trunk/tests/apps: chan...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Thu May 15 11:28:58 CDT 2014
Author: mjordan
Date: Thu May 15 11:28:55 2014
New Revision: 5037
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=5037
Log:
apps/mixmonitor|chanspy: Fix failing tests
Much like the FastAGI tests, these two tests (mixmonitor and chanspy_barge)
required knowledge of the test artifact directory structure but made these
assumptions outside of the TestCase class. This patch ports these two tests
over to use TestCase as well.
Modified:
asterisk/trunk/tests/apps/chanspy/chanspy_barge/configs/ast1/extensions.conf
asterisk/trunk/tests/apps/chanspy/chanspy_barge/run-test
asterisk/trunk/tests/apps/mixmonitor/run-test
Modified: asterisk/trunk/tests/apps/chanspy/chanspy_barge/configs/ast1/extensions.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/apps/chanspy/chanspy_barge/configs/ast1/extensions.conf?view=diff&rev=5037&r1=5036&r2=5037
==============================================================================
--- asterisk/trunk/tests/apps/chanspy/chanspy_barge/configs/ast1/extensions.conf (original)
+++ asterisk/trunk/tests/apps/chanspy/chanspy_barge/configs/ast1/extensions.conf Thu May 15 11:28:55 2014
@@ -22,3 +22,4 @@
exten => play_recording,1,Answer()
exten => play_recording,n,PlayBack(${TESTAUDIO1})
+exten => play_recording,n,Echo()
Modified: asterisk/trunk/tests/apps/chanspy/chanspy_barge/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/apps/chanspy/chanspy_barge/run-test?view=diff&rev=5037&r1=5036&r2=5037
==============================================================================
--- asterisk/trunk/tests/apps/chanspy/chanspy_barge/run-test (original)
+++ asterisk/trunk/tests/apps/chanspy/chanspy_barge/run-test Thu May 15 11:28:55 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
David Vossel <dvossel at digium.com>
+Matt Jordan <mjordan at digium.com>
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
@@ -11,19 +12,23 @@
import os
import signal
import subprocess
+import logging
+
from twisted.application import service, internet
from twisted.internet import reactor, defer
from starpy import manager
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "apps/chanspy/chanspy_barge"
-testdir = "tests/%s" % workingdir
+LOGGER = logging.getLogger(__name__)
-class ChanSpyBarge:
+testdir = "tests/apps/chanspy/chanspy_barge"
+
+class ChanSpyBarge(TestCase):
def __init__(self):
+ super(ChanSpyBarge, self).__init__()
self.last_step = ""
self.passed = False
self.numSpyEvents = 0
@@ -31,21 +36,18 @@
self.talkDetected = 0
self.test_to = 25
- reactor.callWhenRunning(self.run)
- self.agi = fastagi.FastAGIFactory(self.reportPassed)
- reactor.listenTCP(4573, self.agi, self.test_to, '127.0.0.1')
+ self.create_asterisk()
+ self.create_fastagi_factory()
ast_conf_options = {
"transmit_silence" : "yes",
}
- self.ast1 = Asterisk(base=workingdir, ast_conf_options=ast_conf_options)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
self.talkingaudio = os.path.join(os.getcwd(), "%s/sounds/talking" % (testdir))
self.audiofile1 = os.path.join(os.getcwd(), "%s/testaudio1" % (testdir))
- def reportPassed(self, agi):
- print "GOT PASS RESULTS!!!\n"
+ def fastagi_connect(self, agi):
+ LOGGER.info("GOT PASS RESULTS!!!")
sequence = fastagi.InSequence()
sequence.append(agi.execute, "HangUp")
sequence.append(agi.finish)
@@ -55,101 +57,80 @@
def readResult(self):
self.stop_reactor()
- self.logLastStep("Reading results")
- self.ast1.cli_exec("core show locks") # get lock output in case of deadlock before tearing down.
- self.ast1.cli_exec("core show channels")# if channels are still up for some reason, we want to know that as well
+ LOGGER.info("Reading results")
+ self.ast[0].cli_exec("core show locks") # get lock output in case of deadlock before tearing down.
+ self.ast[0].cli_exec("core show channels")# if channels are still up for some reason, we want to know that as well
- self.ast1.cli_exec("core show globals") # The global variables here hold failure conditions
+ self.ast[0].cli_exec("core show globals") # The global variables here hold failure conditions
if self.passed == True:
- print 'SIP ChanSpy test PASSED!'
+ LOGGER.info('SIP ChanSpy test PASSED!')
else:
- print 'SIP ChanSpy Test FAILED'
-
- def stop_reactor(self):
- def __finish_stop(result):
- print "Stopping Reactor ..."
- if reactor.running:
- reactor.stop()
- return result
- df = self.ast1.stop()
- df.addCallback(__finish_stop)
+ LOGGER.error('SIP ChanSpy Test FAILED')
def chanspyEvent(self, ami, event):
- self.logLastStep('Received ChanSpyStart event')
+ LOGGER.info('Received ChanSpyStart event')
self.numSpyEvents += 1
if event['spyeechannel'].count('end_a') > 0:
reactor.callLater(3, self.aHangup)
reactor.callLater(4, self.bHangup)
- def amiOnConnect(self, ami):
- self.logLastStep("Connected to the AMI")
+ def ami_connect(self, ami):
self.ami = ami
self.ami.registerEvent('ChanSpyStart', self.chanspyEvent)
self.ami.setVar(channel = "", variable = "TESTAUDIO1", value = self.audiofile1)
self.ami.setVar(channel = "", variable = "TALK_AUDIO", value = self.talkingaudio)
- def amiLoginError(self, ami):
- self.logLastStep("AMI login failed")
- reactor.callLater(1, self.readResult)
+ reactor.callLater(1, self.callChanSpy)
+ reactor.callLater(2, self.aCall)
- def amiLogin(self):
- self.logLastStep("Logging in to the AMI")
- self.ami_factory = manager.AMIFactory("user", "mysecret")
- self.ami_factory.login('127.0.0.1', 5038).addCallbacks(self.amiOnConnect, self.amiLoginError)
def aCall(self):
- self.logLastStep("A Calling into Wait")
+ LOGGER.info("A Calling into Wait")
self.pja.stdin.write("m\n")
self.pja.stdin.write("sip:play_exten at 127.0.0.1:5060\n")
def bCall(self):
- self.logLastStep("B Calling into Playback")
+ LOGGER.info("B Calling into Playback")
self.pjb.stdin.write("m\n")
self.pjb.stdin.write("sip:play_exten at 127.0.0.1:5060\n")
def aHangup(self):
+ LOGGER.info("Hanging up A")
self.pja.stdin.write("h\n")
def bHangup(self): #calls into chanspy extension and plays audio to A using Barge
+ LOGGER.info("Hanging up B")
self.pjb.stdin.write("h\n")
reactor.callLater(2, self.verifyAudio)
def startProcesses(self):
- self.logLastStep("Starting Processes")
+ LOGGER.info("Starting Processes")
playfilearg = "--play-file=%s.wav" % (self.talkingaudio)
self.pja = subprocess.Popen(['pjsua', '--local-port=5065', '--auto-answer=200', '--null-audio', '--auto-loop'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
self.pjb = subprocess.Popen(['pjsua', '--local-port=5066', '--auto-answer=200', playfilearg, '--null-audio', '--auto-play'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
def stopProcesses(self):
- self.logLastStep("Stopping Processes")
+ LOGGER.info("Stopping Processes")
os.kill(self.pja.pid, signal.SIGKILL)
os.kill(self.pjb.pid, signal.SIGKILL)
- def logLastStep(self, step):
- print step
- self.lastStep = step
-
def callChanSpy(self):
- self.logLastStep("Placing call to ChanSpy extension.")
- self.ast1.cli_originate("SIP/end_b extension chanspytest at test")
+ LOGGER.info("Placing call to ChanSpy extension.")
+ self.ast[0].cli_originate("SIP/end_b extension chanspytest at test")
def verifyAudio(self):
- self.ast1.cli_originate("Local/play_recording at test extension detect_audio at test")
+ LOGGER.info("Verifying Audio")
+ self.ast[0].cli_originate("Local/play_recording at test extension detect_audio at test")
+
+ def stop_asterisk(self):
+ if not self.passed:
+ self.readResult()
def run(self):
- def __finish_start_ops(result):
- # call extensions
- self.amiLogin()
- reactor.callLater(1, self.callChanSpy)
- reactor.callLater(2, self.aCall)
- # stop and read results after timeout
- reactor.callLater(self.test_to, self.readResult)
- return result
+ self.create_ami_factory()
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
def main():
test = ChanSpyBarge()
Modified: asterisk/trunk/tests/apps/mixmonitor/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/apps/mixmonitor/run-test?view=diff&rev=5037&r1=5036&r2=5037
==============================================================================
--- asterisk/trunk/tests/apps/mixmonitor/run-test (original)
+++ asterisk/trunk/tests/apps/mixmonitor/run-test Thu May 15 11:28:55 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
David Vossel <dvossel at digium.com>
+Matt Jordan <mjordan at digium.com>
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
@@ -10,17 +11,21 @@
import sys
import os
import math
+import logging
+
from twisted.internet import reactor
-from starpy import manager
-sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+sys.path.append("lib/python/asterisk")
+from test_case import TestCase
-workingdir = "apps/mixmonitor"
-testdir = "tests/%s" % workingdir
+testdir = "tests/apps/mixmonitor"
-class MixMonitorTest:
+LOGGER = logging.getLogger(__name__)
+
+class MixMonitorTest(TestCase):
def __init__(self):
+ super(MixMonitorTest, self).__init__()
+
self.passed = False
self.last_step = ""
# playback file is 2559 bytes of ulaw, that will come out to be ~41118 of wav
@@ -30,21 +35,18 @@
self.audiofile2size = -1
self.audiofile3size = -1
- reactor.callWhenRunning(self.run)
-
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
-
self.audiofile1 = os.path.join(os.getcwd(), "%s/testaudio1" % (testdir))
self.audiofile2 = os.path.join(os.getcwd(), "%s/testaudio2" % (testdir))
self.audiofile3 = os.path.join(os.getcwd(), "%s/testaudio3" % (testdir))
self.talkingaudio = os.path.join(os.getcwd(), "%s/sounds/talking" % (testdir))
+ self.create_asterisk()
+
def read_result(self):
self.passed = True
self.stop_reactor()
- self.log_last_step("Reading result file from MixMonitor")
+ LOGGER.info("Reading result file from MixMonitor")
if os.path.exists(self.audiofile1 + ".raw"):
self.audiofile1size = os.path.getsize(self.audiofile1 + ".raw")
if os.path.exists(self.audiofile2 + ".raw"):
@@ -52,54 +54,42 @@
if os.path.exists(self.audiofile3 + ".raw"):
self.audiofile3size = os.path.getsize(self.audiofile3 + ".raw")
- self.log_last_step("audiofile1 size is %d, a negative size indicates the file was not present." % (self.audiofile1size, ))
+ LOGGER.debug("audiofile1 size is %d, a negative size indicates the file was not present." % (self.audiofile1size, ))
if math.fabs(self.audiofile1size - self.expectedfilesize) > self.filesizetolerance:
# if this failed mixmonitor is not creating the correct file size for the time we expect.
- self.log_last_step("audiofile1 size is not within the size tolerance.")
+ LOGGER.error("audiofile1 size is not within the size tolerance.")
self.passed = False
- self.log_last_step("audiofile2 size is %d, a negative size indicates the file was not present." % (self.audiofile2size, ))
+ LOGGER.debug("audiofile2 size is %d, a negative size indicates the file was not present." % (self.audiofile2size, ))
if math.fabs(self.audiofile2size - self.expectedfilesize) > self.filesizetolerance:
# if this failed it is likely because StopMixMonitor never let go of audiofile1
- self.log_last_step("audiofile2 size is not within the size tolerance.")
+ LOGGER.error("audiofile2 size is not within the size tolerance.")
self.passed = False
- self.log_last_step("audiofile3 size is %d, a negative size indicates the file was not present." % (self.audiofile3size, ))
+ LOGGER.debug("audiofile3 size is %d, a negative size indicates the file was not present." % (self.audiofile3size, ))
if self.audiofile3size == -1:
# if this failed it is likely because MixMonitor never let go of audiofile2 on hangup
- self.log_last_step("audiofile3 file does not exist.")
+ LOGGER.error("audiofile3 file does not exist.")
self.passed = False
if self.passed == True:
- self.log_last_step("Test Passed... All audio files are the correct.")
-
- def stop_reactor(self):
- def __finish_stop(result):
- print "Stopping Reactor ..."
- if reactor.running:
- reactor.stop()
- return result
- df = self.ast1.stop()
- df.addCallback(__finish_stop)
+ LOGGER.info("Test Passed... All audio files are the correct.")
def launch_test1(self):
- self.log_last_step("Placing call to test1 exten")
- self.ast1.cli_originate("Local/s at listener extension s at test1")
+ LOGGER.info("Placing call to test1 exten")
+ self.ast[0].cli_originate("Local/s at listener extension s at test1")
def launch_test2(self):
- self.log_last_step("Placing call to test2 exten")
- self.ast1.cli_originate("Local/s at listener extension s at test2")
+ LOGGER.info("Placing call to test2 exten")
+ self.ast[0].cli_originate("Local/s at listener extension s at test2")
- def log_last_step(self, step):
- print step
- self.last_step = step
def check_test1(self, ami, event):
- self.log_last_step("Checking Userevent")
+ LOGGER.info("Checking Userevent")
if event.get("userevent").lower() != "test1":
return
status = event.get("status")
- print ("Status of test1 is %s" % (status))
+ LOGGER.debug("Status of test1 is %s" % (status))
if status != "SUCCESS":
self.stop_reactor()
return
@@ -108,19 +98,18 @@
self.launch_test2()
def check_test2(self, ami, event):
- self.log_last_step("Checking Userevent")
+ LOGGER.info("Checking Userevent")
if event.get("userevent").lower() != "test2":
return
status = event.get("status")
- print ("Status of test2 is %s" % (status))
+ LOGGER.debug("Status of test2 is %s" % (status))
if status != "SUCCESS":
self.stop_reactor()
return
self.read_result()
- def ami_on_connect1(self, ami):
- self.log_last_step("Connected to AMI 1")
+ def ami_connect(self, ami):
self.ami = ami
self.ami.registerEvent("UserEvent", self.check_test1)
@@ -129,26 +118,10 @@
self.ami.setVar(channel = "", variable = "TESTAUDIO3", value = self.audiofile3)
self.ami.setVar(channel = "", variable = "TALK_AUDIO", value = self.talkingaudio)
- def ami_login_error(self, ami):
- self.log_last_step("AMI login failed")
- self.stop_reactor()
-
- def ami_login(self):
- self.log_last_step("Logging in to the AMI")
- self.ami_factory1 = manager.AMIFactory("user", "mysecret")
- self.ami_factory1.login('127.0.0.1', 5038).addCallbacks(self.ami_on_connect1, self.ami_login_error)
+ self.launch_test1()
def run(self):
- def __finish_start_ops(result):
- self.ami_login()
- self.launch_test1()
-
- reactor.callLater(30, self.stop_reactor)
- return result
-
- self.log_last_step("Starting Asterisk")
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
+ self.create_ami_factory()
More information about the svn-commits
mailing list