[asterisk-commits] rmudgett: branch rmudgett/parking r1985 - /asterisk/team/rmudgett/parking/tes...

[SVN commits to the Asterisk project]

Author: rmudgett
Date: Fri Aug 26 16:55:09 2011
New Revision: 1985

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=1985
Log:
Update blind transfer to parking test.

* Added some helpful comments and test progress print statements.

* Made wait until both AMI connections are established before starting
test.

* Made require seeing the AMI UnParkedCall event to pass test.

* Set passed flag before stopping the reactor in case that is significant.

Modified:
    asterisk/team/rmudgett/parking/tests/callparking/run-test

Modified: asterisk/team/rmudgett/parking/tests/callparking/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/rmudgett/parking/tests/callparking/run-test?view=diff&rev=1985&r1=1984&r2=1985
==============================================================================
--- asterisk/team/rmudgett/parking/tests/callparking/run-test (original)
+++ asterisk/team/rmudgett/parking/tests/callparking/run-test Fri Aug 26 16:55:09 2011
@@ -11,37 +11,58 @@
 from twisted.internet import reactor
 
 sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
 from asterisk.TestCase import TestCase
 
 
 class CallParkingTest(TestCase):
     def __init__(self):
         TestCase.__init__(self)
+
+        self.ami_count = 0
+        self.got_AMI_UnParkedCall = False
         self.ast1_event = False
         self.ast2_event = False
 
+        # Create two Asterisk instances ...
         self.create_asterisk(2)
 
     def check_result(self):
-        if self.ast1_event is True and self.ast2_event is True:
+        if self.got_AMI_UnParkedCall is TRUE and self.ast1_event is True and self.ast2_event is True:
+            self.passed = True
             self.stop_reactor()
-            self.passed = True
 
+    # Initiate a call to retrieve the parked call.
     def get_parkedcall(self):
+        print "Fetching parked call"
         self.ast[1].cli_originate("IAX2/ast1/701 extension s at ast1")
 
+    # Called when got UnParkedCall AMI event
     def ast1_unparkedcall(self, ami, event):
-        if event.get("exten").lower() != "701":
+        exten = event.get("exten").lower()
+        print "Parked call retrieved from exten:%s" % (exten)
+        if exten != "701":
+            print "Call not retrieved from expected space"
+            self.stop_reactor()
             return
 
+        self.got_AMI_UnParkedCall = True
+        self.check_result()
+
+    # Called when got ParkedCall AMI event
     def ast1_parkedcall(self, ami, event):
-        if event.get("exten").lower() != "701":
+        exten = event.get("exten").lower()
+        print "Call parked at exten:%s" % (exten)
+        if exten != "701":
+            print "Call not parked in expected space"
+            self.stop_reactor()
             return
+
         self.get_parkedcall()
 
+    # Called when got UserEvent AMI event from Ast2
     def ast2_userevent(self, ami, event):
         status = event.get("status")
+        print "ast2_userevent status:%s" % (status)
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -49,8 +70,10 @@
         self.ast2_event = True
         self.check_result()
 
+    # Called when got UserEvent AMI event from Ast1
     def ast1_userevent(self, ami, event):
         status = event.get("status")
+        print "ast1_userevent status:%s" % (status)
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -58,17 +81,25 @@
         self.ast1_event = True
         self.check_result()
 
+    # This is called for each AMI connection established.
     def ami_connect(self, ami):
         TestCase.ami_connect(self, ami)
 
+        # Add AMI event triggers
         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)
 
+        # Initiate test when both AMI connections are established.
+        self.ami_count += 1
+        if self.ami_count == 2:
+            print "Initiating test call"
+            self.ast[0].cli_originate("Local/9876 at ast2/n extension 101 at ast2")
+
+    # This is called when the reactor has started running.
     def run(self):
         TestCase.run(self)
         self.create_ami_factory(2)
@@ -80,11 +111,12 @@
     test.start_asterisk()
     reactor.run()
     test.stop_asterisk()
-    if test.passed != True:
-        return 1
-    return 0
+    if test.passed:
+        return 0
+    return 1
 
 if __name__ == "__main__":
     sys.exit(main() or 0)
 
+
 # vim:sw=4:ts=4:expandtab:textwidth=79