[asterisk-commits] rmudgett: branch rmudgett/srtp r1675 - /asterisk/team/rmudgett/srtp/tests/cha...

[SVN commits to the Asterisk project]

Author: rmudgett
Date: Thu Jun 23 18:37:38 2011
New Revision: 1675

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=1675
Log:
Get noload_res_srtp_attempt_srtp test working.

Modified:
    asterisk/team/rmudgett/srtp/tests/channels/SIP/noload_res_srtp_attempt_srtp/run-test

Modified: asterisk/team/rmudgett/srtp/tests/channels/SIP/noload_res_srtp_attempt_srtp/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/rmudgett/srtp/tests/channels/SIP/noload_res_srtp_attempt_srtp/run-test?view=diff&rev=1675&r1=1674&r2=1675
==============================================================================
--- asterisk/team/rmudgett/srtp/tests/channels/SIP/noload_res_srtp_attempt_srtp/run-test (original)
+++ asterisk/team/rmudgett/srtp/tests/channels/SIP/noload_res_srtp_attempt_srtp/run-test Thu Jun 23 18:37:38 2011
@@ -23,42 +23,57 @@
         self.connected_chan2 = False
         self.connected_srtp1 = False
         self.connected_srtp2 = False
-        self.got_hangup1 = False
+        self.not_acceptable1 = False
 
         # Safe Reactor timeout values on my own machine could be as low as 1.
         self.reactor_timeout = 10
 
         # Create one AGI server
+        # Will call fastagi_connect() once an AGI connection is established.
         self.create_fastagi_factory()
 
         # Create two Asterisk instances ...
         self.create_asterisk(2)
 
+    # This is called when the reactor has started running.
     def run(self):
         TestCase.run(self)
 
-        # Will link up to ami_connect() once the connection is established.
+        # Create one AMI connection (to Ast1)
+        # Will call ami_connect() once the AMI connection is established.
         self.create_ami_factory()
 
     # Once the AMI Factory connects to Asterisk, this function fires.
     def ami_connect(self, ami):
         TestCase.ami_connect(self, ami)
-        self.ami = ami
 
         print "AMI - connected"
-        self.ami.registerEvent('Hangup', self.ami_test_hangup)
+
+        # Register for the AMI event we are interested in.
+        self.ami[ami.id].registerEvent("VarSet", self.ami_test_varset)
 
         print "Initiating test call"
         self.ast[0].cli_exec(
             "originate SIP/2000/2000 extension 1000 at siptest1")
 
-    # This is called whenever a hangup event occurs on Ast1.
-    def ami_test_hangup(self, ami, event):
-        print "Received Hangup event from AMI"
-        self.got_hangup1 = True
+    # This is called whenever an AMI VarSet event occurs on Ast1.
+    def ami_test_varset(self, ami, event):
+        print "Received VarSet event from AMI"
+        #print event
+        if not event.has_key("variable") or not event.has_key("value"):
+            # These entries should always be in the VarSet event.
+            print "Event is missing 'variable' or 'value' entry"
+            return
+        print "  Value of event[variable] = %s" % (event["variable"])
+        print "  Value of event[value] = %s" % (event["value"])
+        if 0 <= event["variable"].startswith("~HASH~SIP_CAUSE~"):
+            cause_code = event["value"].split(" ")[1]
+            if cause_code == "488":
+                # The call failed for the expected reason "488 Not acceptable".
+                self.not_acceptable1 = True
 
-        # We are done
-        self.stop_reactor()
+                # We are done early
+                self.stop_reactor()
 
     # This is called by each Asterisk instance if the call gets connected.
     def fastagi_connect(self, agi):
@@ -85,10 +100,10 @@
         # Determine if the test passed
         print "self.connected_chan1:%s" % (self.connected_chan1)
         print "self.connected_srtp1:%s" % (self.connected_srtp1)
-        print "self.got_hangup1:    %s" % (self.got_hangup1)
+        print "self.not_acceptable1:%s" % (self.not_acceptable1)
         print "self.connected_chan2:%s" % (self.connected_chan2)
         print "self.connected_srtp2:%s" % (self.connected_srtp2)
-        if not self.connected_chan1 and not self.connected_srtp1 and self.got_hangup1 and not self.connected_chan2 and not self.connected_srtp2:
+        if not self.connected_chan1 and not self.connected_srtp1 and self.not_acceptable1 and not self.connected_chan2 and not self.connected_srtp2:
             print "Test passed"
             self.passed = True