[asterisk-commits] mnick: branch mnick/gsoc2010 r730 - /asterisk/team/mnick/gsoc2010/trunk/tests...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Mon Aug 16 14:03:30 CDT 2010
Author: mnick
Date: Mon Aug 16 14:03:26 2010
New Revision: 730
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=730
Log:
added p8
Modified:
asterisk/team/mnick/gsoc2010/trunk/tests/one-step-parking/run-test
Modified: asterisk/team/mnick/gsoc2010/trunk/tests/one-step-parking/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mnick/gsoc2010/trunk/tests/one-step-parking/run-test?view=diff&rev=730&r1=729&r2=730
==============================================================================
--- asterisk/team/mnick/gsoc2010/trunk/tests/one-step-parking/run-test (original)
+++ asterisk/team/mnick/gsoc2010/trunk/tests/one-step-parking/run-test Mon Aug 16 14:03:26 2010
@@ -19,6 +19,7 @@
from asterisk.asterisk import Asterisk
from asterisk.version import AsteriskVersion
+
class OneStepParkingTest:
def __init__(self, argv):
self.passed = False
@@ -30,39 +31,43 @@
# 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)
# FastAGI, listen for results from dialplan
- self.agi_dial_a = fastagi.FastAGIFactory(self.get_result_dial_a)
+ self.agi_dial_a = fastagi.FastAGIFactory(self.get_result_dial_a)
self.agi_b = fastagi.FastAGIFactory(self.get_result_b)
- reactor.listenTCP(4600, self.agi_dial_a, self.timeout, '127.0.0.1') # retrieve userA (Dial 701)
- reactor.listenTCP(4574, self.agi_b, self.timeout, '127.0.0.1') # success
+ reactor.listenTCP(4600, self.agi_dial_a, \
+ self.timeout, '127.0.0.1') # retrieve userA (Dial 701)
+ reactor.listenTCP(4574, self.agi_b, \
+ self.timeout, '127.0.0.1') # success
reactor.callWhenRunning(self.run)
print self.ast_version
- self.asterisk1 = Asterisk(base="/tmp/asterisk-testsuite/one-step-parking")
- self.asterisk1.install_config("tests/one-step-parking/configs/extensions.conf")
- self.asterisk1.install_config("tests/one-step-parking/configs/iax.conf")
- self.asterisk1.install_config("tests/one-step-parking/configs/features.conf")
+ self.asterisk1 = Asterisk( \
+ base="/tmp/asterisk-testsuite/one-step-parking")
+ self.asterisk1.install_configs("tests/one-step-parking/configs")
- self.asterisk2 = Asterisk(base="/tmp/asterisk-testsuite/one-step-parking")
- self.asterisk2.install_config("tests/one-step-parking/configs2/extensions.conf")
- self.asterisk2.install_config("tests/one-step-parking/configs2/iax.conf")
+ self.asterisk2 = Asterisk( \
+ base="/tmp/asterisk-testsuite/one-step-parking")
+ self.asterisk2.install_configs("tests/one-step-parking/configs2")
- # This gets invoked by the dialplan when userA gets parked (701) and userB is trying to retrieve userA by dialing 701
+ # This gets invoked by the dialplan when userA gets parked (701)
+ # and userB is trying to retrieve userA by dialing 701
def get_result_dial_a(self, agi):
self.log_last_step("Dial userA (701)")
- self.asterisk2.cli_originate("IAX2/userA at 127.0.0.1:4570/701 extension retrieve_A at parking")
+ self.asterisk2.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 test succeeds.
def get_result_b(self, agi):
self.log_last_step("got AGI connection from userB -> test successful")
self.passed = True
- reactor.callLater(2,self.read_result)
+ reactor.callLater(2, self.read_result)
return self.finish_agi(agi)
# Finish the AGI connection
@@ -73,27 +78,30 @@
def read_result(self):
if self.done:
- return
+ return
self.done = True
self.log_last_step("Reading results")
- self.asterisk1.cli_exec("core show locks") # get lock output in case of deadlock before tearing down.
- self.asterisk2.cli_exec("core show locks") # get lock output in case of deadlock before tearing down.
- self.asterisk1.cli_exec("core show channels")# if channels are still up for some reason, we want to know that as well
- self.asterisk2.cli_exec("core show channels")# if channels are still up for some reason, we want to know that as well
+ # get lock output in case of deadlock before tearing down.
+ self.asterisk1.cli_exec("core show locks")
+ self.asterisk2.cli_exec("core show locks")
+
+ # 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.stop_asterisk()
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 ..."
reactor.stop()
-
# This is a one step parking test
# 1. A calls B
@@ -105,7 +113,8 @@
self.asterisk1.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 a_dials at parking")
+ self.asterisk1.cli_originate("IAX2/userA at 127.0.0.1:4571/b_exten \
+ extension a_dials at parking")
def start_asterisk(self):
self.log_last_step("Starting Asterisk")
@@ -129,6 +138,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
@@ -142,4 +152,3 @@
if __name__ == "__main__":
sys.exit(main() or 0)
-
More information about the asterisk-commits
mailing list