[asterisk-commits] pabelanger: testsuite/asterisk/trunk r1555 - in /asterisk/trunk/tests: callpa...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Fri May 20 08:35:47 CDT 2011


Author: pabelanger
Date: Fri May 20 08:35:42 2011
New Revision: 1555

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=1555
Log:
Convert callparking test to use TestCase class.

Modified:
    asterisk/trunk/tests/callparking/run-test
    asterisk/trunk/tests/udptl/run-test

Modified: asterisk/trunk/tests/callparking/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/callparking/run-test?view=diff&rev=1555&r1=1554&r2=1555
==============================================================================
--- asterisk/trunk/tests/callparking/run-test (original)
+++ asterisk/trunk/tests/callparking/run-test Fri May 20 08:35:42 2011
@@ -8,83 +8,40 @@
 '''
 
 import sys
-import os
-import math
-from optparse import OptionParser
-from twisted.application import service, internet
 from twisted.internet import reactor
-from starpy import manager
 
 sys.path.append("lib/python")
 from asterisk.asterisk import Asterisk
-from asterisk.version import AsteriskVersion
-
-workingdir = "/tmp/asterisk-testsuite/callparking"
-testdir = "tests/callparking"
+from asterisk.TestCase import TestCase
 
 
-class CallParkingTest:
+class CallParkingTest(TestCase):
     def __init__(self, argv):
-        self.passed = False
+        TestCase.__init__(self, argv)
         self.ast1_event = False
         self.ast2_event = False
 
-        # Test timeout in seconds
-        self.timeout = 30
-        self.last_step = ""
-
-        # get version info
-        parser = OptionParser()
-        parser.add_option("-v", "--version", dest="ast_version",
-         help="Asterisk version string")
-        parser.add_option("-n", dest="test_name",
-                help="Test name")
-
-        (options, args) = parser.parse_args(argv)
-        self.ast_version = AsteriskVersion(options.ast_version)
-
-        reactor.callWhenRunning(self.run)
-
-        self.ast1 = Asterisk(base=workingdir)
-        self.ast1.install_configs("%s/configs/ast1" % (testdir))
-
-        self.ast2 = Asterisk(base=workingdir)
-        self.ast2.install_configs("%s/configs/ast2" % (testdir))
-
-    def stop_reactor(self):
-        print "Stopping Reactor ..."
-        if reactor.running:
-            reactor.stop()
+        self.create_asterisk(2)
 
     def check_result(self):
-        self.log_last_step("Checking results...")
         if self.ast1_event is True and self.ast2_event is True:
             self.stop_reactor()
             self.passed = True
 
     def get_parkedcall(self):
-        self.ast2.cli_originate("IAX2/ast1/701 extension s at ast1")
-
-    def launch_test(self):
-        self.ast1.cli_originate("Local/9876 at ast2/n extension 101 at ast2")
+        self.ast[1].cli_originate("IAX2/ast1/701 extension s at ast1")
 
     def ast1_unparkedcall(self, ami, event):
-        self.log_last_step("UnParkedCall event on ast1 ...")
         if event.get("exten").lower() != "701":
             return
-        self.log_last_step("... Success")
 
     def ast1_parkedcall(self, ami, event):
-        self.log_last_step("ParkedCall event on ast1 ...")
         if event.get("exten").lower() != "701":
             return
-        self.log_last_step("... Success")
         self.get_parkedcall()
 
     def ast2_userevent(self, ami, event):
-        self.log_last_step("UserEvent on ast2 ... ")
         status = event.get("status")
-        print ("... %s" % (status))
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -93,9 +50,7 @@
         self.check_result()
 
     def ast1_userevent(self, ami, event):
-        self.log_last_step("UserEvent on ast1 ... ")
         status = event.get("status")
-        print ("... %s" % (status))
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -103,48 +58,20 @@
         self.ast1_event = True
         self.check_result()
 
-    def ami_on_connect2(self, ami):
-        self.log_last_step("Connected to AMI 2")
-        self.ami2 = ami
-        self.ami2.registerEvent("UserEvent", self.ast2_userevent)
+    def ami_connect(self, ami):
+        TestCase.ami_connect(self, ami)
 
-    def ami_on_connect1(self, ami):
-        self.log_last_step("Connected to AMI 1")
-        self.ami1 = ami
-        self.ami1.registerEvent("ParkedCall", self.ast1_parkedcall)
-        self.ami1.registerEvent("UnParkedCall", self.ast1_unparkedcall)
-        self.ami1.registerEvent("UserEvent", self.ast1_userevent)
-
-    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_factory2 = manager.AMIFactory("user", "mysecret")
-        self.ami_factory1.login('127.0.0.1', 5038).addCallbacks(self.ami_on_connect1, self.ami_login_error)
-        self.ami_factory2.login('127.0.0.2', 5038).addCallbacks(self.ami_on_connect2, self.ami_login_error)
-
-    def start_asterisk(self):
-        self.log_last_step("Starting Asterisk")
-        self.ast1.start()
-        self.ast2.start()
-
-    def stop_asterisk(self):
-        self.ast1.stop()
-        self.ast2.stop()
-
-    def log_last_step(self, step):
-        print step
-        self.last_step = step
+        if ami.id == 0:
+            self.ami[ami.id].registerEvent("ParkedCall", self.ast1_parkedcall)
+            self.ami[ami.id].registerEvent("UnParkedCall", self.ast1_unparkedcall)
+            self.ami[ami.id].registerEvent("UserEvent", self.ast1_userevent)
+            self.ast[ami.id].cli_originate("Local/9876 at ast2/n extension 101 at ast2")
+        elif ami.id == 1:
+            self.ami[ami.id].registerEvent("UserEvent", self.ast2_userevent)
 
     def run(self):
-        self.ami_login()
-        self.launch_test()
-
-        # stop and read results after timeout
-        reactor.callLater(self.timeout, self.stop_reactor)
+        TestCase.run(self)
+        self.create_ami_factory(2)
 
 
 def main(argv=None):

Modified: asterisk/trunk/tests/udptl/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/udptl/run-test?view=diff&rev=1555&r1=1554&r2=1555
==============================================================================
--- asterisk/trunk/tests/udptl/run-test (original)
+++ asterisk/trunk/tests/udptl/run-test Fri May 20 08:35:42 2011
@@ -45,12 +45,9 @@
             self.passed = False
             print "UDPTL test failed!"
 
-    def ami_login(self):
-        self.create_ami_factory(2)
-
     def run(self):
         TestCase.run(self)
-        self.ami_login()
+        self.create_ami_factory(2)
 
 
 def main(argv=None):




More information about the asterisk-commits mailing list