tests/test_signal.py

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