python running glib mainloop in unittest

Question

I'm new to python and trying to set up a functional test environment. The test environment shall receive signals over DBus and evaluate them. The DBus signaling uses the GLib MainLoop. I have the following class encapsulating the loop:

class SignalLoop(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__l = gobject.MainLoop()        
    def run(self):
        self.__l.run()
    def quit(self):
        self.__l.quit() 

and in the module doing the DBus handling I tried:

class __ModuleInitializer:
    def __init__(self):
        print('Module was initialized')
        gobject.threads_init()
        sl = SignalLoop()
        sl.start()

    def __del__(self):
        print('Module was deinitialized')
        sl.quit()

__module_init = __ModuleInitializer()

I also tried the following:

• setUp/tearDownModule() does not work for me - at least in python 2.5.2
• __init__() and __del__() methods and putting all the testcases in a single class. The __del__() never get called and this solution will not scale with a lot of test cases.

When I run the code, the sl.qui() never gets executed and I don't know why. I have to kill my main when running from console because it never returns. However this is not a problem for the PyDev test runner.

Is there any hook which I can use for destructing the test environment? I want to be able to run a single test and a couple of tests as well, so "hacking" it into the methods itself is not an option.

Can you please help me?

Answer 1

I would not start and stop the main loop during the setup and teardown methods. Instead, make your asynchronous request (or whatever), then run the main loop iteration by iteration until your result arrives.

Example:

make_asynchronous_request()
main_context = GLib.MainContext.default()
while main_context.pending():
    main_context.iteration(False)
assert(result_is_ok())

Answer 2

I used ptomato's answer to create one that worked for me. The difference is probably due to some python version. In my case, the code that works is:

make_asynchronous_request()
main_context = GLib.MainLoop.get_context()
while main_context.pending():
    main_context.iteration(False)
assert(result_is_ok())