[asterisk-commits] jpeeler: testsuite/asterisk/trunk r305 - in /asterisk/trunk: lib/python/ test...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Fri May 21 17:41:11 CDT 2010
Author: jpeeler
Date: Fri May 21 17:41:10 2010
New Revision: 305
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=305
Log:
Add new python module to facilitate monitoring manager events when testing.
A basic sample has been provided in the run-test file. More documentation
regarding use is present in the client module in /lib/python.
Some notes:
Since the main purpose of this is to be used by other tests, the ami-monitor
test has not been added to the tests.yaml file.
There is some quirky code present that makes things work properly with the
paths that allows for direct execution of the test, which I call standalone
mode. Obviously other tests do not need to copy this behavior.
Added:
asterisk/trunk/lib/python/client.py (with props)
asterisk/trunk/tests/ami-monitor/
asterisk/trunk/tests/ami-monitor/configs/
asterisk/trunk/tests/ami-monitor/configs/logger.conf (with props)
asterisk/trunk/tests/ami-monitor/configs/manager.conf (with props)
asterisk/trunk/tests/ami-monitor/run-test (with props)
asterisk/trunk/tests/ami-monitor/starpy.conf (with props)
asterisk/trunk/tests/ami-monitor/test-config.yaml (with props)
Added: asterisk/trunk/lib/python/client.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/client.py?view=auto&rev=305
==============================================================================
--- asterisk/trunk/lib/python/client.py (added)
+++ asterisk/trunk/lib/python/client.py Fri May 21 17:41:10 2010
@@ -1,0 +1,338 @@
+#! /usr/bin/env python
+
+from twisted.application import service, internet
+from twisted.internet import reactor
+from starpy import manager, fastagi, utilapplication, menu
+import os, logging, pprint, time, sys, threading
+
+sys.path.append("lib/python")
+from asterisk.asterisk import Asterisk
+
+"""
+This module is meant to be instantiated with the following parameters:
+def __init__(self, myargv, myasterisk, mytester, mytimeout = 5)
+myargv - the arguments passed to main
+myasterisk - the Asterisk instance created from the Asterisk class
+mytester - an instance of the created test class
+mytimeout - (optional) number of seconds before a test is considered to have
+timed out
+
+The mytester param gives the watcher class the necessary information to call
+each test method. Each test method is defined in a class of your choosing
+similar to below. The purpose of each test method is to set the events to be
+monitored for as well as optionally events to send which also need events
+monitored.
+
+ def test0(self, watcher):
+ event1 = [{'response' : 'Success', 'ping' : 'Pong'}]
+ watcher.add_event(event1)
+ event_send = {'Action' : 'ping'}
+ watcher.add_send_event(event_send)
+
+ # identical test
+ def test1(self, watcher):
+ event1 = [{'response' : 'Success', 'ping' : 'Pong'}]
+ watcher.add_event(event1)
+ event_send = {'Action' : 'ping'}
+ watcher.add_send_event(event_send)
+
+ def test2(self, watcher):
+ event1 = [{'event' : 'Alarm', 'channel' : '17' }]
+ event2 = [{'event' : 'Alarm', 'channel' : '18' }]
+ event3 = [{'event' : 'Alarm', 'channel' : '20' }, {'event' : 'Alarm', 'channel' : '19'}]
+ watcher.add_event(event1)
+ watcher.add_event(event2)
+ watcher.add_event(event3)
+ watcher.set_ordered(True)
+
+ # alternative event set up:
+ #watcher.set_events(False,
+ # [[{'event' : 'Alarm', 'channel' : '18' }],
+ # [{'event' : 'Alarm', 'channel' : '17' }],
+ # [{'event' : 'Alarm', 'channel' : '19' }, {'event' : 'Alarm', 'channel' : '20'}]])
+
+Events are described using dictionaries very similar to how they would appear
+in a manager session. Only the portion of the event you want to match should
+be specified.
+
+Events to monitor are added via add_event. Add_event takes a list of
+dictionaries. The reason it is a list is to cover the case where you want to
+specify either event may be considered as a successful match.
+
+Events to send are added via add_send_event. Add_send_event takes a dictionary
+representing the event to send.
+
+The client module contains a set_ordered method which may either be set True or
+False, depending on whether or not the order of the event matching matters. Or
+the set_events method may be used which the first argument sets the ordering
+method and the second argument is a list of list of dictionaries for all events
+to be matched.
+
+These are the supported event monitoring scenarios:
+
+S1: unordered
+[SEND] -> 3 -> 1 -> 2
+
+S2: ordered
+[SEND] -> 1 -> 2 -> 3
+
+S3: optional events
+[SEND] -> 1 -> {2, 3} -> 4
+
+The [SEND] above is optional and corresponds to the add_send_event method,
+which also takes a dictionary representing the event to be sent.
+
+"""
+
+
+
+class EventWatcher():
+
+ def start_asterisk(self):
+ self.log.info("Starting Asterisk")
+ self.asterisk.start()
+ self.asterisk.cli_exec("core set verbose 10")
+ self.asterisk.cli_exec("core set debug 3")
+
+ def stop_asterisk(self):
+ if not self.standalone:
+ self.asterisk.stop()
+
+ def clear_vars(self):
+
+ self.event_list = list()
+ self.count = 0
+ self.send_event_list = list()
+ self.ordered = False
+
+ def __init__(self, myargv, myasterisk, mytester, mytimeout = 5):
+ self.log = logging.getLogger('TestAMI')
+ self.log.setLevel(logging.INFO)
+ self.ami = None
+ self.testcount = 0
+ self.passed = True
+ self.call_id = None
+ self.working_dir = os.path.dirname(myargv[0])
+ self.testobj = mytester
+ self.timeout_sec = mytimeout
+
+ if len(myargv) == 1:
+ self.standalone = True
+ else:
+ self.standalone = False
+ if myasterisk == None:
+ self.passed = False
+ self.log.critical("Fail to pass Asterisk instance!")
+ return
+
+ self.asterisk = myasterisk
+
+ self.reactor_lock = threading.Lock()
+ self.reactor_stopped = False
+
+ self.clear_vars()
+
+ if self.standalone:
+ return
+
+ self.start_asterisk()
+
+ def set_ordered(self, ordered):
+ self.ordered = ordered
+
+ def add_event(self, event):
+ self.event_list.append(event)
+ self.count = self.count + 1
+
+ def add_send_event(self, event):
+ self.send_event_list.append(event)
+
+ def set_events(self, ordered, list):
+ self.event_list = list
+ self.count = len(list)
+ self.ordered = ordered
+
+ def show_events(self):
+ if self.log.getEffectiveLevel() < logging.DEBUG:
+ return
+
+ self.log.debug("Showing events")
+ for n in self.event_list:
+ self.log.debug("Event: %s" % n)
+
+ def get_events(self):
+ return self.event_list
+
+ def check_events(self):
+ if not self.event_list:
+ return False
+
+ for events in self.event_list:
+ matched = False
+ for event in events:
+ if "match" in event:
+ matched = True
+ continue
+ if not matched:
+ return False
+ return True
+
+ def timeout(self):
+ self.log.debug("Timed out")
+ self.passed = False
+ self.reactor_lock.acquire()
+ if not self.reactor_stopped:
+ reactor.stop()
+ self.reactor_stopped = True
+ self.stop_asterisk()
+ self.reactor_lock.release()
+
+ def end_test(self):
+ self.reactor_lock.acquire()
+ if not self.reactor_stopped:
+ self.call_id.cancel()
+ reactor.stop()
+ self.reactor_stopped = True
+ self.stop_asterisk()
+ self.reactor_lock.release()
+ self.log.info("DONE - end_test")
+
+ def dict_in_dict(self, d_set, d_subset):
+ return len(d_subset) == len(set(d_subset.items()) & set(d_set.items()))
+
+ def start(self):
+ utilapplication.UtilApplication.configFiles = (os.getcwd() + '/' + self.working_dir + '/starpy.conf', 'starpy.conf')
+ # Log into AMI
+ amiDF = utilapplication.UtilApplication().amiSpecifier.login().addCallbacks(self.on_connect, self.on_failure)
+
+ def send_events(self, ami):
+ if self.send_event_list:
+ for amessage in self.send_event_list:
+ id = ami.sendMessage(amessage, self.on_sent_response)
+ self.log.debug("Sending %s with id %s" % (amessage, id))
+
+ def on_failure(self, ami):
+ self.log.critical("Stopping asterisk, login failure")
+ self.passed = False
+ self.stop_asterisk()
+
+ def on_connect(self, ami):
+ """Register for AMI events"""
+ # XXX should handle asterisk reboots (at the moment the AMI
+ # interface will just stop generating events), not a practical
+ # problem at the moment, but should have a periodic check to be sure
+ # the interface is still up, and if not, should close and restart
+ ami.status().addCallback(self.on_status, ami=ami)
+
+ if len(self.event_list) or len(self.send_event_list) != 0:
+ self.log.warn("Don't load events outside of a test method!")
+
+ self.load_next_test(ami, True)
+
+ if len(self.event_list) == 0:
+ self.log.error("No events to monitor!")
+
+ ami.registerEvent(None, self.on_any_event)
+
+ self.ami = ami
+ self.send_events(ami)
+
+ def on_sent_response(self, result):
+ # don't care about connection terminated event
+ if result.__str__() != "Connection was closed cleanly: FastAGI connection terminated.":
+ self.log.debug("Result from sent event: %s", result)
+ self.on_any_event(self.ami, result)
+ else:
+ self.log.debug("Ignoring connection close event")
+
+ def connectionLost(reason):
+ self.log.critical("Connection lost: %s", reason)
+
+ def on_any_event(self, ami, event):
+
+ self.log.debug("Running on_any_event")
+
+ if not self.ordered:
+ for next_events in self.event_list:
+ for next_event in next_events:
+ self.log.debug("Looking at %s vs %s" % (next_event, event))
+ if self.dict_in_dict(event, next_event):
+ next_event.update({"match": time.time()})
+ self.log.debug("New event %s" % next_event)
+ self.count = self.count - 1
+ if self.count == 0:
+ self.load_next_test(ami)
+ else:
+ index = abs(self.count - len(self.event_list))
+ for next_event in self.event_list[index]:
+ if self.dict_in_dict(event, next_event):
+ next_event.update({"match": time.time()})
+ self.log.debug("New event %s" % next_event)
+ self.count = self.count - 1
+ if self.count == 0:
+ self.load_next_test(ami)
+ continue
+
+ def exec_next_test(self, count, toexec):
+ self.log.info("Next test count %s", count)
+
+ if not hasattr(self.testobj, toexec):
+ self.log.debug("No more tests")
+ return -1
+
+ self.log.debug("Test method %s exists", toexec)
+ if self.call_id:
+ self.call_id.cancel()
+ method = getattr(self.testobj, toexec)
+ self.log.debug("Begin executing %s", toexec)
+ method(self)
+ self.log.debug("Finish executing %s", toexec)
+ if len(self.event_list) > 0:
+ self.log.debug("Rescheduling timeout")
+ self.call_id = reactor.callLater(self.timeout_sec, self.timeout)
+ return 0
+
+ self.log.warn("Returning, no events added by test method!")
+ return -1 # exception or something
+
+ def load_next_test(self, ami, firstrun=None):
+ if not firstrun:
+ self.log.debug("About to shut down all monitoring")
+ ami.deregisterEvent(None, None)
+
+ passed = self.check_events()
+ if not passed:
+ self.log.debug("TEST FAILED")
+ self.passed = False
+ self.end_test()
+ return
+
+ self.log.debug("Test passed")
+ self.show_events()
+ self.clear_vars()
+
+ toexec = "test" + str(self.testcount)
+ res = self.exec_next_test(self.testcount, toexec)
+ self.log.debug("res %s from exec %s" % (res, toexec))
+ if res == -1:
+ self.ami = None
+ self.end_test()
+ return
+ self.testcount = self.testcount + 1
+ ami.registerEvent(None, self.on_any_event)
+ self.send_events(ami)
+
+ def on_status(self, events, ami=None):
+ self.log.debug("Initial channel status retrieved")
+ if events:
+ self.log.critical("Test expects no channels to have started yet, aborting!")
+ ami.deregisterEvent(None, None)
+ self.passed = False
+ self.end_test()
+ for event in events:
+ self.log.debug("Received event: %s", event)
+
+if __name__ == "__main__":
+ print "This code is meant to be imported"
+
+# vim:sw=4:ts=4:expandtab:textwidth=79
Propchange: asterisk/trunk/lib/python/client.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/lib/python/client.py
------------------------------------------------------------------------------
svn:executable = *
Propchange: asterisk/trunk/lib/python/client.py
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/lib/python/client.py
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/trunk/tests/ami-monitor/configs/logger.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/ami-monitor/configs/logger.conf?view=auto&rev=305
==============================================================================
--- asterisk/trunk/tests/ami-monitor/configs/logger.conf (added)
+++ asterisk/trunk/tests/ami-monitor/configs/logger.conf Fri May 21 17:41:10 2010
@@ -1,0 +1,7 @@
+[general]
+
+[logfiles]
+
+console =>
+messages => notice,warning,error
+full => notice,warning,error,debug,verbose
Propchange: asterisk/trunk/tests/ami-monitor/configs/logger.conf
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/tests/ami-monitor/configs/logger.conf
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/tests/ami-monitor/configs/logger.conf
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/trunk/tests/ami-monitor/configs/manager.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/ami-monitor/configs/manager.conf?view=auto&rev=305
==============================================================================
--- asterisk/trunk/tests/ami-monitor/configs/manager.conf (added)
+++ asterisk/trunk/tests/ami-monitor/configs/manager.conf Fri May 21 17:41:10 2010
@@ -1,0 +1,10 @@
+[general]
+enabled = yes
+port = 5038
+bindaddr = 127.0.0.1
+
+[user]
+secret = mysecret
+read = system,call,log,verbose,agent,user,config,dtmf,reporting,cdr,dialplan
+write = system,call,agent,user,config,command,reporting,originate
+
Propchange: asterisk/trunk/tests/ami-monitor/configs/manager.conf
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/tests/ami-monitor/configs/manager.conf
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/tests/ami-monitor/configs/manager.conf
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/trunk/tests/ami-monitor/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/ami-monitor/run-test?view=auto&rev=305
==============================================================================
--- asterisk/trunk/tests/ami-monitor/run-test (added)
+++ asterisk/trunk/tests/ami-monitor/run-test Fri May 21 17:41:10 2010
@@ -1,0 +1,71 @@
+#! /usr/bin/env python
+
+import sys
+
+from twisted.application import service, internet
+from twisted.internet import reactor
+from starpy import manager, fastagi, utilapplication, menu
+import os, logging, pprint, time, sys
+
+from optparse import OptionParser
+
+"""
+This test module should serve as a template for new event monitoring tests. A
+test class is created with sequential test methods named testX, where X starts
+at 0. The purpose of each test method is to set the events to be monitored for
+as well as optionally events to send which also need events monitored.
+
+Please see the documentation in the client module for more information.
+"""
+
+class Test():
+
+ def test0(self, watcher):
+ event1 = [{'response' : 'Success', 'ping' : 'Pong'}]
+ watcher.add_event(event1)
+ event_send = {'Action' : 'ping'}
+ watcher.add_send_event(event_send)
+
+def main(argv = None):
+ if argv is None:
+ argv = sys.argv
+
+ if len(argv) == 1: # options are NOT being passed
+ sys.path.append("../../lib/python")
+ asterisk = None
+ else:
+ sys.path.append("lib/python")
+ from asterisk.asterisk import Asterisk
+
+ parser = OptionParser()
+ parser.add_option("-v", "--version", dest="ast_version",
+ help="Asterisk version string")
+ (options, args) = parser.parse_args(argv)
+ working_dir = os.path.dirname(argv[0])
+
+ asterisk = Asterisk(base="/tmp/asterisk-testsuite/ami-monitor")
+ asterisk.install_config("tests/ami-monitor/configs/manager.conf")
+ asterisk.install_config("tests/ami-monitor/configs/logger.conf")
+
+ logging.basicConfig()
+ import client
+
+ tester = Test()
+ watcher = client.EventWatcher(argv, asterisk, tester, 5)
+
+ #watcher.log.setLevel(logging.DEBUG)
+ #manager.log.setLevel(logging.DEBUG)
+
+ reactor.callWhenRunning(watcher.start)
+ reactor.run()
+
+ if watcher.passed:
+ watcher.log.info("All tests were good!")
+ return 0
+ watcher.log.error("Test failure")
+ return 1
+
+if __name__ == "__main__":
+ sys.exit(main() or 0)
+
+# vim:sw=4:ts=4:expandtab:textwidth=79
Propchange: asterisk/trunk/tests/ami-monitor/run-test
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/tests/ami-monitor/run-test
------------------------------------------------------------------------------
svn:executable = *
Propchange: asterisk/trunk/tests/ami-monitor/run-test
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/tests/ami-monitor/run-test
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/trunk/tests/ami-monitor/starpy.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/ami-monitor/starpy.conf?view=auto&rev=305
==============================================================================
--- asterisk/trunk/tests/ami-monitor/starpy.conf (added)
+++ asterisk/trunk/tests/ami-monitor/starpy.conf Fri May 21 17:41:10 2010
@@ -1,0 +1,5 @@
+[AMI]
+username=user
+secret=mysecret
+server=localhost
+port=5038
Propchange: asterisk/trunk/tests/ami-monitor/starpy.conf
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/tests/ami-monitor/starpy.conf
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/tests/ami-monitor/starpy.conf
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: asterisk/trunk/tests/ami-monitor/test-config.yaml
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/ami-monitor/test-config.yaml?view=auto&rev=305
==============================================================================
--- asterisk/trunk/tests/ami-monitor/test-config.yaml (added)
+++ asterisk/trunk/tests/ami-monitor/test-config.yaml Fri May 21 17:41:10 2010
@@ -1,0 +1,10 @@
+testinfo:
+ summary: 'Test monitoring a pong event from the Asterisk Manager Interface'
+ description: |
+ 'This test provides monitoring for one select AMI event after logging in.'
+
+properties:
+ minversion: '1.8' #maybe 1.4 can be supported later
+ dependencies:
+ - python : 'twisted'
+ - python : 'starpy'
Propchange: asterisk/trunk/tests/ami-monitor/test-config.yaml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: asterisk/trunk/tests/ami-monitor/test-config.yaml
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: asterisk/trunk/tests/ami-monitor/test-config.yaml
------------------------------------------------------------------------------
svn:mime-type = text/plain
More information about the asterisk-commits
mailing list