tests/test_signal.py

   1 # -*- Mode: Python -*-
   2
   3 import gc
   4 import unittest
   5 import sys
   6 import weakref
   7
   8 from gi.repository import GObject, GLib
   9 from gi import _signalhelper as signalhelper
  10 import testhelper
  11 from compathelper import _long
  12
  13 try:
  14     import cairo
  15     cairo  # PyFlakes
  16     from gi.repository import Regress
  17     has_cairo = True
  18 except ImportError:
  19     has_cairo = False
  20
  21
  22 class C(GObject.GObject):
  23     __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
  24                                   (GObject.TYPE_INT,))}
  25
  26     def do_my_signal(self, arg):
  27         self.arg = arg
  28
  29
  30 class D(C):
  31     def do_my_signal(self, arg2):
  32         self.arg2 = arg2
  33         C.do_my_signal(self, arg2)
  34
  35
  36 class TestSignalCreation(unittest.TestCase):
  37     # Bug 540376.
  38     def test_illegals(self):
  39         self.assertRaises(TypeError, lambda: GObject.signal_new('test',
  40                                                                 None,
  41                                                                 0,
  42                                                                 None,
  43                                                                 (GObject.TYPE_LONG,)))
  44
  45
  46 class TestChaining(unittest.TestCase):
  47     def setUp(self):
  48         self.inst = C()
  49         self.inst.connect("my_signal", self.my_signal_handler_cb, 1, 2, 3)
  50
  51     def my_signal_handler_cb(self, *args):
  52         assert len(args) == 5
  53         assert isinstance(args[0], C)
  54         assert args[0] == self.inst
  55
  56         assert isinstance(args[1], int)
  57         assert args[1] == 42
  58
  59         assert args[2:] == (1, 2, 3)
  60
  61     def test_chaining(self):
  62         self.inst.emit("my_signal", 42)
  63         assert self.inst.arg == 42
  64
  65     def test_chaining2(self):
  66         inst2 = D()
  67         inst2.emit("my_signal", 44)
  68         assert inst2.arg == 44
  69         assert inst2.arg2 == 44
  70
  71 # This is for bug 153718
  72
  73
  74 class TestGSignalsError(unittest.TestCase):
  75     def test_invalid_type(self, *args):
  76         def foo():
  77             class Foo(GObject.GObject):
  78                 __gsignals__ = None
  79         self.assertRaises(TypeError, foo)
  80         gc.collect()
  81
  82     def test_invalid_name(self, *args):
  83         def foo():
  84             class Foo(GObject.GObject):
  85                 __gsignals__ = {'not-exists': 'override'}
  86         # do not stumble over the warning thrown by GLib
  87         old_mask = GLib.log_set_always_fatal(GLib.LogLevelFlags.LEVEL_CRITICAL |
  88                                              GLib.LogLevelFlags.LEVEL_ERROR)
  89         try:
  90             self.assertRaises(TypeError, foo)
  91         finally:
  92             GLib.log_set_always_fatal(old_mask)
  93         gc.collect()
  94
  95
  96 class TestGPropertyError(unittest.TestCase):
  97     def test_invalid_type(self, *args):
  98         def foo():
  99             class Foo(GObject.GObject):
 100                 __gproperties__ = None
 101         self.assertRaises(TypeError, foo)
 102         gc.collect()
 103
 104     def test_invalid_name(self, *args):
 105         def foo():
 106             class Foo(GObject.GObject):
 107                 __gproperties__ = {None: None}
 108
 109         self.assertRaises(TypeError, foo)
 110         gc.collect()
 111
 112
 113 class TestList(unittest.TestCase):
 114     def test_list_names(self):
 115         self.assertEqual(GObject.signal_list_names(C), ('my-signal',))
 116
 117
 118 def my_accumulator(ihint, return_accu, handler_return, user_data):
 119     """An accumulator that stops emission when the sum of handler
 120     returned values reaches 3"""
 121     assert user_data == "accum data"
 122     if return_accu >= 3:
 123         return False, return_accu
 124     return True, return_accu + handler_return
 125
 126
 127 class Foo(GObject.GObject):
 128     my_acc_signal = GObject.Signal(return_type=GObject.TYPE_INT,
 129                                    flags=GObject.SignalFlags.RUN_LAST,
 130                                    accumulator=my_accumulator,
 131                                    accu_data="accum data")
 132
 133     my_other_acc_signal = GObject.Signal(return_type=GObject.TYPE_BOOLEAN,
 134                                          flags=GObject.SignalFlags.RUN_LAST,
 135                                          accumulator=GObject.signal_accumulator_true_handled)
 136
 137     my_acc_first_wins = GObject.Signal(return_type=GObject.TYPE_BOOLEAN,
 138                                        flags=GObject.SignalFlags.RUN_LAST,
 139                                        accumulator=GObject.signal_accumulator_first_wins)
 140
 141
 142 class TestAccumulator(unittest.TestCase):
 143
 144     def test_accumulator(self):
 145         inst = Foo()
 146         inst.my_acc_signal.connect(lambda obj: 1)
 147         inst.my_acc_signal.connect(lambda obj: 2)
 148         # the value returned in the following handler will not be
 149         # considered, because at this point the accumulator already
 150         # reached its limit.
 151         inst.my_acc_signal.connect(lambda obj: 3)
 152         retval = inst.my_acc_signal.emit()
 153         self.assertEqual(retval, 3)
 154
 155     def test_accumulator_true_handled(self):
 156         inst = Foo()
 157         inst.my_other_acc_signal.connect(self._true_handler1)
 158         inst.my_other_acc_signal.connect(self._true_handler2)
 159         # the following handler will not be called because handler2
 160         # returns True, so it should stop the emission.
 161         inst.my_other_acc_signal.connect(self._true_handler3)
 162         self.__true_val = None
 163         inst.my_other_acc_signal.emit()
 164         self.assertEqual(self.__true_val, 2)
 165
 166     def test_accumulator_first_wins(self):
 167         # First signal hit will always win
 168         inst = Foo()
 169         inst.my_acc_first_wins.connect(self._true_handler3)
 170         inst.my_acc_first_wins.connect(self._true_handler1)
 171         inst.my_acc_first_wins.connect(self._true_handler2)
 172         self.__true_val = None
 173         inst.my_acc_first_wins.emit()
 174         self.assertEqual(self.__true_val, 3)
 175
 176     def _true_handler1(self, obj):
 177         self.__true_val = 1
 178         return False
 179
 180     def _true_handler2(self, obj):
 181         self.__true_val = 2
 182         return True
 183
 184     def _true_handler3(self, obj):
 185         self.__true_val = 3
 186         return False
 187
 188
 189 class E(GObject.GObject):
 190     __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
 191                                ())}
 192
 193     # Property used to test detailed signal
 194     prop = GObject.Property(type=int, default=0)
 195
 196     def __init__(self):
 197         GObject.GObject.__init__(self)
 198         self.status = 0
 199
 200     def do_signal(self):
 201         assert self.status == 0
 202         self.status = 1
 203
 204
 205 class F(GObject.GObject):
 206     __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
 207                                ())}
 208
 209     def __init__(self):
 210         GObject.GObject.__init__(self)
 211         self.status = 0
 212
 213     def do_signal(self):
 214         self.status += 1
 215
 216
 217 class TestEmissionHook(unittest.TestCase):
 218     def test_add(self):
 219         self.hook = True
 220         e = E()
 221         e.connect('signal', self._callback)
 222         GObject.add_emission_hook(E, "signal", self._emission_hook)
 223         e.emit('signal')
 224         self.assertEqual(e.status, 3)
 225
 226     def test_remove(self):
 227         self.hook = False
 228         e = E()
 229         e.connect('signal', self._callback)
 230         hook_id = GObject.add_emission_hook(E, "signal", self._emission_hook)
 231         GObject.remove_emission_hook(E, "signal", hook_id)
 232         e.emit('signal')
 233         self.assertEqual(e.status, 3)
 234
 235     def _emission_hook(self, e):
 236         self.assertEqual(e.status, 1)
 237         e.status = 2
 238
 239     def _callback(self, e):
 240         if self.hook:
 241             self.assertEqual(e.status, 2)
 242         else:
 243             self.assertEqual(e.status, 1)
 244         e.status = 3
 245
 246     def test_callback_return_false(self):
 247         self.hook = False
 248         obj = F()
 249
 250         def _emission_hook(obj):
 251             obj.status += 1
 252             return False
 253         GObject.add_emission_hook(obj, "signal", _emission_hook)
 254         obj.emit('signal')
 255         obj.emit('signal')
 256         self.assertEqual(obj.status, 3)
 257
 258     def test_callback_return_true(self):
 259         self.hook = False
 260         obj = F()
 261
 262         def _emission_hook(obj):
 263             obj.status += 1
 264             return True
 265         hook_id = GObject.add_emission_hook(obj, "signal", _emission_hook)
 266         obj.emit('signal')
 267         obj.emit('signal')
 268         GObject.remove_emission_hook(obj, "signal", hook_id)
 269         self.assertEqual(obj.status, 4)
 270
 271     def test_callback_return_true_but_remove(self):
 272         self.hook = False
 273         obj = F()
 274
 275         def _emission_hook(obj):
 276             obj.status += 1
 277             return True
 278         hook_id = GObject.add_emission_hook(obj, "signal", _emission_hook)
 279         obj.emit('signal')
 280         GObject.remove_emission_hook(obj, "signal", hook_id)
 281         obj.emit('signal')
 282         self.assertEqual(obj.status, 3)
 283
 284
 285 class TestMatching(unittest.TestCase):
 286     class Object(GObject.Object):
 287         status = 0
 288         prop = GObject.Property(type=int, default=0)
 289
 290         @GObject.Signal()
 291         def my_signal(self):
 292             pass
 293
 294     @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=692918
 295     def test_signal_handler_block_matching(self):
 296         def dummy(*args):
 297             "Hack to work around: "
 298
 299         def foo(obj):
 300             obj.status += 1
 301
 302         obj = self.Object()
 303         handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)
 304         handler_id
 305
 306         self.assertEqual(obj.status, 0)
 307         obj.emit('my-signal')
 308         self.assertEqual(obj.status, 1)
 309
 310         # Blocking by match criteria disables the foo callback
 311         signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
 312         count = GObject.signal_handlers_block_matched(obj,
 313                                                       GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
 314                                                       signal_id=signal_id, detail=detail,
 315                                                       closure=foo, func=dummy, data=dummy)
 316         self.assertEqual(count, 1)
 317         obj.emit('my-signal')
 318         self.assertEqual(obj.status, 1)
 319
 320         # Unblocking by the same match criteria allows callback to work again
 321         count = GObject.signal_handlers_unblock_matched(obj,
 322                                                         GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
 323                                                         signal_id=signal_id, detail=detail,
 324                                                         closure=foo, func=dummy, data=dummy)
 325         self.assertEqual(count, 1)
 326         obj.emit('my-signal')
 327         self.assertEqual(obj.status, 2)
 328
 329         # Disconnecting by match criteria completely removes the handler
 330         count = GObject.signal_handlers_disconnect_matched(obj,
 331                                                            GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
 332                                                            signal_id=signal_id, detail=detail,
 333                                                            closure=foo, func=dummy, data=dummy)
 334         self.assertEqual(count, 1)
 335         obj.emit('my-signal')
 336         self.assertEqual(obj.status, 2)
 337
 338     def test_signal_handler_find(self):
 339         def foo(obj):
 340             obj.status += 1
 341
 342         obj = self.Object()
 343         handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)
 344
 345         signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
 346         found_id = GObject.signal_handler_find(obj,
 347                                                GObject.SignalMatchType.ID,
 348                                                signal_id=signal_id, detail=detail,
 349                                                closure=None, func=0, data=0)
 350         self.assertEqual(handler_id, found_id)
 351
 352
 353 class TestClosures(unittest.TestCase):
 354     def setUp(self):
 355         self.count = 0
 356         self.emission_stopped = False
 357         self.emission_error = False
 358         self.handler_pending = False
 359
 360     def _callback_handler_pending(self, e):
 361         signal_id, detail = GObject.signal_parse_name('signal', e, True)
 362         self.handler_pending = GObject.signal_has_handler_pending(e, signal_id, detail,
 363                                                                   may_be_blocked=False)
 364
 365     def _callback(self, e):
 366         self.count += 1
 367
 368     def _callback_stop_emission(self, obj, prop, stop_it):
 369         if stop_it:
 370             obj.stop_emission_by_name('notify::prop')
 371             self.emission_stopped = True
 372         else:
 373             self.count += 1
 374
 375     def _callback_invalid_stop_emission_name(self, obj, prop):
 376         # We expect a GLib warning but there currently is no way to test that
 377         # This can at least make sure we don't crash
 378         old_mask = GLib.log_set_always_fatal(GLib.LogLevelFlags.LEVEL_CRITICAL |
 379                                              GLib.LogLevelFlags.LEVEL_ERROR)
 380         try:
 381             obj.stop_emission_by_name('notasignal::baddetail')
 382         finally:
 383             GLib.log_set_always_fatal(old_mask)
 384             self.emission_error = True
 385
 386     def test_disconnect_by_func(self):
 387         e = E()
 388         e.connect('signal', self._callback)
 389         e.disconnect_by_func(self._callback)
 390         e.emit('signal')
 391         self.assertEqual(self.count, 0)
 392
 393     def test_disconnect(self):
 394         e = E()
 395         handler_id = e.connect('signal', self._callback)
 396         self.assertTrue(e.handler_is_connected(handler_id))
 397         e.disconnect(handler_id)
 398         e.emit('signal')
 399         self.assertEqual(self.count, 0)
 400         self.assertFalse(e.handler_is_connected(handler_id))
 401
 402     def test_stop_emission_by_name(self):
 403         e = E()
 404
 405         # Sandwich a callback that stops emission in between a callback that increments
 406         e.connect('notify::prop', self._callback_stop_emission, False)
 407         e.connect('notify::prop', self._callback_stop_emission, True)
 408         e.connect('notify::prop', self._callback_stop_emission, False)
 409
 410         e.set_property('prop', 1234)
 411         self.assertEqual(e.get_property('prop'), 1234)
 412         self.assertEqual(self.count, 1)
 413         self.assertTrue(self.emission_stopped)
 414
 415     def test_stop_emission_by_name_error(self):
 416         e = E()
 417
 418         e.connect('notify::prop', self._callback_invalid_stop_emission_name)
 419         e.set_property('prop', 1234)
 420         self.assertTrue(self.emission_error)
 421
 422     def test_handler_block(self):
 423         e = E()
 424         e.connect('signal', self._callback)
 425         e.handler_block_by_func(self._callback)
 426         e.emit('signal')
 427         self.assertEqual(self.count, 0)
 428
 429     def test_handler_unblock(self):
 430         e = E()
 431         handler_id = e.connect('signal', self._callback)
 432         e.handler_block(handler_id)
 433         e.handler_unblock_by_func(self._callback)
 434         e.emit('signal')
 435         self.assertEqual(self.count, 1)
 436
 437     def test_handler_block_method(self):
 438         # Filed as #375589
 439         class A:
 440             def __init__(self):
 441                 self.a = 0
 442
 443             def callback(self, o):
 444                 self.a = 1
 445                 o.handler_block_by_func(self.callback)
 446
 447         inst = A()
 448         e = E()
 449         e.connect("signal", inst.callback)
 450         e.emit('signal')
 451         self.assertEqual(inst.a, 1)
 452         gc.collect()
 453
 454     def test_gstring(self):
 455         class C(GObject.GObject):
 456             __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_GSTRING,
 457                                           (GObject.TYPE_GSTRING,))}
 458
 459             def __init__(self, test):
 460                 GObject.GObject.__init__(self)
 461                 self.test = test
 462
 463             def do_my_signal(self, data):
 464                 self.data = data
 465                 self.test.assertEqual(len(data), 3)
 466                 return ''.join([data[2], data[1], data[0]])
 467         c = C(self)
 468         data = c.emit("my_signal", "\01\00\02")
 469         self.assertEqual(data, "\02\00\01")
 470
 471     def test_handler_pending(self):
 472         obj = F()
 473         obj.connect('signal', self._callback_handler_pending)
 474         obj.connect('signal', self._callback)
 475
 476         self.assertEqual(self.count, 0)
 477         self.assertEqual(self.handler_pending, False)
 478
 479         obj.emit('signal')
 480         self.assertEqual(self.count, 1)
 481         self.assertEqual(self.handler_pending, True)
 482
 483     def test_signal_handlers_destroy(self):
 484         obj = F()
 485         obj.connect('signal', self._callback)
 486         obj.connect('signal', self._callback)
 487         obj.connect('signal', self._callback)
 488
 489         obj.emit('signal')
 490         self.assertEqual(self.count, 3)
 491
 492         # count should remain at 3 after all handlers are destroyed
 493         GObject.signal_handlers_destroy(obj)
 494         obj.emit('signal')
 495         self.assertEqual(self.count, 3)
 496
 497
 498 class SigPropClass(GObject.GObject):
 499     __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
 500                                   (GObject.TYPE_INT,))}
 501
 502     __gproperties__ = {
 503         'foo': (str, None, None, '', GObject.PARAM_WRITABLE | GObject.PARAM_CONSTRUCT),
 504         }
 505
 506     signal_emission_failed = False
 507
 508     def do_my_signal(self, arg):
 509         self.arg = arg
 510
 511     def do_set_property(self, pspec, value):
 512         if pspec.name == 'foo':
 513             self._foo = value
 514         else:
 515             raise AttributeError('unknown property %s' % pspec.name)
 516         try:
 517             self.emit("my-signal", 1)
 518         except TypeError:
 519             self.signal_emission_failed = True
 520
 521
 522 class TestSigProp(unittest.TestCase):
 523     def test_emit_in_property_setter(self):
 524         obj = SigPropClass()
 525         self.assertFalse(obj.signal_emission_failed)
 526
 527
 528 class CM(GObject.GObject):
 529     __gsignals__ = dict(
 530         test1=(GObject.SignalFlags.RUN_FIRST, None, ()),
 531         test2=(GObject.SignalFlags.RUN_LAST, None, (str,)),
 532         test3=(GObject.SignalFlags.RUN_LAST, int, (GObject.TYPE_DOUBLE,)),
 533         test4=(GObject.SignalFlags.RUN_FIRST, None,
 534                (bool, _long, GObject.TYPE_FLOAT, GObject.TYPE_DOUBLE, int,
 535                 GObject.TYPE_UINT, GObject.TYPE_ULONG)),
 536         test_float=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_FLOAT, (GObject.TYPE_FLOAT,)),
 537         test_double=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_DOUBLE, (GObject.TYPE_DOUBLE,)),
 538         test_int64=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_INT64, (GObject.TYPE_INT64,)),
 539         test_string=(GObject.SignalFlags.RUN_LAST, str, (str,)),
 540         test_object=(GObject.SignalFlags.RUN_LAST, object, (object,)),
 541         test_paramspec=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, ()),
 542         test_paramspec_in=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, (GObject.ParamSpec, )),
 543         test_gvalue=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.Value,)),
 544         test_gvalue_ret=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.TYPE_GTYPE,)),
 545     )
 546
 547     testprop = GObject.Property(type=int)
 548
 549
 550 class _TestCMarshaller:
 551     def setUp(self):
 552         self.obj = CM()
 553         testhelper.connectcallbacks(self.obj)
 554
 555     def test_test1(self):
 556         self.obj.emit("test1")
 557
 558     def test_test2(self):
 559         self.obj.emit("test2", "string")
 560
 561     def test_test3(self):
 562         rv = self.obj.emit("test3", 42.0)
 563         self.assertEqual(rv, 20)
 564
 565     def test_test4(self):
 566         self.obj.emit("test4", True, _long(10), 3.14, 1.78, 20, _long(30), _long(31))
 567
 568     def test_float(self):
 569         rv = self.obj.emit("test-float", 1.234)
 570         self.assertTrue(rv >= 1.233999 and rv <= 1.2400001, rv)
 571
 572     def test_double(self):
 573         rv = self.obj.emit("test-double", 1.234)
 574         self.assertEqual(rv, 1.234)
 575
 576     def test_int64(self):
 577         rv = self.obj.emit("test-int64", 102030405)
 578         self.assertEqual(rv, 102030405)
 579
 580         rv = self.obj.emit("test-int64", GObject.G_MAXINT64)
 581         self.assertEqual(rv, GObject.G_MAXINT64 - 1)
 582
 583         rv = self.obj.emit("test-int64", GObject.G_MININT64)
 584         self.assertEqual(rv, GObject.G_MININT64)
 585
 586     def test_string(self):
 587         rv = self.obj.emit("test-string", "str")
 588         self.assertEqual(rv, "str")
 589
 590     def test_object(self):
 591         rv = self.obj.emit("test-object", self)
 592         self.assertEqual(rv, self)
 593
 594     def test_paramspec(self):
 595         rv = self.obj.emit("test-paramspec")
 596         self.assertEqual(rv.name, "test-param")
 597         self.assertEqual(rv.nick, "test")
 598
 599     @unittest.skipUnless(hasattr(GObject, 'param_spec_boolean'),
 600                          'too old gobject-introspection')
 601     def test_paramspec_in(self):
 602         rv = GObject.param_spec_boolean('mybool', 'test-bool', 'do something',
 603                                         True, GObject.ParamFlags.READABLE)
 604
 605         rv2 = self.obj.emit("test-paramspec-in", rv)
 606         self.assertEqual(type(rv), type(rv2))
 607         self.assertEqual(rv2.name, "mybool")
 608         self.assertEqual(rv2.nick, "test-bool")
 609
 610     def test_C_paramspec(self):
 611         self.notify_called = False
 612
 613         def cb_notify(obj, prop):
 614             self.notify_called = True
 615             self.assertEqual(obj, self.obj)
 616             self.assertEqual(prop.name, "testprop")
 617
 618         self.obj.connect("notify", cb_notify)
 619         self.obj.set_property("testprop", 42)
 620         self.assertTrue(self.notify_called)
 621
 622     def test_gvalue(self):
 623         # implicit int
 624         rv = self.obj.emit("test-gvalue", 42)
 625         self.assertEqual(rv, 42)
 626
 627         # explicit float
 628         v = GObject.Value(GObject.TYPE_FLOAT, 1.234)
 629         rv = self.obj.emit("test-gvalue", v)
 630         self.assertAlmostEqual(rv, 1.234, 4)
 631
 632         # implicit float
 633         rv = self.obj.emit("test-gvalue", 1.234)
 634         self.assertAlmostEqual(rv, 1.234, 4)
 635
 636         # explicit int64
 637         v = GObject.Value(GObject.TYPE_INT64, GObject.G_MAXINT64)
 638         rv = self.obj.emit("test-gvalue", v)
 639         self.assertEqual(rv, GObject.G_MAXINT64)
 640
 641         # explicit uint64
 642         v = GObject.Value(GObject.TYPE_UINT64, GObject.G_MAXUINT64)
 643         rv = self.obj.emit("test-gvalue", v)
 644         self.assertEqual(rv, GObject.G_MAXUINT64)
 645
 646     @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=705291
 647     def test_gvalue_implicit_int64(self):
 648         # implicit int64
 649         rv = self.obj.emit("test-gvalue", GObject.G_MAXINT64)
 650         self.assertEqual(rv, GObject.G_MAXINT64)
 651
 652         # implicit uint64
 653         rv = self.obj.emit("test-gvalue", GObject.G_MAXUINT64)
 654         self.assertEqual(rv, GObject.G_MAXUINT64)
 655
 656     def test_gvalue_ret(self):
 657         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT),
 658                          GObject.G_MAXINT)
 659         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT),
 660                          GObject.G_MAXUINT)
 661         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT64),
 662                          GObject.G_MAXINT64)
 663         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT64),
 664                          GObject.G_MAXUINT64)
 665         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_STRING),
 666                          "hello")
 667
 668 if 'generic-c-marshaller' in GObject.features:
 669     class TestCMarshaller(_TestCMarshaller, unittest.TestCase):
 670         pass
 671 else:
 672     print()
 673     print('** WARNING: LIBFFI disabled, not testing')
 674     print()
 675
 676 # Test for 374653
 677
 678
 679 class TestPyGValue(unittest.TestCase):
 680     def test_none_null_boxed_conversion(self):
 681         class C(GObject.GObject):
 682             __gsignals__ = dict(my_boxed_signal=(
 683                 GObject.SignalFlags.RUN_LAST,
 684                 GObject.TYPE_STRV, ()))
 685
 686         obj = C()
 687         obj.connect('my-boxed-signal', lambda obj: None)
 688         sys.last_type = None
 689         obj.emit('my-boxed-signal')
 690         assert not sys.last_type
 691
 692
 693 class TestSignalDecorator(unittest.TestCase):
 694     class Decorated(GObject.GObject):
 695         value = 0
 696
 697         @GObject.Signal
 698         def pushed(self):
 699             """this will push"""
 700             self.value += 1
 701
 702         @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
 703         def pulled(self):
 704             self.value -= 1
 705
 706         stomped = GObject.Signal('stomped', arg_types=(int,), doc='this will stomp')
 707         unnamed = GObject.Signal()
 708
 709     class DecoratedOverride(GObject.GObject):
 710         overridden_closure_called = False
 711         notify_called = False
 712         value = GObject.Property(type=int, default=0)
 713
 714         @GObject.SignalOverride
 715         def notify(self, *args, **kargs):
 716             self.overridden_closure_called = True
 717
 718         def on_notify(self, obj, prop):
 719             self.notify_called = True
 720
 721     def setUp(self):
 722         self.unnamedCalled = False
 723
 724     def onUnnamed(self, obj):
 725         self.unnamedCalled = True
 726
 727     def test_get_signal_args(self):
 728         self.assertEqual(self.Decorated.pushed.get_signal_args(),
 729                          (GObject.SignalFlags.RUN_FIRST, None, tuple(), None, None))
 730         self.assertEqual(self.Decorated.pulled.get_signal_args(),
 731                          (GObject.SignalFlags.RUN_LAST, None, tuple(), None, None))
 732         self.assertEqual(self.Decorated.stomped.get_signal_args(),
 733                          (GObject.SignalFlags.RUN_FIRST, None, (int,), None, None))
 734
 735     def test_closures_called(self):
 736         decorated = self.Decorated()
 737         self.assertEqual(decorated.value, 0)
 738         decorated.pushed.emit()
 739         self.assertEqual(decorated.value, 1)
 740         decorated.pulled.emit()
 741         self.assertEqual(decorated.value, 0)
 742
 743     def test_signal_copy(self):
 744         blah = self.Decorated.stomped.copy('blah')
 745         self.assertEqual(str(blah), blah)
 746         self.assertEqual(blah.func, self.Decorated.stomped.func)
 747         self.assertEqual(blah.flags, self.Decorated.stomped.flags)
 748         self.assertEqual(blah.return_type, self.Decorated.stomped.return_type)
 749         self.assertEqual(blah.arg_types, self.Decorated.stomped.arg_types)
 750         self.assertEqual(blah.__doc__, self.Decorated.stomped.__doc__)
 751
 752     def test_doc_string(self):
 753         # Test the two techniques for setting doc strings on the signals
 754         # class variables, through the "doc" keyword or as the getter doc string.
 755         self.assertEqual(self.Decorated.stomped.__doc__, 'this will stomp')
 756         self.assertEqual(self.Decorated.pushed.__doc__, 'this will push')
 757
 758     def test_unnamed_signal_gets_named(self):
 759         self.assertEqual(str(self.Decorated.unnamed), 'unnamed')
 760
 761     def test_unnamed_signal_gets_called(self):
 762         obj = self.Decorated()
 763         obj.connect('unnamed', self.onUnnamed)
 764         self.assertEqual(self.unnamedCalled, False)
 765         obj.emit('unnamed')
 766         self.assertEqual(self.unnamedCalled, True)
 767
 768     def test_overridden_signal(self):
 769         # Test that the pushed signal is called in with super and the override
 770         # which should both increment the "value" to 3
 771         obj = self.DecoratedOverride()
 772         obj.connect("notify", obj.on_notify)
 773         self.assertEqual(obj.value, 0)
 774         obj.value = 1
 775         self.assertEqual(obj.value, 1)
 776         self.assertTrue(obj.overridden_closure_called)
 777         self.assertTrue(obj.notify_called)
 778
 779
 780 class TestSignalConnectors(unittest.TestCase):
 781     class CustomButton(GObject.GObject):
 782         on_notify_called = False
 783         value = GObject.Property(type=int)
 784
 785         @GObject.Signal(arg_types=(int,))
 786         def clicked(self, value):
 787             self.value = value
 788
 789     def setUp(self):
 790         self.obj = None
 791         self.value = None
 792
 793     def on_clicked(self, obj, value):
 794         self.obj = obj
 795         self.value = value
 796
 797     def test_signal_notify(self):
 798         def on_notify(obj, param):
 799             obj.on_notify_called = True
 800
 801         obj = self.CustomButton()
 802         obj.connect('notify', on_notify)
 803         self.assertFalse(obj.on_notify_called)
 804         obj.notify('value')
 805         self.assertTrue(obj.on_notify_called)
 806
 807     def test_signal_emit(self):
 808         # standard callback connection with different forms of emit.
 809         obj = self.CustomButton()
 810         obj.connect('clicked', self.on_clicked)
 811
 812         # vanilla
 813         obj.emit('clicked', 1)
 814         self.assertEqual(obj.value, 1)
 815         self.assertEqual(obj, self.obj)
 816         self.assertEqual(self.value, 1)
 817
 818         # using class signal as param
 819         self.obj = None
 820         self.value = None
 821         obj.emit(self.CustomButton.clicked, 1)
 822         self.assertEqual(obj, self.obj)
 823         self.assertEqual(self.value, 1)
 824
 825         # using bound signal as param
 826         self.obj = None
 827         self.value = None
 828         obj.emit(obj.clicked, 1)
 829         self.assertEqual(obj, self.obj)
 830         self.assertEqual(self.value, 1)
 831
 832         # using bound signal with emit
 833         self.obj = None
 834         self.value = None
 835         obj.clicked.emit(1)
 836         self.assertEqual(obj, self.obj)
 837         self.assertEqual(self.value, 1)
 838
 839     def test_signal_class_connect(self):
 840         obj = self.CustomButton()
 841         obj.connect(self.CustomButton.clicked, self.on_clicked)
 842         obj.emit('clicked', 2)
 843         self.assertEqual(obj, self.obj)
 844         self.assertEqual(self.value, 2)
 845
 846     def test_signal_bound_connect(self):
 847         obj = self.CustomButton()
 848         obj.clicked.connect(self.on_clicked)
 849         obj.emit('clicked', 3)
 850         self.assertEqual(obj, self.obj)
 851         self.assertEqual(self.value, 3)
 852
 853
 854 class _ConnectDataTestBase(object):
 855     # Notes:
 856     #  - self.Object is overridden in sub-classes.
 857     #  - Numeric suffixes indicate the number of user data args passed in.
 858     Object = None
 859
 860     def run_connect_test(self, emit_args, user_data, flags=0):
 861         obj = self.Object()
 862         callback_args = []
 863
 864         def callback(*args):
 865             callback_args.append(args)
 866             return 0
 867
 868         obj.connect_data('sig-with-int64-prop', callback, connect_flags=flags, *user_data)
 869         obj.emit('sig-with-int64-prop', *emit_args)
 870         self.assertEqual(len(callback_args), 1)
 871         return callback_args[0]
 872
 873     def test_0(self):
 874         obj, value = self.run_connect_test([GObject.G_MAXINT64], user_data=[])
 875         self.assertIsInstance(obj, self.Object)
 876         self.assertEqual(value, GObject.G_MAXINT64)
 877
 878     def test_1(self):
 879         obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
 880                                                  user_data=['mydata'])
 881         self.assertIsInstance(obj, self.Object)
 882         self.assertEqual(value, GObject.G_MAXINT64)
 883         self.assertEqual(data, 'mydata')
 884
 885     def test_after_0(self):
 886         obj, value = self.run_connect_test([GObject.G_MAXINT64],
 887                                            user_data=[],
 888                                            flags=GObject.ConnectFlags.AFTER)
 889         self.assertIsInstance(obj, self.Object)
 890         self.assertEqual(value, GObject.G_MAXINT64)
 891
 892     def test_after_1(self):
 893         obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
 894                                                  user_data=['mydata'],
 895                                                  flags=GObject.ConnectFlags.AFTER)
 896         self.assertIsInstance(obj, self.Object)
 897         self.assertEqual(value, GObject.G_MAXINT64)
 898         self.assertEqual(data, 'mydata')
 899
 900     def test_swaped_0(self):
 901         # Swapped only works with a single user data argument.
 902         with self.assertRaises(ValueError):
 903             self.run_connect_test([GObject.G_MAXINT64],
 904                                   user_data=[],
 905                                   flags=GObject.ConnectFlags.SWAPPED)
 906
 907     def test_swaped_1(self):
 908         # Notice obj and data are reversed in the return.
 909         data, value, obj = self.run_connect_test([GObject.G_MAXINT64],
 910                                                  user_data=['mydata'],
 911                                                  flags=GObject.ConnectFlags.SWAPPED)
 912         self.assertIsInstance(obj, self.Object)
 913         self.assertEqual(value, GObject.G_MAXINT64)
 914         self.assertEqual(data, 'mydata')
 915
 916     def test_swaped_2(self):
 917         # Swapped only works with a single user data argument.
 918         with self.assertRaises(ValueError):
 919             self.run_connect_test([GObject.G_MAXINT64],
 920                                   user_data=[1, 2],
 921                                   flags=GObject.ConnectFlags.SWAPPED)
 922
 923     def test_after_and_swapped_0(self):
 924         # Swapped only works with a single user data argument.
 925         with self.assertRaises(ValueError):
 926             self.run_connect_test([GObject.G_MAXINT64],
 927                                   user_data=[],
 928                                   flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
 929
 930     def test_after_and_swapped_1(self):
 931         # Notice obj and data are reversed in the return.
 932         data, value, obj = self.run_connect_test([GObject.G_MAXINT64],
 933                                                  user_data=['mydata'],
 934                                                  flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
 935         self.assertIsInstance(obj, self.Object)
 936         self.assertEqual(value, GObject.G_MAXINT64)
 937         self.assertEqual(data, 'mydata')
 938
 939     def test_after_and_swapped_2(self):
 940         # Swapped only works with a single user data argument.
 941         with self.assertRaises(ValueError):
 942             self.run_connect_test([GObject.G_MAXINT64],
 943                                   user_data=[],
 944                                   flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
 945
 946
 947 class TestConnectDataNonIntrospected(unittest.TestCase, _ConnectDataTestBase):
 948     # This tests connect_data with non-introspected signals
 949     # (created in Python in this case).
 950     class Object(GObject.Object):
 951         test = GObject.Signal()
 952         sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
 953                                              arg_types=[GObject.TYPE_INT64],
 954                                              flags=GObject.SignalFlags.RUN_LAST)
 955
 956
 957 @unittest.skipUnless(has_cairo, 'built without cairo support')
 958 class TestConnectDataIntrospected(unittest.TestCase, _ConnectDataTestBase):
 959     # This tests connect_data with introspected signals brought in from Regress.
 960     Object = Regress.TestObj
 961
 962
 963 class TestInstallSignals(unittest.TestCase):
 964     # These tests only test how signalhelper.install_signals works
 965     # with the __gsignals__ dict and therefore does not need to use
 966     # GObject as a base class because that would automatically call
 967     # install_signals within the meta-class.
 968     class Base(object):
 969         __gsignals__ = {'test': (0, None, tuple())}
 970
 971     class Sub1(Base):
 972         pass
 973
 974     class Sub2(Base):
 975         @GObject.Signal
 976         def sub2test(self):
 977             pass
 978
 979     def setUp(self):
 980         self.assertEqual(len(self.Base.__gsignals__), 1)
 981         signalhelper.install_signals(self.Base)
 982         self.assertEqual(len(self.Base.__gsignals__), 1)
 983
 984     def test_subclass_gets_empty_gsignals_dict(self):
 985         # Installing signals will add the __gsignals__ dict to a class
 986         # if it doesn't already exists.
 987         self.assertFalse('__gsignals__' in self.Sub1.__dict__)
 988         signalhelper.install_signals(self.Sub1)
 989         self.assertTrue('__gsignals__' in self.Sub1.__dict__)
 990         # Sub1 should only contain an empty signals dict, this tests:
 991         # https://bugzilla.gnome.org/show_bug.cgi?id=686496
 992         self.assertEqual(self.Sub1.__dict__['__gsignals__'], {})
 993
 994     def test_subclass_with_decorator_gets_gsignals_dict(self):
 995         self.assertFalse('__gsignals__' in self.Sub2.__dict__)
 996         signalhelper.install_signals(self.Sub2)
 997         self.assertTrue('__gsignals__' in self.Sub2.__dict__)
 998         self.assertEqual(len(self.Base.__gsignals__), 1)
 999         self.assertEqual(len(self.Sub2.__gsignals__), 1)
1000         self.assertTrue('sub2test' in self.Sub2.__gsignals__)
1001
1002         # Make sure the vfunc was added
1003         self.assertTrue(hasattr(self.Sub2, 'do_sub2test'))
1004
1005
1006 # For this test to work with both python2 and 3 we need to dynamically
1007 # exec the given code due to the new syntax causing an error in python 2.
1008 annotated_class_code = """
1009 class AnnotatedSignalClass(GObject.GObject):
1010     @GObject.Signal
1011     def sig1(self, a:int, b:float):
1012         pass
1013
1014     @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
1015     def sig2_with_return(self, a:int, b:float) -> str:
1016         return "test"
1017 """
1018
1019
1020 @unittest.skipUnless(sys.version_info >= (3, 0),
1021                      'Argument annotations require Python 3')
1022 class TestPython3Signals(unittest.TestCase):
1023     AnnotatedClass = None
1024
1025     def setUp(self):
1026         exec(annotated_class_code, globals(), globals())
1027         self.assertTrue('AnnotatedSignalClass' in globals())
1028         self.AnnotatedClass = globals()['AnnotatedSignalClass']
1029
1030     def test_annotations(self):
1031         self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig1.func),
1032                          (None, (int, float)))
1033         self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig2_with_return.func),
1034                          (str, (int, float)))
1035
1036         self.assertEqual(self.AnnotatedClass.sig2_with_return.get_signal_args(),
1037                          (GObject.SignalFlags.RUN_LAST, str, (int, float), None, None))
1038         self.assertEqual(self.AnnotatedClass.sig2_with_return.arg_types,
1039                          (int, float))
1040         self.assertEqual(self.AnnotatedClass.sig2_with_return.return_type,
1041                          str)
1042
1043     def test_emit_return(self):
1044         obj = self.AnnotatedClass()
1045         self.assertEqual(obj.sig2_with_return.emit(1, 2.0),
1046                          'test')
1047
1048
1049 class TestSignalModuleLevelFunctions(unittest.TestCase):
1050     def test_signal_list_ids_with_invalid_type(self):
1051         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
1052             GObject.signal_list_ids(GObject.TYPE_INVALID)
1053
1054     def test_signal_list_ids(self):
1055         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
1056             GObject.signal_list_ids(GObject.TYPE_INT)
1057
1058         ids = GObject.signal_list_ids(C)
1059         self.assertEqual(len(ids), 1)
1060         # Note canonicalized names
1061         self.assertEqual(GObject.signal_name(ids[0]), 'my-signal')
1062         # There is no signal 0 in gobject
1063         self.assertEqual(GObject.signal_name(0), None)
1064
1065     def test_signal_lookup_with_invalid_type(self):
1066         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
1067             GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INVALID)
1068
1069     def test_signal_lookup(self):
1070         ids = GObject.signal_list_ids(C)
1071         self.assertEqual(ids[0], GObject.signal_lookup('my_signal', C))
1072         self.assertEqual(ids[0], GObject.signal_lookup('my-signal', C))
1073
1074         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
1075             GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INT)
1076
1077         # Invalid signal names return 0 instead of raising
1078         self.assertEqual(GObject.signal_lookup('NOT_A_SIGNAL_NAME', C),
1079                          0)
1080
1081     def test_signal_query(self):
1082         my_signal_id, = GObject.signal_list_ids(C)
1083
1084         # Form is: (id, name, gtype, arg_count, return_type, (arg_type1, ...))
1085         my_signal_expected_query_result = [my_signal_id, 'my-signal', C.__gtype__,
1086                                            1, GObject.TYPE_NONE, (GObject.TYPE_INT,)]
1087         # signal_query(name, type)
1088         self.assertEqual(list(GObject.signal_query('my-signal', C)), my_signal_expected_query_result)
1089         # signal_query(signal_id)
1090         self.assertEqual(list(GObject.signal_query(my_signal_id)), my_signal_expected_query_result)
1091         # invalid query returns None instead of raising
1092         self.assertEqual(GObject.signal_query(0), None)
1093         self.assertEqual(GObject.signal_query('NOT_A_SIGNAL', C),
1094                          None)
1095
1096
1097 @unittest.skipUnless(has_cairo, 'built without cairo support')
1098 class TestIntrospectedSignals(unittest.TestCase):
1099     def test_object_param_signal(self):
1100         obj = Regress.TestObj()
1101
1102         def callback(obj, obj_param):
1103             self.assertEqual(obj_param.props.int, 3)
1104             self.assertGreater(obj_param.__grefcount__, 1)
1105             obj.called = True
1106
1107         obj.called = False
1108         obj.connect('sig-with-obj', callback)
1109         obj.emit_sig_with_obj()
1110         self.assertTrue(obj.called)
1111
1112     def test_connect_after(self):
1113         obj = Regress.TestObj()
1114
1115         def callback(obj, obj_param):
1116             obj.called = True
1117
1118         obj.called = False
1119         obj.connect_after('sig-with-obj', callback)
1120         obj.emit_sig_with_obj()
1121         self.assertTrue(obj.called)
1122
1123     def test_int64_param_from_py(self):
1124         obj = Regress.TestObj()
1125
1126         def callback(obj, i):
1127             obj.callback_i = i
1128             return i
1129
1130         obj.callback_i = None
1131         obj.connect('sig-with-int64-prop', callback)
1132         rv = obj.emit('sig-with-int64-prop', GObject.G_MAXINT64)
1133         self.assertEqual(rv, GObject.G_MAXINT64)
1134         self.assertEqual(obj.callback_i, GObject.G_MAXINT64)
1135
1136     def test_uint64_param_from_py(self):
1137         obj = Regress.TestObj()
1138
1139         def callback(obj, i):
1140             obj.callback_i = i
1141             return i
1142
1143         obj.callback_i = None
1144         obj.connect('sig-with-uint64-prop', callback)
1145         rv = obj.emit('sig-with-uint64-prop', GObject.G_MAXUINT64)
1146         self.assertEqual(rv, GObject.G_MAXUINT64)
1147         self.assertEqual(obj.callback_i, GObject.G_MAXUINT64)
1148
1149     def test_int64_param_from_c(self):
1150         obj = Regress.TestObj()
1151
1152         def callback(obj, i):
1153             obj.callback_i = i
1154             return i
1155
1156         obj.callback_i = None
1157
1158         obj.connect('sig-with-int64-prop', callback)
1159         obj.emit_sig_with_int64()
1160         self.assertEqual(obj.callback_i, GObject.G_MAXINT64)
1161
1162     def test_uint64_param_from_c(self):
1163         obj = Regress.TestObj()
1164
1165         def callback(obj, i):
1166             obj.callback_i = i
1167             return i
1168
1169         obj.callback_i = None
1170
1171         obj.connect('sig-with-uint64-prop', callback)
1172         obj.emit_sig_with_uint64()
1173         self.assertEqual(obj.callback_i, GObject.G_MAXUINT64)
1174
1175     def test_intarray_ret(self):
1176         obj = Regress.TestObj()
1177
1178         def callback(obj, i):
1179             obj.callback_i = i
1180             return [i, i + 1]
1181
1182         obj.callback_i = None
1183
1184         try:
1185             obj.connect('sig-with-intarray-ret', callback)
1186         except TypeError as e:
1187             # compat with g-i 1.34.x
1188             if 'unknown signal' in str(e):
1189                 return
1190             raise
1191
1192         rv = obj.emit('sig-with-intarray-ret', 42)
1193         self.assertEqual(obj.callback_i, 42)
1194         self.assertEqual(type(rv), GLib.Array)
1195         self.assertEqual(rv.len, 2)
1196
1197     @unittest.skip('https://bugzilla.gnome.org/show_bug.cgi?id=669496')
1198     def test_array_parm(self):
1199         obj = Regress.TestObj()
1200
1201         def callback(obj, arr):
1202             obj.callback_arr = arr
1203
1204         obj.connect('sig-with-array-prop', callback)
1205         obj.callback_arr = None
1206         self.assertEqual(obj.emit('sig-with-array-prop', [1, 2, GObject.G_MAXUINT]), None)
1207         self.assertEqual(obj.callback_arr, [1, 2, GObject.G_MAXUINT])
1208
1209
1210 class _ConnectObjectTestBase(object):
1211     # Notes:
1212     #  - self.Object is overridden in sub-classes.
1213     #  - Numeric suffixes indicate the number of user data args passed in.
1214     Object = None
1215     SwapObject = None
1216
1217     def run_connect_test(self, emit_args, user_data, flags=0):
1218         obj = self.Object()
1219         callback_args = []
1220         swap_obj = self.SwapObject()
1221
1222         def callback(*args):
1223             callback_args.append(args)
1224             return 0
1225
1226         if flags & GObject.ConnectFlags.AFTER:
1227             connect_func = obj.connect_object_after
1228         else:
1229             connect_func = obj.connect_object
1230
1231         connect_func('sig-with-int64-prop', callback, swap_obj, *user_data)
1232         obj.emit('sig-with-int64-prop', *emit_args)
1233         self.assertEqual(len(callback_args), 1)
1234         return callback_args[0]
1235
1236     def test_0(self):
1237         obj, value = self.run_connect_test([GObject.G_MAXINT64], user_data=[])
1238         self.assertIsInstance(obj, self.SwapObject)
1239         self.assertEqual(value, GObject.G_MAXINT64)
1240
1241     def test_1(self):
1242         obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
1243                                                  user_data=['mydata'])
1244         self.assertIsInstance(obj, self.SwapObject)
1245         self.assertEqual(value, GObject.G_MAXINT64)
1246         self.assertEqual(data, 'mydata')
1247
1248     def test_2(self):
1249         obj, value, data1, data2 = self.run_connect_test([GObject.G_MAXINT64],
1250                                                          user_data=['mydata1', 'mydata2'])
1251         self.assertIsInstance(obj, self.SwapObject)
1252         self.assertEqual(value, GObject.G_MAXINT64)
1253         self.assertEqual(data1, 'mydata1')
1254         self.assertEqual(data2, 'mydata2')
1255
1256     def test_after_0(self):
1257         obj, value = self.run_connect_test([GObject.G_MAXINT64],
1258                                            user_data=[],
1259                                            flags=GObject.ConnectFlags.AFTER)
1260         self.assertIsInstance(obj, self.SwapObject)
1261         self.assertEqual(value, GObject.G_MAXINT64)
1262
1263     def test_after_1(self):
1264         obj, value, data = self.run_connect_test([GObject.G_MAXINT64],
1265                                                  user_data=['mydata'],
1266                                                  flags=GObject.ConnectFlags.AFTER)
1267         self.assertIsInstance(obj, self.SwapObject)
1268         self.assertEqual(value, GObject.G_MAXINT64)
1269         self.assertEqual(data, 'mydata')
1270
1271     def test_after_2(self):
1272         obj, value, data1, data2 = self.run_connect_test([GObject.G_MAXINT64],
1273                                                          user_data=['mydata1', 'mydata2'],
1274                                                          flags=GObject.ConnectFlags.AFTER)
1275         self.assertIsInstance(obj, self.SwapObject)
1276         self.assertEqual(value, GObject.G_MAXINT64)
1277         self.assertEqual(data1, 'mydata1')
1278         self.assertEqual(data2, 'mydata2')
1279
1280
1281 class TestConnectGObjectNonIntrospected(unittest.TestCase, _ConnectObjectTestBase):
1282     # This tests connect_object with non-introspected signals
1283     # (created in Python in this case).
1284     class Object(GObject.Object):
1285         test = GObject.Signal()
1286         sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
1287                                              arg_types=[GObject.TYPE_INT64],
1288                                              flags=GObject.SignalFlags.RUN_LAST)
1289
1290     # Object passed for swapping is GObject based.
1291     class SwapObject(GObject.Object):
1292         pass
1293
1294
1295 @unittest.skipUnless(has_cairo, 'built without cairo support')
1296 class TestConnectGObjectIntrospected(unittest.TestCase, _ConnectObjectTestBase):
1297     # This tests connect_object with introspected signals brought in from Regress.
1298     Object = Regress.TestObj
1299
1300     # Object passed for swapping is GObject based.
1301     class SwapObject(GObject.Object):
1302         pass
1303
1304
1305 class TestConnectPyObjectNonIntrospected(unittest.TestCase, _ConnectObjectTestBase):
1306     # This tests connect_object with non-introspected signals
1307     # (created in Python in this case).
1308     class Object(GObject.Object):
1309         test = GObject.Signal()
1310         sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
1311                                              arg_types=[GObject.TYPE_INT64],
1312                                              flags=GObject.SignalFlags.RUN_LAST)
1313
1314     # Object passed for swapping is pure Python
1315     SwapObject = object
1316
1317
1318 @unittest.skipUnless(has_cairo, 'built without cairo support')
1319 class TestConnectPyObjectIntrospected(unittest.TestCase, _ConnectObjectTestBase):
1320     # This tests connect_object with introspected signals brought in from Regress.
1321     Object = Regress.TestObj
1322
1323     # Object passed for swapping is pure Python
1324     SwapObject = object
1325
1326
1327 class _RefCountTestBase(object):
1328     # NOTE: ref counts are always one more than expected because the getrefcount()
1329     # function adds a ref for the input argument.
1330
1331     # Sub-classes set this
1332     Object = None
1333
1334     class PyData(object):
1335         pass
1336
1337     def test_callback_ref_count_del(self):
1338         def callback(obj, value):
1339             return value // 2
1340
1341         callback_ref = weakref.ref(callback)
1342         self.assertEqual(sys.getrefcount(callback), 2)
1343
1344         obj = self.Object()
1345         obj.connect('sig-with-int64-prop', callback)
1346         self.assertEqual(sys.getrefcount(callback), 3)
1347
1348         del callback
1349         self.assertEqual(sys.getrefcount(callback_ref()), 2)
1350
1351         res = obj.emit('sig-with-int64-prop', 42)
1352         self.assertEqual(res, 21)
1353         self.assertEqual(sys.getrefcount(callback_ref), 2)
1354
1355         del obj
1356         self.assertIsNone(callback_ref())
1357
1358     def test_callback_ref_count_disconnect(self):
1359         def callback(obj, value):
1360             return value // 2
1361
1362         callback_ref = weakref.ref(callback)
1363         self.assertEqual(sys.getrefcount(callback), 2)
1364
1365         obj = self.Object()
1366         handler_id = obj.connect('sig-with-int64-prop', callback)
1367         self.assertEqual(sys.getrefcount(callback), 3)
1368
1369         del callback
1370         self.assertEqual(sys.getrefcount(callback_ref()), 2)
1371
1372         res = obj.emit('sig-with-int64-prop', 42)
1373         self.assertEqual(res, 21)
1374         self.assertEqual(sys.getrefcount(callback_ref), 2)
1375
1376         obj.disconnect(handler_id)
1377         self.assertIsNone(callback_ref())
1378
1379     def test_callback_ref_count_disconnect_by_func(self):
1380         def callback(obj, value):
1381             return value // 2
1382
1383         callback_ref = weakref.ref(callback)
1384         self.assertEqual(sys.getrefcount(callback), 2)
1385
1386         obj = self.Object()
1387         obj.connect('sig-with-int64-prop', callback)
1388         self.assertEqual(sys.getrefcount(callback), 3)
1389
1390         del callback
1391         self.assertEqual(sys.getrefcount(callback_ref()), 2)
1392
1393         res = obj.emit('sig-with-int64-prop', 42)
1394         self.assertEqual(res, 21)
1395         self.assertEqual(sys.getrefcount(callback_ref), 2)
1396
1397         obj.disconnect_by_func(callback_ref())
1398         self.assertIsNone(callback_ref())
1399
1400     def test_user_data_ref_count(self):
1401         def callback(obj, value, data):
1402             return value // 2
1403
1404         data = self.PyData()
1405         data_ref = weakref.ref(data)
1406         self.assertEqual(sys.getrefcount(data), 2)
1407
1408         obj = self.Object()
1409         obj.connect('sig-with-int64-prop', callback, data)
1410         self.assertEqual(sys.getrefcount(data), 3)
1411
1412         del data
1413         self.assertEqual(sys.getrefcount(data_ref()), 2)
1414
1415         res = obj.emit('sig-with-int64-prop', 42)
1416         self.assertEqual(res, 21)
1417         self.assertEqual(sys.getrefcount(data_ref()), 2)
1418
1419         del obj
1420         self.assertIsNone(data_ref())
1421
1422     @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=688064
1423     def test_object_ref_count(self):
1424         # connect_object() should only weakly reference the object passed in
1425         # and auto-disconnect the signal when the object is destroyed.
1426         def callback(data, value):
1427             return value // 2
1428
1429         data = GObject.Object()
1430         data_ref = weakref.ref(data)
1431         self.assertEqual(sys.getrefcount(data), 2)
1432
1433         obj = self.Object()
1434         handler_id = obj.connect_object('sig-with-int64-prop', callback, data)
1435         self.assertEqual(sys.getrefcount(data), 2)
1436
1437         res = obj.emit('sig-with-int64-prop', 42)
1438         self.assertEqual(res, 21)
1439         self.assertEqual(sys.getrefcount(data), 2)
1440
1441         del data
1442
1443         self.assertIsNone(data_ref())
1444         self.assertFalse(obj.handler_is_connected(handler_id))
1445
1446
1447 class TestRefCountsNonIntrospected(unittest.TestCase, _RefCountTestBase):
1448     class Object(GObject.Object):
1449         sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
1450                                              arg_types=[GObject.TYPE_INT64],
1451                                              flags=GObject.SignalFlags.RUN_LAST)
1452
1453
1454 @unittest.skipUnless(has_cairo, 'built without cairo support')
1455 class TestRefCountsIntrospected(unittest.TestCase, _RefCountTestBase):
1456     Object = Regress.TestObj
1457
1458
1459 if __name__ == '__main__':
1460     unittest.main()