tests/test_signal.py

   1 # -*- Mode: Python -*-
   2
   3 import gc
   4 import unittest
   5 import sys
   6
   7 from common import gobject, testhelper
   8
   9 class C(gobject.GObject):
  10     __gsignals__ = { 'my_signal': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE,
  11                                    (gobject.TYPE_INT,)) }
  12     def do_my_signal(self, arg):
  13         self.arg = arg
  14
  15 class D(C):
  16     def do_my_signal(self, arg2):
  17         self.arg2 = arg2
  18         C.do_my_signal(self, arg2)
  19
  20 class TestSignalCreation(unittest.TestCase):
  21     # Bug 540376.
  22     def test_illegals(self):
  23         self.assertRaises(TypeError, lambda: gobject.signal_new('test',
  24                                                                 None,
  25                                                                 0,
  26                                                                 gobject.TYPE_NONE,
  27                                                                 (gobject.TYPE_LONG,)))
  28
  29
  30 class TestChaining(unittest.TestCase):
  31     def setUp(self):
  32         self.inst = C()
  33         self.inst.connect("my_signal", self.my_signal_handler_cb, 1, 2, 3)
  34
  35     def my_signal_handler_cb(self, *args):
  36         assert len(args) == 5
  37         assert isinstance(args[0], C)
  38         assert args[0] == self.inst
  39
  40         assert isinstance(args[1], int)
  41         assert args[1] == 42
  42
  43         assert args[2:] == (1, 2, 3)
  44
  45     def testChaining(self):
  46         self.inst.emit("my_signal", 42)
  47         assert self.inst.arg == 42
  48
  49     def testChaining(self):
  50         inst2 = D()
  51         inst2.emit("my_signal", 44)
  52         assert inst2.arg == 44
  53         assert inst2.arg2 == 44
  54
  55 # This is for bug 153718
  56 class TestGSignalsError(unittest.TestCase):
  57     def testInvalidType(self, *args):
  58         def foo():
  59             class Foo(gobject.GObject):
  60                 __gsignals__ = None
  61         self.assertRaises(TypeError, foo)
  62         gc.collect()
  63
  64     def testInvalidName(self, *args):
  65         def foo():
  66             class Foo(gobject.GObject):
  67                 __gsignals__ = {'not-exists' : 'override'}
  68         self.assertRaises(TypeError, foo)
  69         gc.collect()
  70
  71 class TestGPropertyError(unittest.TestCase):
  72     def testInvalidType(self, *args):
  73         def foo():
  74             class Foo(gobject.GObject):
  75                 __gproperties__ = None
  76         self.assertRaises(TypeError, foo)
  77         gc.collect()
  78
  79     def testInvalidName(self, *args):
  80         def foo():
  81             class Foo(gobject.GObject):
  82                 __gproperties__ = { None: None }
  83
  84         self.assertRaises(TypeError, foo)
  85         gc.collect()
  86
  87 class TestList(unittest.TestCase):
  88     def testListObject(self):
  89         self.assertEqual(gobject.signal_list_names(C), ('my-signal',))
  90
  91
  92 def my_accumulator(ihint, return_accu, handler_return, user_data):
  93     """An accumulator that stops emission when the sum of handler
  94     returned values reaches 3"""
  95     assert user_data == "accum data"
  96     if return_accu >= 3:
  97         return False, return_accu
  98     return True, return_accu + handler_return
  99
 100 class Foo(gobject.GObject):
 101     __gsignals__ = {
 102         'my-acc-signal': (gobject.SIGNAL_RUN_LAST, gobject.TYPE_INT,
 103                                    (), my_accumulator, "accum data"),
 104         'my-other-acc-signal': (gobject.SIGNAL_RUN_LAST, gobject.TYPE_BOOLEAN,
 105                                 (), gobject.signal_accumulator_true_handled)
 106         }
 107
 108 class TestAccumulator(unittest.TestCase):
 109
 110     def testAccumulator(self):
 111         inst = Foo()
 112         inst.connect("my-acc-signal", lambda obj: 1)
 113         inst.connect("my-acc-signal", lambda obj: 2)
 114         ## the value returned in the following handler will not be
 115         ## considered, because at this point the accumulator already
 116         ## reached its limit.
 117         inst.connect("my-acc-signal", lambda obj: 3)
 118         retval = inst.emit("my-acc-signal")
 119         self.assertEqual(retval, 3)
 120
 121     def testAccumulatorTrueHandled(self):
 122         inst = Foo()
 123         inst.connect("my-other-acc-signal", self._true_handler1)
 124         inst.connect("my-other-acc-signal", self._true_handler2)
 125         ## the following handler will not be called because handler2
 126         ## returns True, so it should stop the emission.
 127         inst.connect("my-other-acc-signal", self._true_handler3)
 128         self.__true_val = None
 129         inst.emit("my-other-acc-signal")
 130         self.assertEqual(self.__true_val, 2)
 131
 132     def _true_handler1(self, obj):
 133         self.__true_val = 1
 134         return False
 135     def _true_handler2(self, obj):
 136         self.__true_val = 2
 137         return True
 138     def _true_handler3(self, obj):
 139         self.__true_val = 3
 140         return False
 141
 142 class E(gobject.GObject):
 143     __gsignals__ = { 'signal': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE,
 144                                 ()) }
 145     def __init__(self):
 146         gobject.GObject.__init__(self)
 147         self.status = 0
 148
 149     def do_signal(self):
 150         assert self.status == 0
 151         self.status = 1
 152
 153 class F(gobject.GObject):
 154     __gsignals__ = { 'signal': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE,
 155                                 ()) }
 156     def __init__(self):
 157         gobject.GObject.__init__(self)
 158         self.status = 0
 159
 160     def do_signal(self):
 161         self.status += 1
 162
 163 class TestEmissionHook(unittest.TestCase):
 164     def testAdd(self):
 165         self.hook = True
 166         e = E()
 167         e.connect('signal', self._callback)
 168         gobject.add_emission_hook(E, "signal", self._emission_hook)
 169         e.emit('signal')
 170         self.assertEqual(e.status, 3)
 171
 172     def testRemove(self):
 173         self.hook = False
 174         e = E()
 175         e.connect('signal', self._callback)
 176         hook_id = gobject.add_emission_hook(E, "signal", self._emission_hook)
 177         gobject.remove_emission_hook(E, "signal", hook_id)
 178         e.emit('signal')
 179         self.assertEqual(e.status, 3)
 180
 181     def _emission_hook(self, e):
 182         self.assertEqual(e.status, 1)
 183         e.status = 2
 184
 185     def _callback(self, e):
 186         if self.hook:
 187             self.assertEqual(e.status, 2)
 188         else:
 189             self.assertEqual(e.status, 1)
 190         e.status = 3
 191
 192     def testCallbackReturnFalse(self):
 193         self.hook = False
 194         obj = F()
 195         def _emission_hook(obj):
 196             obj.status += 1
 197             return False
 198         hook_id = gobject.add_emission_hook(obj, "signal", _emission_hook)
 199         obj.emit('signal')
 200         obj.emit('signal')
 201         self.assertEqual(obj.status, 3)
 202
 203     def testCallbackReturnTrue(self):
 204         self.hook = False
 205         obj = F()
 206         def _emission_hook(obj):
 207             obj.status += 1
 208             return True
 209         hook_id = gobject.add_emission_hook(obj, "signal", _emission_hook)
 210         obj.emit('signal')
 211         obj.emit('signal')
 212         gobject.remove_emission_hook(obj, "signal", hook_id)
 213         self.assertEqual(obj.status, 4)
 214
 215     def testCallbackReturnTrueButRemove(self):
 216         self.hook = False
 217         obj = F()
 218         def _emission_hook(obj):
 219             obj.status += 1
 220             return True
 221         hook_id = gobject.add_emission_hook(obj, "signal", _emission_hook)
 222         obj.emit('signal')
 223         gobject.remove_emission_hook(obj, "signal", hook_id)
 224         obj.emit('signal')
 225         self.assertEqual(obj.status, 3)
 226
 227 class TestClosures(unittest.TestCase):
 228     def setUp(self):
 229         self.count = 0
 230
 231     def _callback(self, e):
 232         self.count += 1
 233
 234     def testDisconnect(self):
 235         e = E()
 236         e.connect('signal', self._callback)
 237         e.disconnect_by_func(self._callback)
 238         e.emit('signal')
 239         self.assertEqual(self.count, 0)
 240
 241     def testHandlerBlock(self):
 242         e = E()
 243         e.connect('signal', self._callback)
 244         e.handler_block_by_func(self._callback)
 245         e.emit('signal')
 246         self.assertEqual(self.count, 0)
 247
 248     def testHandlerUnBlock(self):
 249         e = E()
 250         signal_id = e.connect('signal', self._callback)
 251         e.handler_block(signal_id)
 252         e.handler_unblock_by_func(self._callback)
 253         e.emit('signal')
 254         self.assertEqual(self.count, 1)
 255
 256     def testHandlerBlockMethod(self):
 257         # Filed as #375589
 258         class A:
 259             def __init__(self):
 260                 self.a = 0
 261
 262             def callback(self, o):
 263                 self.a = 1
 264                 o.handler_block_by_func(self.callback)
 265
 266         inst = A()
 267         e = E()
 268         e.connect("signal", inst.callback)
 269         e.emit('signal')
 270         self.assertEqual(inst.a, 1)
 271         gc.collect()
 272
 273     def testGString(self):
 274         class C(gobject.GObject):
 275             __gsignals__ = { 'my_signal': (gobject.SIGNAL_RUN_LAST, gobject.TYPE_GSTRING,
 276                                            (gobject.TYPE_GSTRING,)) }
 277             def __init__(self, test):
 278                 gobject.GObject.__init__(self)
 279                 self.test = test
 280             def do_my_signal(self, data):
 281                 self.data = data
 282                 self.test.assertEqual(len(data), 3)
 283                 return ''.join([data[2], data[1], data[0]])
 284         c = C(self)
 285         data = c.emit("my_signal", "\01\00\02")
 286         self.assertEqual(data, "\02\00\01")
 287
 288 class SigPropClass(gobject.GObject):
 289     __gsignals__ = { 'my_signal': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE,
 290                                    (gobject.TYPE_INT,)) }
 291
 292     __gproperties__ = {
 293         'foo': (str, None, None, '', gobject.PARAM_WRITABLE|gobject.PARAM_CONSTRUCT),
 294         }
 295
 296     signal_emission_failed = False
 297
 298     def do_my_signal(self, arg):
 299         self.arg = arg
 300
 301     def do_set_property(self, pspec, value):
 302         if pspec.name == 'foo':
 303             self._foo = value
 304         else:
 305             raise AttributeError, 'unknown property %s' % pspec.name
 306         try:
 307             self.emit("my-signal", 1)
 308         except TypeError:
 309             self.signal_emission_failed = True
 310
 311
 312 class TestSigProp(unittest.TestCase):
 313     def testEmitInPropertySetter(self):
 314         obj = SigPropClass()
 315         self.failIf(obj.signal_emission_failed)
 316
 317 f = gobject.SIGNAL_RUN_FIRST
 318 l = gobject.SIGNAL_RUN_LAST
 319 float = gobject.TYPE_FLOAT
 320 double = gobject.TYPE_DOUBLE
 321 uint = gobject.TYPE_UINT
 322 ulong = gobject.TYPE_ULONG
 323
 324 class CM(gobject.GObject):
 325     __gsignals__ = dict(
 326         test1=(f, None, ()),
 327         test2=(l, None, (str,)),
 328         test3=(l, int, (double,)),
 329         test4=(f, None, (bool, long, float, double, int, uint, ulong)),
 330         test_float=(l, float, (float,)),
 331         test_double=(l, double, (double, )),
 332         test_string=(l, str, (str, )),
 333         test_object=(l, object, (object, )),
 334     )
 335
 336 class _TestCMarshaller:
 337     def setUp(self):
 338         self.obj = CM()
 339         testhelper.connectcallbacks(self.obj)
 340
 341     def testTest1(self):
 342         self.obj.emit("test1")
 343
 344     def testTest2(self):
 345         self.obj.emit("test2", "string")
 346
 347     def testTest3(self):
 348         rv = self.obj.emit("test3", 42.0)
 349         self.assertEqual(rv, 20)
 350
 351     def testTest4(self):
 352         self.obj.emit("test4", True, 10L, 3.14, 1.78, 20, 30L, 31L)
 353
 354     def testTestReturnFloat(self):
 355         rv = self.obj.emit("test-float", 1.234)
 356         self.failUnless(rv >= 1.233999 and rv <= 1.2400001, rv)
 357
 358     def testTestReturnDouble(self):
 359         rv = self.obj.emit("test-double", 1.234)
 360         self.assertEqual(rv, 1.234)
 361
 362     def testTestReturnString(self):
 363         rv = self.obj.emit("test-string", "str")
 364         self.assertEqual(rv, "str")
 365
 366     def testTestReturnObject(self):
 367         rv = self.obj.emit("test-object", self)
 368         self.assertEqual(rv, self)
 369
 370 if 'generic-c-marshaller' in gobject.features:
 371     class TestCMarshaller(_TestCMarshaller, unittest.TestCase):
 372         pass
 373 else:
 374     print
 375     print '** WARNING: LIBFFI disabled, not testing'
 376     print
 377
 378 # Test for 374653
 379 class TestPyGValue(unittest.TestCase):
 380     def testNoneNULLBoxedConversion(self):
 381         class C(gobject.GObject):
 382             __gsignals__ = dict(my_boxed_signal=(
 383                 gobject.SIGNAL_RUN_LAST,
 384                 gobject.type_from_name('GStrv'), ()))
 385
 386         obj = C()
 387         obj.connect('my-boxed-signal', lambda obj: None)
 388         sys.last_type = None
 389         obj.emit('my-boxed-signal')
 390         assert not sys.last_type
 391
 392 if __name__ == '__main__':
 393     unittest.main()