[asterisk-commits] mmichelson: branch mmichelson/bridge-tests r3319 - /asterisk/team/mmichelson/...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Thu Jul 12 17:46:26 CDT 2012


Author: mmichelson
Date: Thu Jul 12 17:46:24 2012
New Revision: 3319

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=3319
Log:
Adding more progress to two-party bridge test object


Modified:
    asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py

Modified: asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py?view=diff&rev=3319&r1=3318&r2=3319
==============================================================================
--- asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py (original)
+++ asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py Thu Jul 12 17:46:24 2012
@@ -15,6 +15,199 @@
 from TestCase import TestCase
 
 LOGGER = logging.getLogger(__name__)
+
+class BridgeTest(object):
+    '''
+    Callable class that runs a bridge test call
+    '''
+    def __init__(self, ami_uut, ami_alice, ami_bob):
+        # Indicates if the UUT thinks the call is bridged
+        self.bridged = False
+        # Indicates that Alice thinks the call is bridged
+        self.alice_ready = False
+        # Indicates that Bob thinks the call is bridged
+        self.bob_ready = False
+
+        # Indicates Alice received expected DTMF over bridged call
+        self.alice_dtmf = False
+        # Indicates Bob received expected DTMF over bridged call
+        self.bob_dtmf = False
+
+        # Indicates Alice's channel has hung up
+        self.alice_hungup = False
+        # Indicates Bob's channel has hung up
+        self.bob_hungup = False
+
+        # Channel name of Alice's channel in the UUT
+        self.alice_channel = None
+        # Channel name of Bob's channel in the UUT
+        self.bob_channel = None
+
+        ami_alice.registerEvent('UserEvent', self.alice_user_callback)
+        ami_alice.registerEvent('Hangup', self.alice_hangup_callback)
+        ami_bob.registerEvent('UserEvent', self.bob_user_callback)
+        ami_bob.registerEvent('Hangup', self.bob_hangup_callback)
+        ami_uut.registerEvent('Bridge', self.uut_bridge_callback)
+        ami_uut.registerEvent('TestEvent', self.uut_test_callback)
+
+    def __del__():
+        ami_alice.deregisterEvent('UserEvent', self.alice_user_callback)
+        ami_alice.registerEvent('Hangup', self.alice_hangup_callback)
+        ami_bob.deregisterEvent('UserEvent', self.bob_user_callback)
+        ami_bob.registerEvent('Hangup', self.bob_hangup_callback)
+        ami_uut.deregisterEvent('Bridge', self.uut_bridge_callback)
+        ami_uut.deregisterEvent('TestEvent', self.uut_test_callback)
+
+    def __call__(self, test_run):
+        # Step 1: Initiate a call from Alice to Bob
+        self.ami_alice.originate(channel = 'Local/test_call',
+                exten = test_run['Extension'],
+                context = 'default',
+                priority = '1')
+
+        # Step 2: Wait for events to fire that indicate that
+        # the call has been set up
+        # XXX There is probably a better way to do this
+        while not (self.bridged and self.alice_ready and self.bob_ready):
+            pass
+
+        # Step 3: Send DTMF to determine if the media path
+        # has been set up correctly.
+        self.check_audio_path()
+
+        # XXX Again, there's probably a better way to do this
+        while not(self.alice_dtmf and self.bob_dtmf):
+            pass
+
+        # Step 4: Check connected line IDs on the UUT to make sure
+        # proper information is being passed around
+        self.check_identities()
+
+        # Step 5: If any features are supposed to be run, then do it.
+        if 'features' in test_run:
+            self.enact_features(test_run['features'])
+
+        # Step 6: If necessary, send a hangup to the appropriate party.
+        if 'hangup' in test_run:
+            self.send_hangup(test_run['hangup'])
+
+        # Step 7: Wait for call to be hung up
+        while not(self.alice_hungup and self.bob_hungup):
+            pass
+
+    def uut_bridge_callback(self, ami, event):
+        self.uut_alice_channel = event.get('channel1')
+        self.uut_bob_channel = event.get('channel2')
+        if event.get('bridgestate') == 'Link':
+            LOGGER.info("Bridge is up")
+            self.bridged = True
+        else
+            LOGGER.info("Bridge is down")
+            self.bridged = False
+        LOGGER.info("Got bridge callback")
+
+    def uut_test_callback(self, ami, event):
+        LOGGER.info("Got test callback")
+
+    def alice_user_callback(self, ami, event):
+        if (event.get('userevent') == 'Connected'):
+            self.alice_ready = True
+            self.alice_channel = event.get('channel')
+            LOGGER.info("Alice is set up")
+
+        if (event.get('userevent') == 'DTMF'):
+            self.alice_dtmf = True
+            LOGGER.info("Audio path from Bob to Alice is functional")
+
+    def alice_hangup_callback(self, ami, event):
+        LOGGER.info("Alice has hung up")
+        self.alice_hungup = True
+
+    def bob_hangup_callback(self, ami, event):
+        LOGGER.info("Bob has hung up")
+        self.bob_hungup = True
+
+    def bob_user_callback(self, ami, event):
+        if (event.get('userevent') == 'Connected'):
+            self.bob_ready = True
+            self.bob_channel = event.get('channel')
+            LOGGER.info("Bob is set up")
+
+        if (event.get('userevent') == 'DTMF'):
+            LOGGER.info("Audio path from Alice to Bob is functional")
+
+    def check_audio_path(self):
+        # XXX For simplification purposes, I only send one
+        # digit. This may be increased. I may also alter Asterisk
+        # to have another manager action that can play multiple
+        # tones because it's a bit ridiculous that I can only play
+        # one at a time
+        self.ami_alice.playDTMF(self.alice_channel, '1')
+        self.ami_bob.playDTMF(self.bob_channel, '1')
+
+    def check_identities(self):
+        alice_connected = self.ami_uut.getVar(self.uut_alice_channel, 'CONNECTEDLINE(all)')
+        bob_connected = self.ami_uut.getVar(self.uut_bob_channel, 'CONNECTEDLINE(all)')
+        alice_bridgepeer = self.ami_uut.getVar(self.uut_alice_channel, 'BRIDGEPEER')
+        bob_bridgepeer = self.ami_uut.getVar(self.uut_bob_channel, 'BRIDGEPEER')
+        # XXX Check the connected line values here
+
+    def enact_features(self, features):
+        for feature in features:
+            if ('what' not in feature or
+                    'who' not in feature or
+                    'success' not in feature):
+                raise "Missing required feature configuration"
+
+            # It's possible that a previous feature execution
+            while not self.bridged:
+                pass
+
+            feature_dtmf = self.decode_feature(feature['What'])
+
+            if (feature['who'] == 'alice' or feature['who'] == 'caller'):
+                ami = self.ami_alice
+                channel = self.alice_channel
+            elif (feature['who'] == 'bob' or feature['who'] == 'callee'):
+                ami = self.ami_bob
+                channel = self.bob_channel
+            else:
+                raise "Invalid party selected for feature operation"
+
+            ami.playDTMF(channel, feature_dtmf)
+
+            #XXX Need to wait for a test event to tell us if the feature
+            # passed or failed
+
+    def decode_feature(self, what):
+        if what == 'disconnect':
+            dtmf = '0'
+        elif what == 'attendedtransfer':
+            dtmf = '#'
+        elif what == 'blindtransfer':
+            dtmf = '*'
+        elif what == 'automonitor':
+            dtmf = '9'
+        elif what == 'automixmonitor':
+            dtmf = '8'
+        elif what == 'park':
+            dtmf = '7'
+        else:
+            raise "Invalid feature selection"
+
+        return dtmf
+
+    def send_hangup(self, target):
+        if (target == 'alice' or target == 'caller'):
+            ami = self.ami_alice
+            channel = self.alice_channel
+        elif (target == 'bob' or target == 'callee'):
+            ami = self.ami_bob
+            channel = self.bob_channel
+        else:
+            raise "Invlid party selected for hangup"
+
+        ami.hangup(channel)
 
 class BridgeTestCase(TestCase):
     '''
@@ -29,7 +222,6 @@
         TestCase.__init__(self, test_path)
         self.create_asterisk(3)
         self._test_runs = []
-        self._current_run = 0
         self.ami_uut = None
         self.ami_alice = None
         self.ami_bob = None
@@ -38,6 +230,8 @@
             LOGGER.warning("No configuration provided. Bailing.")
             raise Exception
 
+        # Just a quick sanity check so we can die early if
+        # the tests are badly misconfigured
         for test_run in test_config:
             if not 'Extension' in test_run:
                 LOGGER.error("No configured extension in test run")
@@ -47,112 +241,23 @@
 
     def ami_connect(self, ami):
         if (ami.id == 0):
-            ami.registerEvent('UserEvent', self.alice_user_callback)
             self.ami_uut = ami
         elif (ami.id == 1):
-            ami.registerEvent('NewState', self.newstate_callback)
             self.ami_alice = ami
         elif (ami.id == 2):
-            ami.registerEvent('UserEvent', self.bob_user_callback)
             self.ami_bob = ami
         else:
             LOGGER.warning("Unexpected AMI ID %d recieved" % ami.id)
 
         if self.ami_uut and self.ami_alice and self.ami_bob:
             # We can get started with the test!
-            self.run_iterations
-
-    def run_iterations(self):
+            self.run_tests
+
+    def run_tests(self):
         for iteration in self._test_runs:
-            self.run_test(iteration)
-
-    def run_test(self, test_run):
-        # Step 1: Initiate a call from Alice to Bob
-        self.ami_alice.originate(channel = 'Local/test_call',
-                exten = test_run['Extension'],
-                context = 'default',
-                priority = '1')
-
-        # Step 2: Wait for events to fire that indicate that
-        # the call has been set up
-
-        # XXX There is probably a better way to do this
-        while not (self.bridged and self.alice_ready and self.bob_ready):
-            pass
-
-        # Step 3: Send DTMF to determine if the media path
-        # has been set up correctly.
-        self.check_audio_path()
-
-        # XXX Again, there's probably a better way to do this
-        while not(self.alice_dtmf and self.bob_dtmf):
-            pass
-
-        # Step 4: Check connected line IDs on the UUT to make sure
-        # proper information is being passed around
-        self.check_identities()
-
-        # Step 5: If any features are supposed to be run, then do it.
-        if 'features' in test_run:
-            self.enact_features(test_run['features'])
-
-        # Step 6: If necessary, send a hangup to the appropriate party.
-        if 'hangup' in test_run:
-            self.send_hangup(test_run['hangup'])
-
-    # Step 2: Ensure that the call gets set up by getting
-    # events from the participants
-
-    def bridge_callback(self, ami, event):
-        self.alice_channel = event.get('channel1')
-        self.bob_channel = event.get('channel2')
-        self.bridged = True
-        LOGGER.info("Got bridge callback")
-
-    def alice_user_callback(self, ami, event):
-        if (event.get('userevent') == 'Ready'):
-            self.alice_ready = True
-            LOGGER.info("Alice is set up")
-
-        if (event.get('userevent') == 'DTMF'):
-            self.alice_dtmf = True
-            LOGGER.info("Audio path from Bob to Alice is functional")
-
-    def bob_user_callback(self, ami, event):
-        if (event.get('userevent') == 'Ready'):
-            LOGGER.info("Bob is set up")
-
-        if (event.get('userevent') == 'DTMF'):
-            LOGGER.info("Audio path from Alice to Bob is functional")
-
-    def check_audio_path(self):
-        # Step 3: Check that media is set up properly. Both
-        # Alice and Bob should be waiting to read DTMF digits.
-        # Tell each to play digits out to the other side.
-
-        # XXX For simplification purposes, I only send one
-        # digit. This may be increased. I may also alter Asterisk
-        # to have another manager action that can play multiple
-        # tones because it's a bit ridiculous that I can only play
-        # one at a time
-        msg = "Action: PlayDTMF\r\nDigit: 1\r\n\r\n"
-        self.ami_alice.sendMessage(msg)
-        self.ami_bob.sendMessage(msg)
-
-    def check_identities(self):
-        # Step 4: Check that the UUT has expected identity information
-        # for parties involved. We get this information using getvar
-
-        alice_connected = self.ami_uut.getVar(self.alice_channel, 'CONNECTEDLINE(all)')
-        bob_connected = self.ami_uut.getVar(self.bob_channel, 'CONNECTEDLINE(all)')
-        alice_bridgepeer = self.ami_uut.getVar(self.alice_channel, 'BRIDGEPEER')
-        bob_bridgepeer = self.ami_uut.getVar(self.bob_channel, 'BRIDGEPEER')
-        # XXX Check the connected line values here
-
-    def enact_features(self, features):
-        #XXX todo
-        pass
-
-    def send_hangup(self, target):
-        #XXX todo
-        pass
+            try:
+                test_run = BridgeTest(self.ami_uut, self.ami_alice,
+                        self.ami_bob)
+                test_run(iteration)
+            except:
+                print("AARGH")




More information about the asterisk-commits mailing list