File: test-activation-forking.py

package info (click to toggle)
dbus 1.2.24-4%2Bsqueeze3
  • links: PTS, VCS
  • area: main
  • in suites: squeeze-lts
  • size: 9,420 kB
  • ctags: 6,204
  • sloc: ansic: 72,608; sh: 10,562; xml: 6,322; makefile: 640; python: 72
file content (60 lines) | stat: -rw-r--r-- 1,765 bytes parent folder | download | duplicates (5)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
#!/usr/bin/env python

import os,sys

try:
    import gobject
    import dbus
    import dbus.mainloop.glib
except:
    print "Failed import, aborting test"
    sys.exit(0)

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
loop = gobject.MainLoop()

exitcode = 0

bus = dbus.SessionBus()
bus_iface = dbus.Interface(bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus'), 'org.freedesktop.DBus')

o = bus.get_object('org.freedesktop.DBus.TestSuiteForkingEchoService', '/org/freedesktop/TestSuite')
i = dbus.Interface(o, 'org.freedesktop.TestSuite')

# Start it up
reply = i.Echo("hello world")
print "TestSuiteForkingEchoService initial reply OK"

def ignore(*args, **kwargs):
    pass

# Now monitor for exits, when that happens, start it up again.
# The goal here is to try to hit any race conditions in activation.
counter = 0
def on_forking_echo_owner_changed(name, old, new):
    global counter
    global o
    global i
    if counter > 10:
        print "Activated 10 times OK, TestSuiteForkingEchoService pass"
        loop.quit()
        return
    counter += 1
    if new == '':
        o = bus.get_object('org.freedesktop.DBus.TestSuiteForkingEchoService', '/org/freedesktop/TestSuite')
        i = dbus.Interface(o, 'org.freedesktop.TestSuite')
        i.Echo("counter %r" % counter)
        i.Exit(reply_handler=ignore, error_handler=ignore)

bus_iface.connect_to_signal('NameOwnerChanged', on_forking_echo_owner_changed, arg0='org.freedesktop.DBus.TestSuiteForkingEchoService')

i.Exit(reply_handler=ignore, error_handler=ignore)

def check_counter():
    if counter == 0:
        print "Failed to get NameOwnerChanged for TestSuiteForkingEchoService"
        sys.exit(1)
gobject.timeout_add(15000, check_counter)

loop.run()
sys.exit(0)