[asterisk-commits] rmudgett: testsuite/asterisk/trunk r2395 - /asterisk/trunk/tests/callparking/

[SVN commits to the Asterisk project]

Author: rmudgett
Date: Tue Sep 20 12:33:33 2011
New Revision: 2395

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=2395
Log:
Add comments and logging to callparking external test.

The review was requested for a sanity check.
Review: https://reviewboard.asterisk.org/r/1449/

Modified:
    asterisk/trunk/tests/callparking/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=2395&r1=2394&r2=2395
==============================================================================
--- asterisk/trunk/tests/callparking/run-test (original)
+++ asterisk/trunk/tests/callparking/run-test Tue Sep 20 12:33:33 2011
@@ -8,40 +8,64 @@
 '''
 
 import sys
+import logging
 from twisted.internet import reactor
 
 sys.path.append("lib/python")
-from asterisk.asterisk import Asterisk
 from asterisk.TestCase import TestCase
+
+logger = logging.getLogger(__name__)
 
 
 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):
+        logger.info("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()
+        logger.info("Parked call retrieved from exten:" + str(exten))
+        if exten != "701":
+            logger.info("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()
+        logger.info("Call parked at exten:" + str(exten))
+        if exten != "701":
+            logger.info("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")
+        logger.info("ast2_userevent status:" + str(status))
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -49,8 +73,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")
+        logger.info("ast1_userevent status:" + str(status))
         if status != "SUCCESS":
             self.stop_reactor()
             return
@@ -58,15 +84,23 @@
         self.ast1_event = True
         self.check_result()
 
+    # This is called for each AMI connection established.
     def 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:
+            logger.info("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)
@@ -78,11 +112,13 @@
     test.start_asterisk()
     reactor.run()
     test.stop_asterisk()
-    if test.passed != True:
-        return 1
-    return 0
+    if test.passed:
+        logger.info("Test passed")
+        return 0
+    return 1
 
 if __name__ == "__main__":
     sys.exit(main() or 0)
 
+
 # vim:sw=4:ts=4:expandtab:textwidth=79