1 #!/usr/bin/env python 2 3 import os,sys 4 5 try: 6 import gobject 7 import dbus 8 import dbus.mainloop.glib 9 except: 10 print "Failed import, aborting test" 11 sys.exit(0) 12 13 dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) 14 loop = gobject.MainLoop() 15 16 exitcode = 0 17 18 bus = dbus.SessionBus() 19 bus_iface = dbus.Interface(bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus'), 'org.freedesktop.DBus') 20 21 o = bus.get_object('org.freedesktop.DBus.TestSuiteForkingEchoService', '/org/freedesktop/TestSuite') 22 i = dbus.Interface(o, 'org.freedesktop.TestSuite') 23 24 # Start it up 25 reply = i.Echo("hello world") 26 print "TestSuiteForkingEchoService initial reply OK" 27 28 def ignore(*args, **kwargs): 29 pass 30 31 # Now monitor for exits, when that happens, start it up again. 32 # The goal here is to try to hit any race conditions in activation. 33 counter = 0 34 def on_forking_echo_owner_changed(name, old, new): 35 global counter 36 global o 37 global i 38 if counter > 10: 39 print "Activated 10 times OK, TestSuiteForkingEchoService pass" 40 loop.quit() 41 return 42 counter += 1 43 if new == '': 44 o = bus.get_object('org.freedesktop.DBus.TestSuiteForkingEchoService', '/org/freedesktop/TestSuite') 45 i = dbus.Interface(o, 'org.freedesktop.TestSuite') 46 i.Echo("counter %r" % counter) 47 i.Exit(reply_handler=ignore, error_handler=ignore) 48 49 bus_iface.connect_to_signal('NameOwnerChanged', on_forking_echo_owner_changed, arg0='org.freedesktop.DBus.TestSuiteForkingEchoService') 50 51 i.Exit(reply_handler=ignore, error_handler=ignore) 52 53 def check_counter(): 54 if counter == 0: 55 print "Failed to get NameOwnerChanged for TestSuiteForkingEchoService" 56 sys.exit(1) 57 gobject.timeout_add(15000, check_counter) 58 59 loop.run() 60 sys.exit(0) 61
