test/test-signals.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
   4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
   5 #
   6 # Licensed under the Academic Free License version 2.1
   7 #
   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.
  14 #
  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.
  19 #
  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
  23
  24
  25 import sys
  26 import os
  27 import unittest
  28 import time
  29 import logging
  30
  31 builddir = os.environ["DBUS_TOP_BUILDDIR"]
  32 pydir = builddir
  33
  34 sys.path.insert(0, pydir)
  35 sys.path.insert(0, pydir + 'dbus')
  36
  37 import dbus
  38 import _dbus_bindings
  39 import gobject
  40 import dbus.glib
  41 import dbus.service
  42
  43
  44 logging.basicConfig()
  45 logging.getLogger().setLevel(1)
  46 logger = logging.getLogger('test-signals')
  47
  48
  49 pkg = dbus.__file__
  50 if not pkg.startswith(pydir):
  51     raise Exception("DBus modules (%s) are not being picked up from the package"%pkg)
  52
  53 if not _dbus_bindings.__file__.startswith(pydir):
  54     raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__)
  55
  56 class TestSignals(unittest.TestCase):
  57     def setUp(self):
  58         logger.info('setUp()')
  59         self.bus = dbus.SessionBus()
  60         self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
  61         self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.TestSuiteInterface")
  62         self.in_test = None
  63
  64     def signal_test_impl(self, name, test_removal=False):
  65         self.in_test = name
  66         # using append rather than assignment here to avoid scoping issues
  67         result = []
  68
  69         def _timeout_handler():
  70             logger.debug('_timeout_handler for %s: current state %s', name, self.in_test)
  71             if self.in_test == name:
  72                 main_loop.quit()
  73         def _signal_handler(s, sender, path):
  74             logger.debug('_signal_handler for %s: current state %s', name, self.in_test)
  75             if self.in_test not in (name, name + '+removed'):
  76                 return
  77             logger.info('Received signal from %s:%s, argument is %r',
  78                         sender, path, s)
  79             result.append('received')
  80             main_loop.quit()
  81         def _rm_timeout_handler():
  82             logger.debug('_timeout_handler for %s: current state %s', name, self.in_test)
  83             if self.in_test == name + '+removed':
  84                 main_loop.quit()
  85
  86         logger.info('Testing %s', name)
  87         match = self.iface.connect_to_signal('SignalOneString', _signal_handler,
  88                                              sender_keyword='sender',
  89                                              path_keyword='path')
  90         logger.info('Waiting for signal...')
  91         self.iface.EmitSignal('SignalOneString', 0)
  92         source_id = gobject.timeout_add(1000, _timeout_handler)
  93         main_loop.run()
  94         if not result:
  95             raise AssertionError('Signal did not arrive within 1 second')
  96         logger.debug('Removing match')
  97         match.remove()
  98         gobject.source_remove(source_id)
  99         if test_removal:
 100             self.in_test = name + '+removed'
 101             logger.info('Testing %s', name)
 102             result = []
 103             self.iface.EmitSignal('SignalOneString', 0)
 104             source_id = gobject.timeout_add(1000, _rm_timeout_handler)
 105             main_loop.run()
 106             if result:
 107                 raise AssertionError('Signal should not have arrived, but did')
 108             gobject.source_remove(source_id)
 109
 110     def testSignal(self):
 111         self.signal_test_impl('Signal')
 112
 113     def testRemoval(self):
 114         self.signal_test_impl('Removal', True)
 115
 116     def testSignalAgain(self):
 117         self.signal_test_impl('SignalAgain')
 118
 119     def testRemovalAgain(self):
 120         self.signal_test_impl('RemovalAgain', True)
 121
 122 if __name__ == '__main__':
 123     main_loop = gobject.MainLoop()
 124     gobject.threads_init()
 125     dbus.glib.init_threads()
 126
 127     logger.info('Starting unit test')
 128     unittest.main()