3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
6 # Licensed under the Academic Free License version 2.1
8 # This program is free software; you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # This program is free software; you can redistribute it and/or modify
11 # it under the terms of the GNU General Public License as published by
12 # the Free Software Foundation; either version 2 of the License, or
13 # (at your option) any later version.
15 # This program is distributed in the hope that it will be useful,
16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 # GNU General Public License for more details.
20 # You should have received a copy of the GNU General Public License
21 # along with this program; if not, write to the Free Software
22 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
31 builddir
= os
.path
.normpath(os
.environ
["DBUS_TOP_BUILDDIR"])
32 pydir
= os
.path
.normpath(os
.environ
["DBUS_TOP_SRCDIR"])
42 logging
.getLogger().setLevel(1)
43 logger
= logging
.getLogger('test-signals')
47 if not pkg
.startswith(pydir
):
48 raise Exception("DBus modules (%s) are not being picked up from the package"%pkg
)
50 if not _dbus_bindings
.__file
__.startswith(builddir
):
51 raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings
.__file
__)
53 class TestSignals(unittest
.TestCase
):
55 logger
.info('setUp()')
56 self
.bus
= dbus
.SessionBus()
57 self
.remote_object
= self
.bus
.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
58 self
.iface
= dbus
.Interface(self
.remote_object
, "org.freedesktop.DBus.TestSuiteInterface")
61 def signal_test_impl(self
, name
, test_removal
=False):
63 # using append rather than assignment here to avoid scoping issues
66 def _timeout_handler():
67 logger
.debug('_timeout_handler for %s: current state %s', name
, self
.in_test
)
68 if self
.in_test
== name
:
70 def _signal_handler(s
, sender
, path
):
71 logger
.debug('_signal_handler for %s: current state %s', name
, self
.in_test
)
72 if self
.in_test
not in (name
, name
+ '+removed'):
74 logger
.info('Received signal from %s:%s, argument is %r',
76 result
.append('received')
78 def _rm_timeout_handler():
79 logger
.debug('_timeout_handler for %s: current state %s', name
, self
.in_test
)
80 if self
.in_test
== name
+ '+removed':
83 logger
.info('Testing %s', name
)
84 match
= self
.iface
.connect_to_signal('SignalOneString', _signal_handler
,
85 sender_keyword
='sender',
87 logger
.info('Waiting for signal...')
88 self
.iface
.EmitSignal('SignalOneString', 0)
89 source_id
= gobject
.timeout_add(1000, _timeout_handler
)
92 raise AssertionError('Signal did not arrive within 1 second')
93 logger
.debug('Removing match')
95 gobject
.source_remove(source_id
)
97 self
.in_test
= name
+ '+removed'
98 logger
.info('Testing %s', name
)
100 self
.iface
.EmitSignal('SignalOneString', 0)
101 source_id
= gobject
.timeout_add(1000, _rm_timeout_handler
)
104 raise AssertionError('Signal should not have arrived, but did')
105 gobject
.source_remove(source_id
)
107 def testSignal(self
):
108 self
.signal_test_impl('Signal')
110 def testRemoval(self
):
111 self
.signal_test_impl('Removal', True)
113 def testSignalAgain(self
):
114 self
.signal_test_impl('SignalAgain')
116 def testRemovalAgain(self
):
117 self
.signal_test_impl('RemovalAgain', True)
119 if __name__
== '__main__':
120 main_loop
= gobject
.MainLoop()
121 gobject
.threads_init()
122 dbus
.glib
.init_threads()
124 logger
.info('Starting unit test')