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

SVN commits to the Digium repositories svn-commits at lists.digium.com
Wed Jul 11 17:16:14 CDT 2012


Author: mmichelson
Date: Wed Jul 11 17:16:11 2012
New Revision: 3312

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=3312
Log:
Initial work on a two-party bridge test object.

It is nowhere near complete, so don't judge me yet.


Added:
    asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py   (with props)

Added: 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=auto&rev=3312
==============================================================================
--- asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py (added)
+++ asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py Wed Jul 11 17:16:11 2012
@@ -1,0 +1,158 @@
+#!/usr/bin/env python
+'''
+Copyright (C) 2012, Digium, Inc.
+Mark Michelson <mmichelson at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import sys
+import logging
+import uuid
+
+sys.path.append("lib/python")
+from TestCase import TestCase
+
+LOGGER = logging.getLogger(__name__)
+
+class BridgeTestCase(TestCase):
+    '''
+    Class that handles tests involving two-party bridges.
+    There are three Asterisk instances used for this test.
+    0 : the unit under test "UUT"
+    1 : the unit from which calls originate, also known as "Alice"
+    2 : the unit where calls terminate, also known as "Bob"
+    '''
+
+    def __init__(self, test_path = '', test_config = None):
+        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
+
+        if test_config is None:
+            LOGGER.warning("No configuration provided. Bailing.")
+            raise Exception
+
+        for test_run in test_config:
+            if not 'Extension' in test_run:
+                LOGGER.error("No configured extension in test run")
+                raise Exception
+
+            self._test_runs.append(iteration)
+
+    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):
+        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

Propchange: asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py
------------------------------------------------------------------------------
    svn:keywords = Author Date Id Revision

Propchange: asterisk/team/mmichelson/bridge-tests/lib/python/asterisk/BridgeTestCase.py
------------------------------------------------------------------------------
    svn:mime-type = text/plain




More information about the svn-commits mailing list