[asterisk-commits] mmichelson: branch mmichelson/bridge-tests r3312 - /asterisk/team/mmichelson/...
SVN commits to the Asterisk project
asterisk-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 asterisk-commits
mailing list