[svn-commits] jpeeler: branch jpeeler/event_watcher r256 - /asterisk/team/jpeeler/event_wat...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Tue Apr 27 18:19:11 CDT 2010
Author: jpeeler
Date: Tue Apr 27 18:19:06 2010
New Revision: 256
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=256
Log:
finalize changes for back to back testing to work properly
Modified:
asterisk/team/jpeeler/event_watcher/tests/ami-monitor/client.py
asterisk/team/jpeeler/event_watcher/tests/ami-monitor/test.py
Modified: asterisk/team/jpeeler/event_watcher/tests/ami-monitor/client.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/jpeeler/event_watcher/tests/ami-monitor/client.py?view=diff&rev=256&r1=255&r2=256
==============================================================================
--- asterisk/team/jpeeler/event_watcher/tests/ami-monitor/client.py (original)
+++ asterisk/team/jpeeler/event_watcher/tests/ami-monitor/client.py Tue Apr 27 18:19:06 2010
@@ -7,9 +7,6 @@
from basicproperty import common, propertied, basic
class EventWatcher(propertied.Propertied):
-
- def retard(self):
- print "DEBUG: I don't know python, can I call this?"
def clear_vars(self):
@@ -23,7 +20,7 @@
self.log.setLevel(logging.INFO)
self.ami = None
self.testcount = 0
- #self.next_test = None
+ self.passed = True
self.clear_vars()
@@ -54,6 +51,9 @@
return self.eventList
def check_events(self):
+ if not self.eventList:
+ return False
+
for events in self.eventList:
matched = False
for event in events:
@@ -66,6 +66,7 @@
def timeout(self):
print "DEBUG: timed out"
+ self.passed = False
reactor.stop()
def dict_in_dict(self, d_set, d_subset):
@@ -95,6 +96,11 @@
# the interface is still up, and if not, should close and restart
self.log.debug('onAMIConnect')
ami.status().addCallback(self.onStatus, ami=ami)
+
+ if len(self.eventList) or len(self.sendEventList) != 0:
+ self.log.critical("Don't load events outside of a test method!")
+
+ self.loadNextTest(ami, True)
if len(self.eventList) == 0:
self.log.critical("No events to monitor!")
@@ -129,7 +135,7 @@
self.log.critical("new event %s" % next_event)
self.count = self.count - 1
if self.count == 0:
- self.execNextTest(ami)
+ self.loadNextTest(ami)
#else:
#self.log.critical("??????????? NO MATCH ")
else:
@@ -140,7 +146,7 @@
self.log.critical("new event %s" % next_event)
self.count = self.count - 1
if self.count == 0:
- self.execNextTest(ami)
+ self.loadNextTest(ami)
continue
def end_test(self):
@@ -148,21 +154,24 @@
reactor.stop()
self.log.critical("DEBUG: DONE, end_test")
- def execNextTest(self, ami):
- self.log.debug("About to shut down all monitoring")
- ami.deregisterEvent(None, None)
-
- passed = self.check_events()
- if not passed:
- print "TEST FAILED"
- self.end_test()
- return
- else:
+ def loadNextTest(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:
+ print "TEST FAILED"
+ self.passed = False
+ self.end_test()
+ return
+
print "DEBUG: TEST GOOD"
-
- self.clear_vars()
+ self.show_events()
+ self.clear_vars()
+
toexec = "test" + str(self.testcount)
- res = self.next_test(self, self.testcount, passed, toexec)
+ res = self.next_test(self, self.testcount, toexec)
print "DEBUG: res %s from exec %s" % (res, toexec)
if res == -1:
self.ami = None
@@ -177,6 +186,7 @@
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)
Modified: asterisk/team/jpeeler/event_watcher/tests/ami-monitor/test.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/jpeeler/event_watcher/tests/ami-monitor/test.py?view=diff&rev=256&r1=255&r2=256
==============================================================================
--- asterisk/team/jpeeler/event_watcher/tests/ami-monitor/test.py (original)
+++ asterisk/team/jpeeler/event_watcher/tests/ami-monitor/test.py Tue Apr 27 18:19:06 2010
@@ -30,11 +30,27 @@
event_send = {'Action' : 'ping'}
watcher.add_send_event(event_send)
print "ENDWOOT2"
+
+ #def test2(self, watcher):
+ #scenario 2 with optional event matching
+ #event1 = [{'event' : 'Alarm', 'channel' : '18' }]
+ #event2 = [{'event' : 'Alarm', 'channel' : '17' }]
+ #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)
- def next_test(self, watcher, count, passed, toexec):
- print "NEXT TEST YAYA ", count, passed
- if not passed:
- return 1
+ 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)
+
+ def next_test(self, watcher, count, toexec):
+ print "NEXT TEST YAYA ", count
if not hasattr(self, toexec):
print "DEBUG: no more tests"
@@ -58,43 +74,18 @@
manager.log.setLevel(logging.DEBUG)
watcher = client.EventWatcher()
watcher.log.setLevel(logging.DEBUG)
-
- #scenario 2 with optional event matching
- #event1 = [{'event' : 'Alarm', 'channel' : '18' }]
- #event2 = [{'event' : 'Alarm', 'channel' : '17' }]
- #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)
-
- #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)
-
watcher.next_test(self.next_test)
-
- event1 = [{'response' : 'Success', 'ping' : 'Pong'}]
- watcher.add_event(event1)
- event_send = {'Action' : 'ping'}
- watcher.add_send_event(event_send)
reactor.callWhenRunning(watcher.main)
self.callID = reactor.callLater(5, watcher.timeout)
reactor.run()
- watcher.show_events()
-
- if watcher.check_events():
+ if watcher.passed:
print "DEBUG: all good!"
return 0
print "DEBUG: FAIL"
return 1
-
+
def main():
mytest = Test()
mytest.main()
More information about the svn-commits
mailing list