[svn-commits] mmichelson: branch mmichelson/bridge-tests r3319 - /asterisk/team/mmichelson/...
SVN commits to the Digium repositories
svn-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 svn-commits
mailing list