[svn-commits] mnick: branch mnick/gsoc2010 r720 - /asterisk/team/mnick/gsoc2010/trunk/tests...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Sun Aug 15 16:28:22 CDT 2010
Author: mnick
Date: Sun Aug 15 16:28:18 2010
New Revision: 720
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=720
Log:
install_configs and pep8
Modified:
asterisk/team/mnick/gsoc2010/trunk/tests/callparking/run-test
Modified: asterisk/team/mnick/gsoc2010/trunk/tests/callparking/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mnick/gsoc2010/trunk/tests/callparking/run-test?view=diff&rev=720&r1=719&r2=720
==============================================================================
--- asterisk/team/mnick/gsoc2010/trunk/tests/callparking/run-test (original)
+++ asterisk/team/mnick/gsoc2010/trunk/tests/callparking/run-test Sun Aug 15 16:28:18 2010
@@ -19,6 +19,7 @@
from asterisk.asterisk import Asterisk
from asterisk.version import AsteriskVersion
+
class CallParkingTest:
def __init__(self, argv):
self.passed = False
@@ -30,7 +31,8 @@
# get version info
parser = OptionParser()
- parser.add_option("-v", "--version", dest="ast_version", help="Asterisk version string")
+ parser.add_option("-v", "--version", dest="ast_version", \
+ help="Asterisk version string")
(options, args) = parser.parse_args(argv)
self.ast_version = AsteriskVersion(options.ast_version)
print self.ast_version
@@ -40,55 +42,65 @@
reactor.listenTCP(4574, self.agi_b, self.timeout, '127.0.0.1')
self.agi_c = fastagi.FastAGIFactory(self.get_result_c)
- reactor.listenTCP(4575, self.agi_c, self.timeout, '127.0.0.1')
+ reactor.listenTCP(4575, self.agi_c, self.timeout, '127.0.0.1')
- self.agi_retrieve_A = fastagi.FastAGIFactory(self.get_result_retrieve_A)
- reactor.listenTCP(4576, self.agi_retrieve_A, self.timeout, '127.0.0.1')
+ self.agi_retrieve_A = fastagi.FastAGIFactory( \
+ self.get_result_retrieve_A)
+ reactor.listenTCP(4576, self.agi_retrieve_A, self.timeout, '127.0.0.1')
- self.agi_retrieve_B = fastagi.FastAGIFactory(self.get_result_retrieve_B)
- reactor.listenTCP(4577, self.agi_retrieve_B, self.timeout, '127.0.0.1')
+ self.agi_retrieve_B = fastagi.FastAGIFactory( \
+ self.get_result_retrieve_B)
+ reactor.listenTCP(4577, self.agi_retrieve_B, self.timeout, '127.0.0.1')
- reactor.callWhenRunning(self.run)
+ reactor.callWhenRunning(self.run)
- self.asterisk1 = Asterisk(base="/tmp/asterisk-testsuite/callparking/asterisk1")
- self.asterisk1.install_config("tests/callparking/configs/extensions.conf")
- self.asterisk1.install_config("tests/callparking/configs/iax.conf")
- self.asterisk1.install_config("tests/callparking/configs/logger.conf")
+ self.asterisk1 = Asterisk( \
+ base="/tmp/asterisk-testsuite/callparking/asterisk1")
+ self.asterisk1.install_configs("tests/callparking/configs")
- self.asterisk2 = Asterisk(base="/tmp/asterisk-testsuite/callparking/asterisk2")
- self.asterisk2.install_config("tests/callparking/configs2/extensions.conf")
- self.asterisk2.install_config("tests/callparking/configs2/iax.conf")
- self.asterisk2.install_config("tests/callparking/configs2/logger.conf")
+ self.asterisk2 = Asterisk( \
+ base="/tmp/asterisk-testsuite/callparking/asterisk2")
+ self.asterisk2.install_configs("tests/callparking/configs2")
- self.asterisk3 = Asterisk(base="/tmp/asterisk-testsuite/callparking/asterisk3")
- self.asterisk3.install_config("tests/callparking/configs3/extensions.conf")
- self.asterisk3.install_config("tests/callparking/configs3/iax.conf")
- self.asterisk3.install_config("tests/callparking/configs3/logger.conf")
+ self.asterisk3 = Asterisk( \
+ base="/tmp/asterisk-testsuite/callparking/asterisk3")
+ self.asterisk3.install_configs("tests/callparking/configs3")
- # This gets invoked by the dialplan when userB has parked userA to 701. Initiate new call userB <--> userC now.
+ # This gets invoked by the dialplan when userB has parked userA to 701.
+ # Initiate new call userB <--> userC now.
def get_result_b(self, agi):
- self.log_last_step("got AGI connection from userB. Initiate new call between userB <--> userC now")
- self.asterisk1.cli_originate("IAX2/userB at 127.0.0.1:4572/c_exten extension bridge_to_B at parking")
+ self.log_last_step("got AGI connection from userB. \
+ Initiate new call between userB and userC now")
+ self.asterisk1.cli_originate("IAX2/userB at 127.0.0.1:4572/c_exten \
+ extension bridge_to_B at parking")
return self.finish_agi(agi)
- # This gets invoked by the dialplan when userC has parked userB to 702. Initiate new call between userC and parked userA (@701) now.
+ # This gets invoked by the dialplan when userC has parked userB to 702.
+ # Initiate new call between userC and parked userA (@701) now.
def get_result_c(self, agi):
- self.log_last_step("got AGI connection from userC. Now retrieve userA -> call 701")
- self.asterisk3.cli_originate("IAX2/userA at 127.0.0.1:4570/701 extension retrieve_A at parking")
- return self.finish_agi(agi)
+ self.log_last_step("got AGI connection from userC. \
+ Now retrieve userA -> call 701")
+ self.asterisk3.cli_originate("IAX2/userA at 127.0.0.1:4570/701 \
+ extension retrieve_A at parking")
+ return self.finish_agi(agi)
- # This gets invoked by the dialplan when userC has retrieved userA successfully. Initiate new call between userC and parked userB (@702) now.
+ # This gets invoked by the dialplan when userC has retrieved userA.
+ # Initiate new call between userC and parked userB (@702) now.
def get_result_retrieve_A(self, agi):
- self.log_last_step("got AGI connection from userC. userA retrieved successfully, now retrieve userB")
- self.asterisk3.cli_originate("IAX2/userA at 127.0.0.1:4570/702 extension retrieve_B at parking")
- return self.finish_agi(agi)
+ self.log_last_step("got AGI connection from userC. \
+ userA retrieved successfully, now retrieve userB")
+ self.asterisk3.cli_originate("IAX2/userA at 127.0.0.1:4570/702 \
+ extension retrieve_B at parking")
+ return self.finish_agi(agi)
- # This gets invoked by the dialplan when userC has retrieved userB successfully. Test successful!
+ # This gets invoked by the dialplan when userC has retrieved userB.
+ # Test successful!
def get_result_retrieve_B(self, agi):
- self.log_last_step("got AGI connection from userC. userB retrieved successfully -> Test successful")
- self.passed = True
- reactor.callLater(2, self.read_result)
- return self.finish_agi(agi)
+ self.log_last_step("got AGI connection from userC. \
+ userB retrieved successfully -> Test successful")
+ self.passed = True
+ reactor.callLater(2, self.read_result)
+ return self.finish_agi(agi)
# Finish the AGI connection
def finish_agi(self, agi):
@@ -103,18 +115,21 @@
self.log_last_step("Reading results")
- self.asterisk1.cli_exec("core show locks") # get lock output in case of deadlock before tearing down.
+ # get lock output in case of deadlock before tearing down.
+ self.asterisk1.cli_exec("core show locks")
self.asterisk2.cli_exec("core show locks")
self.asterisk3.cli_exec("core show locks")
- self.asterisk1.cli_exec("core show channels") # if channels are still up for some reason, we want to know that as well
+ # if channels are still up for some reason, we want to know that
+ self.asterisk1.cli_exec("core show channels")
self.asterisk2.cli_exec("core show channels")
self.asterisk3.cli_exec("core show channels")
if (self.passed):
self.log_last_step("Test Passed...")
else:
- self.log_last_step("Test Failed... view result of 'core show globals' in log for more detailed failure results.")
+ self.log_last_step("Test Failed... view result of \
+ 'core show globals' in log for more detailed failure results.")
if reactor.running:
print "Stopping Reactor ..."
@@ -134,7 +149,8 @@
self.asterisk3.cli_exec("core set verbose 10")
self.log_last_step("Originating call to begin test")
- self.asterisk1.cli_originate("IAX2/userA at 127.0.0.1:4571/b_exten extension bridge_to_A at parking")
+ self.asterisk1.cli_originate("IAX2/userA at 127.0.0.1:4571/b_exten \
+ extension bridge_to_A at parking")
def start_asterisk(self):
self.log_last_step("Starting Asterisk")
@@ -160,6 +176,7 @@
# stop and read results after timeout
reactor.callLater(self.timeout, self.read_result)
+
def main(argv=None):
if argv is None:
argv = sys.argv
More information about the svn-commits
mailing list