[asterisk-commits] pabelanger: branch pabelanger/testClass r1120 - in /asterisk/team/pabelanger/...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Wed Dec 8 20:46:52 CST 2010


Author: pabelanger
Date: Wed Dec  8 20:46:48 2010
New Revision: 1120

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=1120
Log:
Move AMIFactory into TestCase :D

Modified:
    asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py
    asterisk/team/pabelanger/testClass/tests/manager/login/run-test
    asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf
    asterisk/team/pabelanger/testClass/tests/udptl/run-test

Modified: asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py (original)
+++ asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py Wed Dec  8 20:46:48 2010
@@ -10,6 +10,7 @@
 import sys
 from optparse import OptionParser
 from twisted.internet import reactor
+from starpy import manager
 
 from asterisk import Asterisk
 
@@ -38,6 +39,12 @@
         reactor.callWhenRunning(self.run)
 
     def create_asterisk(self, count=1):
+        """
+
+        Keywork arguments:
+        count --
+
+        """
         for c in range(count):
             print "Creating Asterisk instance %d ..." % (c + 1)
             self.ast.append(Asterisk(base=self.options.base))
@@ -48,22 +55,55 @@
             self.ast[c].install_configs("%s/configs/ast%d" %
                     (self.options.test_name, c + 1))
 
+    def create_ami_factory(self, count=1, username=None, secret=None, port=5038):
+        """
+
+        Keywork arguments:
+        count --
+        username --
+        secret --
+        port --
+
+        """
+        for c in range(count):
+            host = "127.0.0.%d" % (c + 1)
+            print "Creating AMIFactory %d ..." % (c + 1)
+            ami_method = getattr(self, "ami_connect%d" % (c + 1))
+            self.ami_factory = manager.AMIFactory(username, secret)
+            self.ami_factory.login(host).addCallbacks(ami_method,
+                    self.ami_login_error)
+
     def start_asterisk(self):
+        """
+
+        """
         for index, item in enumerate(self.ast):
             print "Starting Asterisk instance %d ..." % (index + 1)
             self.ast[index].start()
 
     def stop_asterisk(self):
+        """
+
+        """
         for index, item in enumerate(self.ast):
             print "Stopping Asterisk instance %d ..." % (index + 1)
             self.ast[index].stop()
 
     def stop_reactor(self):
+        """
+
+        """
         print "Stopping Reactor ..."
         if reactor.running:
             reactor.stop()
 
     def run(self):
+        """
+
+        """
         reactor.callLater(self.reactor_timeout, self.stop_reactor)
 
+    def ami_login_error(self):
+        print "Error logging into AMI"
+        self.stop_reactor()
 

Modified: asterisk/team/pabelanger/testClass/tests/manager/login/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/manager/login/run-test?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/manager/login/run-test (original)
+++ asterisk/team/pabelanger/testClass/tests/manager/login/run-test Wed Dec  8 20:46:48 2010
@@ -9,7 +9,7 @@
 
 import sys
 from twisted.internet import reactor
-from starpy import manager
+
 
 sys.path.append("lib/python")
 from asterisk.asterisk import Asterisk
@@ -17,38 +17,20 @@
 
 class AMILoginTest(TestCase):
     def __init__(self, argv):
-        self.last_step = ""
-        self.ami = None
-
         TestCase.__init__(self, argv)
         self.create_asterisk()
 
-    def log_last_step(self, step):
-        print step
-        self.last_step = step
-
-    def on_error(self, ami):
-        print "ERROR, Last Step: %s" % self.last_step
-        self.stop_reactor()
-
-    def on_logoff(self, ami):
-        self.log_last_step("Logoff Successful")
+    def ami_logoff(self, ami):
         self.passed = True
         self.stop_reactor()
 
-    def on_connect(self, ami):
-        self.log_last_step("Connected to the AMI")
-        self.ami = ami
-        self.ami.logoff().addCallbacks(self.on_logoff, self.on_error)
-        self.log_last_step("Logging off")
+    def ami_connect1(self, ami):
+        print "AMI Connect"
+        ami.logoff().addCallbacks(self.ami_logoff, self.ami_login_error)
 
     def run(self):
         TestCase.run(self)
-
-        self.log_last_step("Logging in to the AMI")
-        self.ami_factory = manager.AMIFactory("user", "mysecret")
-        self.ami_factory.login('127.0.0.1', 5038).addCallbacks(self.on_connect,
-                                                               self.on_error)
+        self.create_ami_factory(1, "user", "mysecret")
 
 def main(argv=None):
     if argv is None:

Modified: asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf (original)
+++ asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf Wed Dec  8 20:46:48 2010
@@ -1,7 +1,7 @@
 [general]
 enabled = yes
-port = 5039
-bindaddr = 127.0.0.1
+port = 5038
+bindaddr = 127.0.0.2
 
 [user]
 secret = mysecret

Modified: asterisk/team/pabelanger/testClass/tests/udptl/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/udptl/run-test?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/udptl/run-test (original)
+++ asterisk/team/pabelanger/testClass/tests/udptl/run-test Wed Dec  8 20:46:48 2010
@@ -17,21 +17,18 @@
 
 class UdptlTest(TestCase):
     def __init__(self, argv):
-        self.last_step = ""
         self.reactor_timeout = 60
 
         TestCase.__init__(self, argv)
         self.create_asterisk(2)
 
-    def ami_on_connect1(self, ami):
-        self.log_last_step("Connected to AMI 1")
-        self.ami1 = ami
-        self.ami1.originate(channel = "SIP/666 at receiver",
+    def ami_connect1(self, ami):
+        print "AMI Connect 1"
+        ami.originate(channel = "SIP/666 at receiver",
                             application = "SendFax",
                             data = "tests/udptl/1page.tif")
 
     def check_recv_fax_result(self, ami, event):
-        self.log_last_step("Checking Userevent")
         if event.get("userevent").lower() != "recvstatus":
             return
         self.stop_reactor()
@@ -45,25 +42,13 @@
             self.passed = False
             print "UDPTL test failed!"
 
-    def ami_on_connect2(self, ami):
-        self.log_last_step("Connected to AMI 2")
+    def ami_connect2(self, ami):
+        print "AMI Connect 2"
         self.ami2 = ami
         self.ami2.registerEvent("UserEvent", self.check_recv_fax_result)
 
-    def ami_login_error(self, ami):
-        self.log_last_step("AMI login failed")
-        reactor.callLater(1, 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.1', 5039).addCallbacks(self.ami_on_connect2, self.ami_login_error)
-
-    def log_last_step(self, step):
-        print step
-        self.lastStep = step
+        self.create_ami_factory(2, "user", "mysecret")
 
     def run(self):
         TestCase.run(self)




More information about the asterisk-commits mailing list