D-Bus testing: improved test timeouts and looping in GLib

It was observed that raising an exception in the signal handler did
not propagate to the calling Python code when it was currently in
loop.get_context().iteration().

This commit adds debug logging for timeouts and some utility methods
which avoid the problem by adding logging to the iteration()
call. Python then re-raises the exception in the D-Bus logging code.

This looks like a deficiency (bug?!) of the Python bindings for
iteration() which is merely worked around.

However, the utility methods may also be useful by their own right,
for example to avoid the awkward loop.run() + loop.quit() pairs.
This commit is contained in:
Patrick Ohly 2012-09-25 10:45:17 +02:00
parent f6e979c2fd
commit dab4c96f00
1 changed files with 25 additions and 2 deletions

View File

@ -555,12 +555,14 @@ class DBusUtil(Timeout):
timeout = self.getTestProperty("timeout", defTimeout)
timeout_handle = None
if timeout and not debugger and not os.environ.get("SYNCEVOLUTION_LOCAL_CHILD_DELAY", None):
logging.printf('killing test in %d seconds' % timeout)
def timedout():
error = "%s timed out after %d seconds, current quit events: %s" % (self.id(), timeout, self.quit_events)
if Timeout.debugTimeout:
print error
logging.log('%s\n%s' % (error, ''.join(traceback.format_stack())))
raise Exception(error)
timeout_handle = self.addTimeout(timeout, timedout, use_glib=False)
else:
logging.printf('killing test disabled')
try:
self.running = True
unittest.TestCase.run(self, result)
@ -628,6 +630,27 @@ class DBusUtil(Timeout):
# same for failure
result.failures[-1] = (result.failures[-1][0], result.failures[-1][1] + report)
def loopIteration(self, message, may_block=True):
'''Trigger glib, allow it to wait once (optional).'''
# Calling this is necessary, Python uses it to re-raise a
# timeout exception that was originally raised inside the C
# glib code. If None, the caller must do it.
if message:
logging.log(message)
loop.get_context().iteration(may_block)
def runUntil(self, state, check, until):
'''Loop until 'check' throws an exception or 'until' returns True.
Use check=lambda: (expr1, expr2, ...) when more than one check is needed.
'''
message = 'waiting for ' + state
while True:
logging.log(message)
check()
if until():
break
self.loopIteration(None)
def isServerRunning(self):
"""True while the syncevo-dbus-server executable is still running"""
return DBusUtil.pserver and DBusUtil.pserver.poll() == None