[Modeling-cvs] NotificationFramework/tests compare_perfs.py,NONE,1.1 run.py,1.2,1.3 test_Notificatio
Status: Abandoned
Brought to you by:
sbigaret
|
From: <sbi...@us...> - 2003-05-25 19:06:53
|
Update of /cvsroot/modeling/NotificationFramework/tests
In directory sc8-pr-cvs1:/tmp/cvs-serv30958/tests
Modified Files:
run.py test_NotificationCenter.py
Added Files:
compare_perfs.py
Log Message:
RFE #742511 Added the ability to register multiple callbacks for an
object, triggered by env. variable
NOTIFICATION_CENTER_MULTIPLE_CALLBACKS_PER_OBSERVER (it is NOT enabled
by default)
Fixed addObserver() which did not inform the user that a single callback
can be registered per object --> this possibly led to buggy
behaviour. Now it raises ValueError when this happens (this could make
existing code fails because of that --be prepared to correct this.
Added script tests/compare_perfs.py which calculates the performance
penalty implied by the activation of the ability to handle multiple
callbacks per object.
Fixed: when generic and specific observers were registered,
postNotification() made the list of observers grow (observers are
copied back to the list). This slowed down the framework and a
observer could then be notified more than once for the same
notification.
Fixed test_Notification.py and test_NotificationCenter.py: they couldn't
be executed as stand-alone scripts. Added tests for RFE #742511 and
fixed bugs
--- NEW FILE: compare_perfs.py ---
#! /usr/bin/env python
#-----------------------------------------------------------------------------
#
# Modeling Framework: an Object-Relational Bridge for python
# (c) 2001, 2002, 2003 Sebastien Bigaret
#
# This file is part of the Modeling Framework.
#
# The Modeling Framework is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation; either version 2 of the License, or (at
# your option) any later version.
#
# The Modeling Framework is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the Modeling Framework; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
#-----------------------------------------------------------------------------
"""
Evaluates the performance penalty implied by the activation of the ability to
handle multiple callbacks per object
"""
import os, sys, time
import utils
from test_NotificationCenter import TestObject
from NotificationFramework import NotificationCenter as NC
utils.fixpath()
N=1000
def loop(nb_iterations):
o1=TestObject()
o2=TestObject()
NC.addObserver(o1, TestObject.callback, 'MSG')
NC.addObserver(o2, TestObject.callback, 'MSG')
NC.addObserver(o1, TestObject.callback, 'MSG', '01')
NC.addObserver(o2, TestObject.callback, 'MSG', '01')
t=time.time()
i=0
while i<nb_iterations:
NC.postNotification('MSG','01')
i+=1
return time.time()-t
def run():
try:
del os.environ[NC.env_handlesMultipleCallbackPerObserver]
except:
pass
reload(NC)
t0=loop(N)
print 'Nb of iterations: %i'%N
print 'Without multiple callbacks per observer: %.6f'%t0
os.environ[NC.env_handlesMultipleCallbackPerObserver]='1'
reload(NC)
t1=loop(N)
print 'With multiple callbacks per observer: %.6f'%t1
print 'Ratio: %.4f'%(t1/t0)
if __name__=='__main__':
#import profile
#profile.run('run()', 'profile.out')
run()
Index: run.py
===================================================================
RCS file: /cvsroot/modeling/NotificationFramework/tests/run.py,v
retrieving revision 1.2
retrieving revision 1.3
diff -C2 -d -r1.2 -r1.3
*** run.py 10 Jan 2003 10:45:44 -0000 1.2
--- run.py 25 May 2003 19:06:50 -0000 1.3
***************
*** 46,53 ****
def test_suite():
import test_Notification
- import test_NotificationCenter
suite=unittest.TestSuite()
suite.addTest(test_Notification.test_suite())
- suite.addTest(test_NotificationCenter.test_suite())
return suite
--- 46,51 ----
***************
*** 65,69 ****
if args: usage(me, 1) #raise 'Unexpected arguments', args
utils.fixpath(zope_path)
! return utils.run_suite(test_suite(), verbosity=verbose)
if __name__ == "__main__":
--- 63,70 ----
if args: usage(me, 1) #raise 'Unexpected arguments', args
utils.fixpath(zope_path)
!
! errs=utils.run_suite(test_suite(), verbosity=verbose)
! import test_NotificationCenter
! errs+=test_NotificationCenter.run_tests(verbosity=verbose)
if __name__ == "__main__":
Index: test_NotificationCenter.py
===================================================================
RCS file: /cvsroot/modeling/NotificationFramework/tests/test_NotificationCenter.py,v
retrieving revision 1.4
retrieving revision 1.5
diff -C2 -d -r1.4 -r1.5
*** test_NotificationCenter.py 10 Jan 2003 10:45:44 -0000 1.4
--- test_NotificationCenter.py 25 May 2003 19:06:50 -0000 1.5
***************
*** 25,30 ****
"""Test the Model"""
! import unittest
! from NotificationFramework import NotificationCenter
# ZODB specific checks
--- 25,32 ----
"""Test the Model"""
! import unittest, os, sys
! import utils
! from NotificationFramework import NotificationCenter as NC
! utils.fixpath()
# ZODB specific checks
***************
*** 39,48 ****
class TestObject:
def __init__(self):
! self.state=0
self.nbOfNotificationsReceived=0
! def callMe(self, *kw):
self.nbOfNotificationsReceived += 1
self.state=kw[0].object()
self.userInfo=kw[0].userInfo()
class TestObjectUnregister:
--- 41,54 ----
class TestObject:
def __init__(self):
! self.state=self.state_2=0
self.nbOfNotificationsReceived=0
! def callback(self, *kw):
self.nbOfNotificationsReceived += 1
self.state=kw[0].object()
self.userInfo=kw[0].userInfo()
+ def callback_2(self, *kw):
+ self.nbOfNotificationsReceived += 1
+ self.state_2=kw[0].object()
+ self.userInfo_2=kw[0].userInfo()
class TestObjectUnregister:
***************
*** 52,56 ****
self.state=kw[0].object()
self.userInfo=kw[0].userInfo()
! NotificationCenter.removeObserver(self)
if _is_ZODB_available:
--- 58,62 ----
self.state=kw[0].object()
self.userInfo=kw[0].userInfo()
! NC.removeObserver(self)
if _is_ZODB_available:
***************
*** 64,74 ****
class TestNotificationCenter(unittest.TestCase):
"Tests for the module NotificationCenter"
! def test_1_initial(self):
"[NotificationCenter] Initial state"
! self.failIf(NotificationCenter._observers(), "test initial")
! def test_2_silent(self):
"[NotificationCenter] Tests that functions keep silent"
- NC=NotificationCenter
one=TestObject()
self.failIf(NC.removeObserver(one), "Unregister an unregistered object")
--- 70,79 ----
class TestNotificationCenter(unittest.TestCase):
"Tests for the module NotificationCenter"
! def test_01_initial(self):
"[NotificationCenter] Initial state"
! self.failIf(NC._observers(), "test initial")
! def test_02_silent(self):
"[NotificationCenter] Tests that functions keep silent"
one=TestObject()
self.failIf(NC.removeObserver(one), "Unregister an unregistered object")
***************
*** 82,88 ****
NC.removeObserver(one)
! def test_3_add_remove(self):
"[NotificationCenter] Tests that addition+removal leaves state untouched"
- NC=NotificationCenter
one=TestObject()
observers_before=NC._observers()
--- 87,92 ----
NC.removeObserver(one)
! def test_03_add_remove(self):
"[NotificationCenter] Tests that addition+removal leaves state untouched"
one=TestObject()
observers_before=NC._observers()
***************
*** 107,130 ****
self.failUnlessEqual(observers_before, observers_after)
! def test_4_generic_postNotification(self):
"[NotificationCenter] Tests generic observers"
- NC=NotificationCenter
one=TestObject()
two=TestObject()
! NC.addObserver(one, TestObject.callMe, 'notif_test_4')
! NC.addObserver(two, TestObject.callMe, 'notif_test_4')
NC.postNotification('notif_test_4', 2);
self.failIf(one.state!= 2)
self.failIf(two.state!= 2)
! def test_5a_specific_postNotification(self):
"[NotificationCenter] Tests specific observers"
- NC=NotificationCenter
one=TestObject()
two=TestObject()
userInfo={'1': 1, '2': '2'}
! NC.addObserver(one, TestObject.callMe, 'notif_test_5', 10)
! NC.addObserver(two, TestObject.callMe, 'notif_test_5', 20)
NC.postNotification('notif_test_5', 2);
self.failIf(one.state!= 0, "one shouldn't have received 'notif_test_5'")
--- 111,148 ----
self.failUnlessEqual(observers_before, observers_after)
! def test_04_generic_postNotification(self):
"[NotificationCenter] Tests generic observers"
one=TestObject()
two=TestObject()
! NC.addObserver(one, TestObject.callback, 'notif_test_4')
! NC.addObserver(two, TestObject.callback, 'notif_test_4')
NC.postNotification('notif_test_4', 2);
self.failIf(one.state!= 2)
self.failIf(two.state!= 2)
+
+ def test_04b_generic_one_callback_per_notification(self):
+ "[NotificationCenter] (generic) One callback per notification"
+ if not os.environ.get(NC.env_handlesMultipleCallbackPerObserver):
+ return
+ one=TestObject()
+ NC.addObserver(one, TestObject.callback, 'notif_test_4b_MSG_ONE')
+ NC.addObserver(one, TestObject.callback_2, 'notif_test_4b_MSG_TWO')
+ NC.postNotification('notif_test_4b_MSG_ONE', 'MSG_ONE');
+ self.failIf(one.state != 'MSG_ONE')
+ self.failIf(one.state_2 != 0)
+ self.failIf(one.nbOfNotificationsReceived != 1)
+ NC.postNotification('notif_test_4b_MSG_TWO', 'MSG_TWO');
+ self.failIf(one.state != 'MSG_ONE')
+ self.failIf(one.state_2 != 'MSG_TWO')
+ self.failIf(one.nbOfNotificationsReceived != 2)
! def test_05a_specific_postNotification(self):
"[NotificationCenter] Tests specific observers"
one=TestObject()
two=TestObject()
userInfo={'1': 1, '2': '2'}
! NC.addObserver(one, TestObject.callback, 'notif_test_5', 10)
! NC.addObserver(two, TestObject.callback, 'notif_test_5', 20)
NC.postNotification('notif_test_5', 2);
self.failIf(one.state!= 0, "one shouldn't have received 'notif_test_5'")
***************
*** 138,145 ****
self.failIf(two.state!= 20, "two hasn't received 'notif_test_5'")
! def test_5b_specific_postNotification(self):
"[NotificationCenter] Tests specific observers unregistering themselves at notification time"
! # see comments in NoticicationCenter._listToNotifyForNotification()
! NC=NotificationCenter
one=TestObject()
unregister=TestObjectUnregister()
--- 156,162 ----
self.failIf(two.state!= 20, "two hasn't received 'notif_test_5'")
! def test_05b_specific_postNotification(self):
"[NotificationCenter] Tests specific observers unregistering themselves at notification time"
! # see comments in NotificationCenter._listToNotifyForNotification()
one=TestObject()
unregister=TestObjectUnregister()
***************
*** 147,153 ****
userInfo={'1': 1, '2': '2'}
! NC.addObserver(one, TestObject.callMe, 'notif_test_5b', 10)
NC.addObserver(unregister, TestObjectUnregister.callMe, 'notif_test_5b',10)
! NC.addObserver(two, TestObject.callMe, 'notif_test_5b', 10)
NC.postNotification('notif_test_5b', 10)
self.failIf(one.state!= 10, "one should have received 'notif_test_5b'")
--- 164,170 ----
userInfo={'1': 1, '2': '2'}
! NC.addObserver(one, TestObject.callback, 'notif_test_5b', 10)
NC.addObserver(unregister, TestObjectUnregister.callMe, 'notif_test_5b',10)
! NC.addObserver(two, TestObject.callback, 'notif_test_5b', 10)
NC.postNotification('notif_test_5b', 10)
self.failIf(one.state!= 10, "one should have received 'notif_test_5b'")
***************
*** 155,170 ****
self.failIf(two.state!= 10, "two should have received 'notif_test_5b'")
! def test_6_ZTests(self):
"[NotificationCenter] Tests the specific checks made when ZODB.Persistent is available"
if not _is_ZODB_available:
return
- NC=NotificationCenter
one=ZTestObject()
self.failUnlessRaises(ValueError, NC.addObserver,
! one, TestObject.callMe, 'notif_test_6')
! def test_7_codeObjectObservers(self):
"[NotificationCenter] Tests for code objects as observers"
- NC=NotificationCenter
one=TestObject()
import testmodule
--- 172,205 ----
self.failIf(two.state!= 10, "two should have received 'notif_test_5b'")
! def test_05c_specific_one_callback_per_notification(self):
! "[NotificationCenter] (specific) One callback per notification"
! if not os.environ.get(NC.env_handlesMultipleCallbackPerObserver):
! return
! one=TestObject()
! NC.addObserver(one, TestObject.callback, 'notif_test_5c_MSG','01')
! NC.addObserver(one, TestObject.callback_2, 'notif_test_5c_MSG', '02')
! NC.postNotification('notif_test_5c_MSG', '00');
! self.failIf(one.state != 0)
! self.failIf(one.state_2 != 0)
! self.failIf(one.nbOfNotificationsReceived != 0)
! NC.postNotification('notif_test_5c_MSG', '01');
! self.failIf(one.state != '01')
! self.failIf(one.state_2 != 0)
! self.failIf(one.nbOfNotificationsReceived != 1)
! NC.postNotification('notif_test_5c_MSG', '02');
! self.failIf(one.state != '01')
! self.failIf(one.state_2 != '02')
! self.failIf(one.nbOfNotificationsReceived != 2)
!
! def test_06_ZTests(self):
"[NotificationCenter] Tests the specific checks made when ZODB.Persistent is available"
if not _is_ZODB_available:
return
one=ZTestObject()
self.failUnlessRaises(ValueError, NC.addObserver,
! one, TestObject.callback, 'notif_test_6')
! def test_07_codeObjectObservers(self):
"[NotificationCenter] Tests for code objects as observers"
one=TestObject()
import testmodule
***************
*** 180,184 ****
code=code.compile_command('import testmodule;'\
'observer_object=testmodule.one')
! NC.addObserver(code, TestObject.callMe, 'notif_test_7')
NC.postNotification('notif_test_7', 7, userInfo)
self.failIf(one.state!= 7, "one hasn't received 'notif_test_7'")
--- 215,219 ----
code=code.compile_command('import testmodule;'\
'observer_object=testmodule.one')
! NC.addObserver(code, TestObject.callback, 'notif_test_7')
NC.postNotification('notif_test_7', 7, userInfo)
self.failIf(one.state!= 7, "one hasn't received 'notif_test_7'")
***************
*** 190,196 ****
self.failUnlessEqual(observers_before, NC._observers())
! def test_8_observersAsModuleFunction(self):
"[NotificationCenter] Tests a module function as an observer"
- NC=NotificationCenter
import testmodule
# saves the current set of observers so that it can be compared
--- 225,230 ----
self.failUnlessEqual(observers_before, NC._observers())
! def test_08_observersAsModuleFunction(self):
"[NotificationCenter] Tests a module function as an observer"
import testmodule
# saves the current set of observers so that it can be compared
***************
*** 207,216 ****
'couldnt remove FunctionType observer?')
! def test_9_observerMultiplyRegistered(self):
"[NotificationCenter] observerMultiplyRegistered"
- NC=NotificationCenter
one=TestObject()
! NC.addObserver(one, TestObject.callMe, 'notif_test_9')
! NC.addObserver(one, TestObject.callMe, 'notif_test_9',
sameObserverRegistersOnce=0)
NC.postNotification('notif_test_9', 9);
--- 241,249 ----
'couldnt remove FunctionType observer?')
! def test_09_observerMultiplyRegistered(self):
"[NotificationCenter] observerMultiplyRegistered"
one=TestObject()
! NC.addObserver(one, TestObject.callback, 'notif_test_9')
! NC.addObserver(one, TestObject.callback, 'notif_test_9',
sameObserverRegistersOnce=0)
NC.postNotification('notif_test_9', 9);
***************
*** 226,231 ****
# Tests that multiply registered observer can be removed and still
# remains registered
! NC.addObserver(one, TestObject.callMe, 'notif_test_9')
! NC.addObserver(one, TestObject.callMe, 'notif_test_9',
sameObserverRegistersOnce=0)
NC.removeObserver(one, 'notif_test_9') # remove one
--- 259,264 ----
# Tests that multiply registered observer can be removed and still
# remains registered
! NC.addObserver(one, TestObject.callback, 'notif_test_9')
! NC.addObserver(one, TestObject.callback, 'notif_test_9',
sameObserverRegistersOnce=0)
NC.removeObserver(one, 'notif_test_9') # remove one
***************
*** 234,240 ****
self.failIf(one.nbOfNotificationsReceived!=3)
def _test_observer(self):
"Test with objects as observers"
! NotificationCenter.addObserver()
def test_suite():
--- 267,301 ----
self.failIf(one.nbOfNotificationsReceived!=3)
+ def test_10_post_to_generic_specific_observers_leaves_obsvs_untouched(self):
+ "[NotificationCenter] post to generic specific observers leaves obsvs untouched"
+ # There was a bug in _listToNotifyForNotification triggered when
+ # generic and specific observers are listening to a single notification.
+ # This bug increases the size of the observers for a notification each
+ # time postNotification() was called
+ one=TestObject()
+ two=TestObject()
+ NC.addObserver(one, TestObject.callback, 'notif_test_10')
+ NC.addObserver(two, TestObject.callback, 'notif_test_10')
+ NC.addObserver(one, TestObject.callback, 'notif_test_10', '10')
+ NC.addObserver(two, TestObject.callback, 'notif_test_10', '10')
+ observers_before=NC._observers()
+ NC.postNotification('notif_test_10', '10');
+ NC.postNotification('notif_test_10', '10');
+ observers_after=NC._observers()
+ self.failUnlessEqual(observers_before, observers_after)
+
+ def test_11_addObserver_and_multiple_callbacks(self):
+ "[NotificationCenter] addObserver() and multiple callbacks"
+ one=TestObject()
+ NC.addObserver(one, TestObject.callback, 'notif_test_11')
+ if not os.environ.get(NC.env_handlesMultipleCallbackPerObserver):
+ self.assertRaises(ValueError, NC.addObserver,
+ one, TestObject.callback_2, 'notif_test_11b')
+ else:
+ NC.addObserver(one, TestObject.callback_2, 'notif_test_11b')
+
def _test_observer(self):
"Test with objects as observers"
! NC.addObserver()
def test_suite():
***************
*** 243,248 ****
return suite
!
if __name__ == "__main__":
! errs = utils.run_suite(test_suite())
sys.exit(errs and 1 or 0)
--- 304,320 ----
return suite
! def run_tests(verbosity=0):
! try:
! del os.environ[NC.env_handlesMultipleCallbackPerObserver]
! except:
! pass
! reload(NC)
! errs = utils.run_suite(test_suite(), verbosity=verbosity)
! os.environ[NC.env_handlesMultipleCallbackPerObserver]='1'
! reload(NC)
! errs += utils.run_suite(test_suite(), verbosity=verbosity)
! return errs
!
if __name__ == "__main__":
! errs = run_tests()
sys.exit(errs and 1 or 0)
|