[svn-commits] mjordan: testsuite/asterisk/trunk r5036 - in /asterisk/trunk/tests/fastagi: e...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Thu May 15 11:25:31 CDT 2014
Author: mjordan
Date: Thu May 15 11:25:27 2014
New Revision: 5036
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=5036
Log:
fastagi: Fix failing tests by upgrading them to use TestCase
A number of FastAGI tests started failing due to failures in installing their
configuration files. This occurred due to a restructuring of the log file
location. Since this is managed by the TestCase class, it was easier to port
them over to that framework rather than reproduce the logic for managing the
location of the test artifacts.
Note that this isn't a full clean up of these tests, merely a port over to
the correct classes.
Modified:
asterisk/trunk/tests/fastagi/execute/run-test
asterisk/trunk/tests/fastagi/get-data/run-test
asterisk/trunk/tests/fastagi/hangup/run-test
asterisk/trunk/tests/fastagi/say-alpha/run-test
asterisk/trunk/tests/fastagi/say-date/run-test
asterisk/trunk/tests/fastagi/say-datetime/run-test
asterisk/trunk/tests/fastagi/say-digits/run-test
asterisk/trunk/tests/fastagi/say-number/run-test
asterisk/trunk/tests/fastagi/say-phonetic/run-test
asterisk/trunk/tests/fastagi/say-time/run-test
Modified: asterisk/trunk/tests/fastagi/execute/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/execute/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/execute/run-test (original)
+++ asterisk/trunk/tests/fastagi/execute/run-test Thu May 15 11:25:27 2014
@@ -1,25 +1,34 @@
#!/usr/bin/env python
-'''
-Copyright (C) 2010, Digium, Inc.
+"""Test AGI EXEC command
+
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland at digium.com>
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
-'''
+"""
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
-
-workingdir = "fastagi/execute"
-testdir = "tests/fastagi"
-
-class FastAGIExecuteTest:
+from asterisk.test_case import TestCase
+
+TESTDIR = "tests/fastagi"
+
+LOGGER = logging.getLogger(__name__)
+
+class FastAGIExecuteTest(TestCase):
+ """Class that manages the test"""
+
def __init__(self):
+ """Constructor"""
+ super(FastAGIExecuteTest, self).__init__()
+
self.passed = {
'SendDTMF': False,
'System': False,
@@ -32,118 +41,139 @@
self.agi = ""
# Listen for results from dialplan
- self.agi_factory = fastagi.FastAGIFactory(self.do_test)
- reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
-
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_fastagi_factory()
+ self.agi = None
+
+ self.create_asterisk(base_configs_path="%s/configs" % TESTDIR)
def on_get_variable2_failure(self, reason):
+ """Failure handler for second variable retrieval"""
self.passed['Set [Channel]'] = False
- print 'Could not test value of channel variable.'
- print reason.getTraceback()
- self.result_changed()
+ LOGGER.error('Could not test value of channel variable.')
+ LOGGER.error(reason.getTraceback())
+ self.read_result()
def on_get_variable2(self, result):
+ """Handler for retrieving second variable"""
if result == "sure to go":
self.passed['Set [Channel]'] = True
- print "Yay. LAMB was", result
+ LOGGER.info("Yay. LAMB was %s" % result)
else:
self.passed['Set [Channel]'] = False
- print "Channel variable LAMB is %s",\
- "but we expected 'sure to go.'" % result
-
- self.result_changed()
-
- def on_Set2_failure(self, reason):
+ LOGGER.error("Channel variable LAMB is %s"
+ " but we expected 'sure to go.'" % result)
+
+ self.read_result()
+
+ def on_set2_failure(self, reason):
+ """Failure handler for second variable setting"""
self.passed['Set [Channel]'] = False
- print 'Could not set channel variable:', reason.getTraceback()
-
- def on_Set2_success(self, result):
+ LOGGER.error('Could not set channel variable:')
+ LOGGER.error(reason.getTraceback())
+
+ def on_set2_success(self, result):
+ """Handler for setting second variable"""
self.agi.getVariable('LAMB').addCallback(
self.on_get_variable2).addErrback(self.on_get_variable2_failure)
self.passed['Set [Channel]'] = True
- # 4. Set(DEBUG=${VAR})
- def do_Set2(self):
+ def do_set2(self):
+ """Set the second variable (LAMB is sure to go)"""
self.agi.execute("Set", "LAMB=sure to go").addCallback(
- self.on_Set2_success).addErrback(self.on_Set2_failure)
+ self.on_set2_success).addErrback(self.on_set2_failure)
def on_get_variable_failure(self, reason):
+ """Failure handler for retrieving variable"""
self.passed['Set [Global]'] = False
- print 'Could not test value of global variable:',reason.getTraceback()
- self.do_Set2()
+ LOGGER.error('Could not test value of global variable:')
+ LOGGER.error(reason.getTraceback())
+ self.do_set2()
def on_get_variable(self, result):
+ """Handler for retrieving variable"""
if result == "white as snow":
self.passed['Set [Global]'] = True
- print "Yay. FLEECE is", result
+ LOGGER.info("Yay. FLEECE is %s" % result)
else:
self.passed['Set [Global]'] = False
- print "Global variable FLEECE is %s",\
- "but we expected 'white as snow.'" % result
-
- self.do_Set2()
-
- def on_Set_failure(self, reason):
+ LOGGER.error("Global variable FLEECE is %s"
+ " but we expected 'white as snow.'" % result)
+
+ self.do_set2()
+
+ def on_set_failure(self, reason):
+ """Failure handler for setting variable"""
self.passed['Set [Global]'] = False
- print 'Could not set global variable:', reason.getTraceback()
-
- def on_Set_success(self, result):
+ LOGGER.error('Could not set global variable:')
+ LOGGER.error(reason.getTraceback())
+
+ def on_set_success(self, result):
+ """Handler for setting variable"""
self.agi.getVariable('FLEECE').addCallback(
self.on_get_variable).addErrback(self.on_get_variable_failure)
self.passed['Set [Global]'] = True
- # 3. Set(GLOBAL(DEBUG)=${VAR})
- def do_Set(self):
+ def do_set(self):
+ """Set global variable (FLEECE is white as snow)"""
self.agi.execute("Set", "GLOBAL(FLEECE)=white as snow").addCallback(
- self.on_Set_success).addErrback(self.on_Set_failure)
-
- def on_System_failure(self, reason):
+ self.on_set_success).addErrback(self.on_set_failure)
+
+ def on_system_failure(self, reason):
+ """Failure handler for running System command"""
self.passed['System'] = False
- print 'Could not execute system command:', reason.getTraceback()
- self.do_Set()
-
- def on_System_success(self, result):
+ LOGGER.error('Could not execute system command:')
+ LOGGER.error(reason.getTraceback())
+ self.do_set()
+
+ def on_system_success(self, result):
+ """Handler for running System command"""
self.passed['System'] = True
- self.do_Set()
-
- # 2. System(command arg1 arg2 etc)
- def do_System(self):
+ self.do_set()
+
+ def do_system(self):
+ """Run a command using System"""
self.agi.execute("System", "echo little lamb").addCallback(
- self.on_System_success).addErrback(self.on_System_failure)
-
- def on_SendDTMF_failure(self, reason):
+ self.on_system_success).addErrback(self.on_system_failure)
+
+ def on_send_dtmf_failure(self, reason):
+ """Failure handler for SendDTMF"""
self.passed['SendDTMF'] = False
- print 'Could not send DTMF:', reason.getTraceback()
- self.do_System()
-
- def on_SendDTMF_success(self, result):
+ LOGGER.error('Could not send DTMF:')
+ LOGGER.error(reason.getTraceback())
+ self.do_system()
+
+ def on_send_dtmf_success(self, result):
+ """Handler for SendDTMF"""
self.passed['SendDTMF'] = True
- self.do_System()
-
- # This gets invoked by the dialplan when the call is answered
- # We're going to use fastagi.execute() to send some commands that
- # FastAGI doesn't inherently support.
- # 1. SendDTMF(digits[,timeout_ms])
- # 2. System(command arg1 arg2 etc)
- # 3. Set(GLOBAL(DEBUG)=${VAR})
- # 4. Set(DEBUG=${VAR})
-
- # 1. SendDTMF(digits[,timeout_ms])
- # Play "Mary had a Little Lamb." We don't really need a timeout here,
- # but let's send one anyway just so one of our tests actually sends
- # an argument.
- def do_test(self, agi):
+ self.do_system()
+
+ def fastagi_connect(self, agi):
+ """Handle a FastAGI connection
+
+ Keyword Arguments:
+ agi The AGI object that just connected
+ """
+ # This gets invoked by the dialplan when the call is answered
+ # We're going to use fastagi.execute() to send some commands that
+ # FastAGI doesn't inherently support.
+ # 1. SendDTMF(digits[,timeout_ms])
+ # 2. System(command arg1 arg2 etc)
+ # 3. Set(GLOBAL(DEBUG)=${VAR})
+ # 4. Set(DEBUG=${VAR})
+
+ # 1. SendDTMF(digits[,timeout_ms])
+ # Play "Mary had a Little Lamb." We don't really need a timeout here,
+ # but let's send one anyway just so one of our tests actually sends
+ # an argument.
+
self.agi = agi
return agi.execute(
"SendDTMF", "3212333", 300, comma_delimiter=True
- ).addCallback(self.on_SendDTMF_success
- ).addErrback(self.on_SendDTMF_failure)
-
- # Read test results and dialplan globals
+ ).addCallback(self.on_send_dtmf_success
+ ).addErrback(self.on_send_dtmf_failure)
+
def read_result(self):
+ """Determine if we passed or failed and kill the AGI"""
self.agi.finish()
self.stop_reactor()
for test in self.passed:
@@ -151,40 +181,18 @@
self.overall_result = False
if self.overall_result is True:
- print "Success"
- else:
- print "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.info("Test succeeded")
+ else:
+ LOGGER.error("Test failed")
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/507 at agitest extension echo at agitest")
-
- # Read result before timeout
- def result_changed(self):
- # If self.passed["SendDTMF"] is False, not one test has completed yet
- # and we should wait for the timeout.
- if self.passed["SendDTMF"] is not False:
- self.read_result()
+ """Start the test"""
+ LOGGER.info("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/507 at agitest extension echo at agitest")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
-
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
+ super(FastAGIExecuteTest, self).run()
+ self.launch_test()
def main():
test = FastAGIExecuteTest()
Modified: asterisk/trunk/tests/fastagi/get-data/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/get-data/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/get-data/run-test (original)
+++ asterisk/trunk/tests/fastagi/get-data/run-test Thu May 15 11:25:27 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland 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.
@@ -9,30 +10,34 @@
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/get-data"
testdir = "tests/fastagi"
-class FastAGIGetDataTest:
+LOGGER = logging.getLogger(__name__)
+
+class FastAGIGetDataTest(TestCase):
def __init__(self):
+ super(FastAGIGetDataTest, self).__init__()
+
self.passed = False
self.timeout = 30
# Listen for results from dialplan
- self.agi_factory = fastagi.FastAGIFactory(self.do_test)
- reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.create_fastagi_factory()
+ self.agi = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def on_failure(self, reason):
- print 'Could not run GET DATA: ', reason.getTraceback()
+ LOGGER.error('Could not run GET DATA: ')
+ LOGGER.error(reason.getTraceback())
self.agi.finish()
# result is (str digits, bool timeout)
@@ -40,63 +45,40 @@
self.passed = True
if result[1] is True:
self.passed = False
- print "GET DATA timed out. This should never, ever happen!"
+ LOGGER.error("GET DATA timed out. This should never, ever happen!")
if result[0] == "3212333":
- print "Got the input we expected"
+ LOGGER.info("Got the input we expected")
else:
- print "Got '%s' which wasn't what we expected." % result[0]
+ LOGGER.error("Got '%s' which wasn't what we expected." % result[0])
self.passed = False
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# send GET DATA command and wait for results
- def do_test(self, agi):
+ def fastagi_connect(self, agi):
self.agi = agi
- print "Connection established."
+ LOGGER.error("Connection established.")
return agi.getData("beep", timeout=10, maxDigits=10).addCallback(
self.finish_test).addErrback(self.on_failure)
- # Read test results and dialplan globals
def read_result(self):
+ self.agi.finish()
self.stop_reactor()
-
if self.passed is True:
- print "Success"
+ LOGGER.info("Test Success")
else:
- print "Failed"
-
- def stop_reactor(self):
- reactor.callLater(1, self.stop_asterisk)
+ LOGGER.error("Failed")
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/502 at agitest extension 1 at get-data")
-
- def stop_asterisk(self):
- def __finish_stop(result):
- if reactor.running:
- reactor.stop()
- return result
- print "Stopping Asterisk"
- df = self.ast1.stop()
- df.addCallback(__finish_stop)
-
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.info("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/502 at agitest extension 1 at get-data")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
+ super(FastAGIGetDataTest, self).run()
+ self.launch_test()
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
def main():
test = FastAGIGetDataTest()
Modified: asterisk/trunk/tests/fastagi/hangup/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/hangup/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/hangup/run-test (original)
+++ asterisk/trunk/tests/fastagi/hangup/run-test Thu May 15 11:25:27 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland 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,63 +11,71 @@
import sys
import os
import time
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/channel-status"
+LOGGER = logging.getLogger(__name__)
+
testdir = "tests/fastagi"
-class FastAGIHangupTest:
+class FastAGIHangupTest(TestCase):
def __init__(self):
+ super(FastAGIHangupTest, self).__init__()
+
self.passed = False
self.timeout = 30
self.test = 1
# Listen for results from dialplan
- self.agi_factory = fastagi.FastAGIFactory(self.do_test)
- reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.create_fastagi_factory()
+ self.agi = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def on_hangup_failure(self, reason):
- print 'Could not hangup:', reason.getTraceback()
+ LOGGER.error('Could not hangup: %s' % reason.getTraceback())
def on_failure(self, reason):
- print "Channel is not active. Hangup succeeded."
+ LOGGER.info("Channel is not active. Hangup succeeded.")
self.passed = True
- self.result_changed()
+ self.read_result()
def on_hangup(self, status):
self.test = 2
- self.do_test(self.agi)
+ self.fastagi_connect(self.agi)
def finish_test(self, status):
- print "Channel is still up! Status is", status, "Maybe the channel is still attempting to hangup..."
- print "Testing for closed channel again."
- time.sleep(5)
- self.agi.channelStatus(agi.variables['agi_channel']
- ).addCallback(self.finish_test2).addErrback(self.on_failure)
+ LOGGER.info("Channel is still up! Status is %s - maybe the channel is "
+ "still attempting to hangup..." % status)
+ LOGGER.info("Testing for closed channel again.")
+
+ def __try_again():
+ """Try finishing again later"""
+ self.agi.channelStatus(agi.variables['agi_channel']
+ ).addCallback(self.finish_test2).addErrback(self.on_failure)
+
+ reactor.callLater(__try_again, 5)
def finish_test2(self, status):
- print "Channel is still up! Status is", status
+ LOGGER.error("Channel is still up! Status is %s" % status)
self.passed = False
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# Hangup via fastagi and test channel status
- def do_test(self, agi):
+ def fastagi_connect(self, agi):
self.agi = agi
if self.test == 1:
- print "Connection established."
+ LOGGER.info("Connection established.")
return agi.hangup().addCallback(self.on_hangup).addErrback(
self.on_hangup_failure)
elif self.test == 2:
- print "Testing for closed channel."
+ LOGGER.info("Testing for closed channel.")
# We expect this call to fail, as channelStatus()
# cannot query a dead channel
return agi.channelStatus(agi.variables['agi_channel']
@@ -76,41 +85,17 @@
self.agi.finish()
self.stop_reactor()
if self.passed is True:
- print "Success"
+ LOGGER.info("Test Success")
else:
- print "Failed"
-
- def stop_reactor(self):
- def __finish_stop(result):
- print "Stopping Reactor ..."
- if reactor.running:
- reactor.stop()
- return result
- def __stop_error(reason):
- logger.warning(reason)
-
- df = self.ast1.stop()
- df.addCallback(__finish_stop)
- df.addErrback(__stop_error)
+ LOGGER.error("Failed")
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/503 at agitest extension echo at agitest")
-
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.info("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/503 at agitest extension echo at agitest")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
-
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
+ super(FastAGIHangupTest, self).run()
+ self.launch_test()
def main():
test = FastAGIHangupTest()
Modified: asterisk/trunk/tests/fastagi/say-alpha/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/say-alpha/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/say-alpha/run-test (original)
+++ asterisk/trunk/tests/fastagi/say-alpha/run-test Thu May 15 11:25:27 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland 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.
@@ -9,18 +10,22 @@
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/say-alpha"
testdir = "tests/fastagi"
+LOGGER = logging.getLogger(__name__)
-class FastAGISayAlphaTest:
+class FastAGISayAlphaTest(TestCase):
def __init__(self):
+ super(FastAGISayAlphaTest, self).__init__()
+
self.passed = False
self.notified = False
self.timeout = 30
@@ -28,72 +33,53 @@
# Listen for results from dialplan
self.agi_factory = fastagi.FastAGIFactory(self.do_test)
reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
+ self.agi = None
# Listen for success or failure of talkdetect
self.agi_factory2 = fastagi.FastAGIFactory(self.listen_result)
reactor.listenTCP(4574, self.agi_factory2, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.agi2 = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def listen_result(self, agi):
- print "Got test success confirmation from dialplan."
+ LOGGER.info("Got test success confirmation from dialplan.")
+ self.agi2 = agi
self.notified = True
def on_failure(self, reason):
- print 'SAY ALPHA failed: ', reason.getTraceback()
+ LOGGER.error('SAY PHONETIC failed: %s' % reason.getTraceback())
self.passed = False
def finish_test(self, result):
- print "AGI command reports success."
+ LOGGER.info("AGI command reports success.")
self.passed = True
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# send SAY ALPHA command and wait for results
def do_test(self, agi):
self.agi = agi
- print "Connection established."
+ LOGGER.debug("Connection established.")
return agi.sayAlpha("respect").addCallback(
self.finish_test).addErrback(self.on_failure)
def read_result(self):
self.agi.finish()
+ if self.passed is True and self.notified is True:
+ self.agi2.finish()
+ LOGGER.info("Test passed")
+ else:
+ LOGGER.error("Test failed")
self.stop_reactor()
- if self.passed is True and self.notified is True:
- print "Success"
- else:
- print "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)
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/510 at agitest extension 1 at td_and_agi_notify")
-
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.debug("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/510 at agitest extension 1 at td_and_agi_notify")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
-
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
+ super(FastAGISayAlphaTest, self).run()
+ self.launch_test()
def main():
test = FastAGISayAlphaTest()
Modified: asterisk/trunk/tests/fastagi/say-date/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/say-date/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/say-date/run-test (original)
+++ asterisk/trunk/tests/fastagi/say-date/run-test Thu May 15 11:25:27 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland 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.
@@ -9,18 +10,21 @@
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/say-date"
testdir = "tests/fastagi"
+LOGGER = logging.getLogger(__name__)
-class FastAGISayDateTest:
+class FastAGISayDateTest(TestCase):
def __init__(self):
+ super(FastAGISayDateTest, self).__init__()
self.passed = False
self.notified = False
self.test = "COMPLETE"
@@ -29,74 +33,54 @@
# Listen for results from dialplan
self.agi_factory = fastagi.FastAGIFactory(self.do_test)
reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
+ self.agi = None
# Listen for success or failure of talkdetect
self.agi_factory2 = fastagi.FastAGIFactory(self.listen_result)
reactor.listenTCP(4574, self.agi_factory2, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.agi2 = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def listen_result(self, agi):
- print "Got test success confirmation from dialplan."
+ LOGGER.info("Got test success confirmation from dialplan.")
+ self.agi2 = agi
self.notified = True
def on_failure(self, reason):
- print 'SAY DATE failed: ', reason.getTraceback()
+ LOGGER.error('SAY DATE failed: %s' % reason.getTraceback())
self.passed = False
def finish_test(self, result):
- print "AGI command reports success."
+ LOGGER.info("AGI command reports success.")
self.passed = True
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# send SAY DATE command and wait for results
def do_test(self, agi):
self.agi = agi
- print "Connection established."
+ LOGGER.debug("Connection established.")
return agi.sayDate("1210498014").addCallback(
self.finish_test).addErrback(self.on_failure)
def read_result(self):
self.agi.finish()
+ if self.passed is True and self.notified is True:
+ self.agi2.finish()
+ LOGGER.info("Test passed")
+ else:
+ LOGGER.error("Test failed")
self.stop_reactor()
- if self.passed is True and self.notified is True:
- print "Success"
- else:
- print "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)
-
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/511 at agitest extension 1 at td_and_agi_notify")
-
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.debug("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/511 at agitest extension 1 at td_and_agi_notify")
+
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
-
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
-
+ super(FastAGISayDateTest, self).run()
+ self.launch_test()
def main():
test = FastAGISayDateTest()
Modified: asterisk/trunk/tests/fastagi/say-datetime/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/say-datetime/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/say-datetime/run-test (original)
+++ asterisk/trunk/tests/fastagi/say-datetime/run-test Thu May 15 11:25:27 2014
@@ -1,7 +1,8 @@
#!/usr/bin/env python
'''
-Copyright (C) 2010, Digium, Inc.
+Copyright (C) 2010-2014, Digium, Inc.
Erin Spiceland <espiceland 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.
@@ -9,17 +10,22 @@
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/say-datetime"
testdir = "tests/fastagi"
-class FastAGISayDateTimeTest:
+LOGGER = logging.getLogger(__name__)
+
+class FastAGISayDateTimeTest(TestCase):
def __init__(self):
+ super(FastAGISayDateTimeTest, self).__init__()
+
self.passed = False
self.notified = False
self.timeout = 30
@@ -27,71 +33,54 @@
# Listen for results from dialplan
self.agi_factory = fastagi.FastAGIFactory(self.do_test)
reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
+ self.agi = None
# Listen for success or failure of talkdetect
self.agi_factory2 = fastagi.FastAGIFactory(self.listen_result)
reactor.listenTCP(4574, self.agi_factory2, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.agi2 = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def listen_result(self, agi):
- print "Got test success confirmation from dialplan."
+ LOGGER.info("Got test success confirmation from dialplan.")
+ self.agi2 = agi
self.notified = True
def on_failure(self, reason):
- print 'SAY DATETIME failed: ', reason.getTraceback()
+ LOGGER.error('SAY PHONETIC failed: %s' % reason.getTraceback())
self.passed = False
def finish_test(self, result):
- print "AGI command reports success."
+ LOGGER.info("AGI command reports success.")
self.passed = True
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# send SAY DATETIME command and wait for results
def do_test(self, agi):
self.agi = agi
- print "Connection established."
+ LOGGER.debug("Connection established.")
return agi.sayDateTime("1210498014").addCallback(
self.finish_test).addErrback(self.on_failure)
def read_result(self):
self.agi.finish()
+ if self.passed is True and self.notified is True:
+ self.agi2.finish()
+ LOGGER.info("Test passed")
+ else:
+ LOGGER.error("Test failed")
self.stop_reactor()
- if self.passed is True and self.notified is True:
- print "Success"
- else:
- print "Failure"
-
- 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)
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/512 at agitest extension 1 at td_and_agi_notify")
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.debug("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/516 at agitest extension 1 at td_and_agi_notify")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
- reactor.callLater(self.timeout, self.stop_reactor)
- return result
+ super(FastAGISayDateTimeTest, self).run()
+ self.launch_test()
- print "Starting Asterisk"
- df = self.ast1.start()
- df.addCallback(__finish_start_ops)
def main():
test = FastAGISayDateTimeTest()
Modified: asterisk/trunk/tests/fastagi/say-digits/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/fastagi/say-digits/run-test?view=diff&rev=5036&r1=5035&r2=5036
==============================================================================
--- asterisk/trunk/tests/fastagi/say-digits/run-test (original)
+++ asterisk/trunk/tests/fastagi/say-digits/run-test Thu May 15 11:25:27 2014
@@ -9,18 +9,21 @@
import sys
import os
+import logging
+
from twisted.internet import reactor
from starpy import fastagi
sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
+from asterisk.test_case import TestCase
-workingdir = "fastagi/say-digits"
testdir = "tests/fastagi"
+LOGGER = logging.getLogger(__name__)
-class FastAGISayDigitsTest:
+class FastAGISayDigitsTest(TestCase):
def __init__(self):
+ super(FastAGISayDigitsTest, self).__init__()
self.passed = False
self.notified = False
self.timeout = 30
@@ -28,72 +31,53 @@
# Listen for results from dialplan
self.agi_factory = fastagi.FastAGIFactory(self.do_test)
reactor.listenTCP(4573, self.agi_factory, self.timeout, '127.0.0.1')
+ self.agi = None
# Listen for success or failure of talkdetect
self.agi_factory2 = fastagi.FastAGIFactory(self.listen_result)
reactor.listenTCP(4574, self.agi_factory2, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ self.agi2 = None
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
+ self.create_asterisk(base_configs_path="%s/configs" % testdir)
def listen_result(self, agi):
- print "Got test success confirmation from dialplan."
+ LOGGER.info("Got test success confirmation from dialplan.")
+ self.agi2 = agi
self.notified = True
def on_failure(self, reason):
- print 'SAY DIGITS failed: ', reason.getTraceback()
+ LOGGER.error('SAY DIGITS failed: %s' % reason.getTraceback())
self.passed = False
def finish_test(self, result):
- print "AGI command reports success."
+ LOGGER.info("AGI command reports success.")
self.passed = True
- self.result_changed()
+ self.read_result()
# This gets invoked by the dialplan when the call is answered
# send SAY DIGITS command and wait for results
def do_test(self, agi):
self.agi = agi
- print "Connection established."
+ LOGGER.debug("Connection established.")
return agi.sayDigits("31337").addCallback(
self.finish_test).addErrback(self.on_failure)
def read_result(self):
self.agi.finish()
+ if self.passed is True and self.notified is True:
+ self.agi2.finish()
+ LOGGER.info("Test passed")
+ else:
+ LOGGER.error("Test failed")
self.stop_reactor()
- if self.passed is True and self.notified is True:
- print "Success"
- else:
- print "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)
def launch_test(self):
- print "Originating call to begin test."
- self.ast1.cli_originate("Local/514 at agitest extension 1 at td_and_agi_notify")
-
- # Read result before timeout
- def result_changed(self):
- if self.passed is True:
- self.read_result()
+ LOGGER.debug("Originating call to begin test.")
+ self.ast[0].cli_originate("Local/514 at agitest extension 1 at td_and_agi_notify")
def run(self):
- def __finish_start_ops(result):
- self.launch_test()
[... 406 lines stripped ...]
More information about the svn-commits
mailing list