Cyclone-Team/gramps
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

2005-04-09 Richard Taylor <rjt-gramps@thegrindstone.me.uk>

Browse Source 
* src/GrampsDBCallback.py: block recursive signal calls, improved error
	reporting. All warnings are now printed to stderr even if logging
	is turned off.


svn: r4328
This commit is contained in:
Richard Taylor 2005-04-09 13:04:36 +00:00
parent 4736f4ece9
commit 419cd75ec6
2 changed files with 158 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
gramps2/ChangeLog
View File

@ -1,3 +1,8 @@
2005-04-09 Richard Taylor <rjt-gramps@thegrindstone.me.uk>
* src/GrampsDBCallback.py: block recursive signal calls, improved error
reporting. All warnings are now printed to stderr even if logging
is turned off.
2005-04-08 Don Allingham <don@gramps-project.org> 2005-04-08 Don Allingham <don@gramps-project.org>
* src/AddMedia.py: pychecker changes * src/AddMedia.py: pychecker changes
* src/ChooseParents.py: pychecker changes * src/ChooseParents.py: pychecker changes

210
gramps2/src/GrampsDBCallback.py
View File

@ -234,6 +234,10 @@ class GrampsDBCallback(object):
# containing the list of types of the arguments # containing the list of types of the arguments
# that the callback methods must accept. # that the callback methods must accept.
self._current_key = 0 # counter to give a unique key to each callback. self._current_key = 0 # counter to give a unique key to each callback.
self._current_signals = [] # list of all the signals that are currently
# being emitted by this instance. This is
# used to prevent recursive emittion of the
# same signal.
# To speed up the signal type checking the signals declared by # To speed up the signal type checking the signals declared by
# each of the classes in the inheritance tree of this instance # each of the classes in the inheritance tree of this instance
@ -326,66 +330,94 @@ class GrampsDBCallback(object):
# Check signal exists # Check signal exists
if signal_name not in self.__signal_map.keys(): if signal_name not in self.__signal_map.keys():
self._log("Warning: attempt to emit to unknown signal: %s\n" self._warn("Attempt to emit to unknown signal: %s\n"
" from: file: %s\n" " from: file: %s\n"
" line: %d\n" " line: %d\n"
" func: %s\n" " func: %s\n"
% ((str(signal_name),) + inspect.stack()[1][1:4])) % ((str(signal_name),) + inspect.stack()[1][1:4]))
return return
# type check arguments # check that the signal is not already being emitted. This prevents
arg_types = self.__signal_map[signal_name] # against recursive signal emmissions.
if arg_types == None and len(args) > 0: if signal_name in self._current_signals:
self._log("Warning: signal emitted with " self._warn("Signal recursion blocked. "
"wrong number of args: %s\n" "Signals was : %s\n"
" from: file: %s\n" " from: file: %s\n"
" line: %d\n" " line: %d\n"
" func: %s\n" " func: %s\n"
% ((str(signal_name),) + inspect.stack()[1][1:4])) % ((str(signal_name),) + inspect.stack()[1][1:4]))
return return
if len(args) > 0: try:
if len(args) != len(arg_types): self._current_signals.append(signal_name)
self._log("Warning: signal emitted with "
"wrong number of args: %s\n" # check that args is a tuple. This is a common programming error.
" from: file: %s\n" if not (isinstance(args,tuple) or args == None):
" line: %d\n" self._warn("Signal emitted with argument that is not a tuple.\n"
" func: %s\n" " emit() takes two arguments, the signal name and a \n"
% ((str(signal_name),) + inspect.stack()[1][1:4])) " tuple that contains the arguments that are to be \n"
" passed to the callbacks. If you are passing a signal \n"
" argument it must be done as a single element tuple \n"
" e.g. emit('my-signal',(1,)) \n"
" signal was: %s\n"
" from: file: %s\n"
" line: %d\n"
" func: %s\n"
% ((str(signal_name),) + inspect.stack()[1][1:4]))
return return
if arg_types != None: # type check arguments
for i in range(0,len(arg_types)): arg_types = self.__signal_map[signal_name]
if not isinstance(args[i],arg_types[i]): if arg_types == None and len(args) > 0:
self._log("Warning: signal emitted with " self._warn("Signal emitted with "
"wrong arg types: %s\n" "wrong number of args: %s\n"
" from: file: %s\n" " from: file: %s\n"
" line: %d\n" " line: %d\n"
" func: %s\n" " func: %s\n"
% ((str(signal_name),) + inspect.stack()[1][1:4])) % ((str(signal_name),) + inspect.stack()[1][1:4]))
return
self._log(" arg passed was: %s, type of arg passed %s, type should be: %s\n"
% (args[i],repr(type(args[i])),repr(arg_types[i])))
return
if signal_name in self.__callback_map.keys(): if len(args) > 0:
self._log("emmitting signal: %s\n" % (signal_name,)) if len(args) != len(arg_types):
# Don't bother if there are no callbacks. self._warn("Signal emitted with "
for (key,fn) in self.__callback_map[signal_name]: "wrong number of args: %s\n"
self._log("Calling callback with key: %s\n" % (key,)) " from: file: %s\n"
try: " line: %d\n"
if type(fn) == tuple: # call class method " func: %s\n"
cb[0](fn[1],*args) % ((str(signal_name),) + inspect.stack()[1][1:4]))
elif type(fn) == types.FunctionType or \ return
type(fn) == types.MethodType: # call func
fn(*args) if arg_types != None:
else: for i in range(0,len(arg_types)):
self._log("Warning: badly formed entry in callback map.\n") if not isinstance(args[i],arg_types[i]):
except: self._warn("Signal emitted with "
log("%s: %s" % (self.__class__.__name__, "wrong arg types: %s\n"
"Warning: exception occured in callback function.\n")) " from: file: %s\n"
log("%s: %s" % (self.__class__.__name__, " line: %d\n"
"".join(traceback.format_exception(*sys.exc_info())))) " func: %s\n"
" arg passed was: %s, type of arg passed %s, type should be: %s\n"
% ((str(signal_name),) + inspect.stack()[1][1:4] +\
(args[i],repr(type(args[i])),repr(arg_types[i]))))
return
if signal_name in self.__callback_map.keys():
self._log("emmitting signal: %s\n" % (signal_name,))
# Don't bother if there are no callbacks.
for (key,fn) in self.__callback_map[signal_name]:
self._log("Calling callback with key: %s\n" % (key,))
try:
if type(fn) == tuple: # call class method
cb[0](fn[1],*args)
elif type(fn) == types.FunctionType or \
type(fn) == types.MethodType: # call func
fn(*args)
else:
self._warn("Badly formed entry in callback map.\n")
except:
self._warn("Exception occured in callback function.\n"
"%s" % ("".join(traceback.format_exception(*sys.exc_info())),))
finally:
self._current_signals.remove(signal_name)
# #
# instance signals control methods # instance signals control methods
@ -408,7 +440,10 @@ class GrampsDBCallback(object):
def _log(self,msg): def _log(self,msg):
if GrampsDBCallback.__LOG_ALL or self.__enable_logging: if GrampsDBCallback.__LOG_ALL or self.__enable_logging:
log("%s: %s" % (self.__class__.__name__, str(msg))) log("%s: %s" % (self.__class__.__name__, str(msg)))
def _warn(self,msg):
log("Warning: %s: %s" % (self.__class__.__name__, str(msg)))
# #
# Class methods # Class methods
# #
@ -655,9 +690,70 @@ if __name__ == "__main__":
res=[] res=[]
def fn(s,r=res): def fn(s,r=res):
res.append(s) res.append(s)
t._log = fn t._warn = fn
t.connect('test-lots',fn2), t.emit('test-lots',('a','a',[1,2],t,1.2)) t.connect('test-lots',fn2), t.emit('test-lots',('a','a',[1,2],t,1.2))
assert res[1][0:8] == "Warning:", "Type error not detected" assert res[0][0:6] == "Signal", "Type error not detected"
def test_recursion_block(self):
class TestSignals(GrampsDBCallback):
__signals__ = {
'test-recursion' : (GrampsDBCallback,)
}
def fn(cb):
cb.emit('test-recursion',(t,))
res=[]
def fn2(s,r=res):
res.append(s)
t = TestSignals()
t._warn = fn2
t.connect('test-recursion',fn)
try:
t.emit('test-recursion',(t,))
except RuntimeError:
assert False, "signal recursion not blocked1."
assert res[0][0:6] == "Signal", "signal recursion not blocked"
def test_multisignal_recursion_block(self):
class TestSignals(GrampsDBCallback):
__signals__ = {
'test-top' : (GrampsDBCallback,),
'test-middle' : (GrampsDBCallback,),
'test-bottom' : (GrampsDBCallback,)
}
def top(cb):
cb.emit('test-middle',(t,))
def middle(cb):
cb.emit('test-bottom',(t,))
def bottom(cb):
cb.emit('test-top',(t,))
res=[]
def fn2(s,r=res):
res.append(s)
t = TestSignals()
t._warn = fn2
t.connect('test-top',top)
t.connect('test-middle',middle)
t.connect('test-bottom',bottom)
try:
t.emit('test-top',(t,))
except RuntimeError:
assert False, "multisignal recursion not blocked1."
assert res[0][0:6] == "Signal", "multisignal recursion not blocked"
unittest.main() unittest.main()