Imported Upstream version 3.7.91.1
[platform/upstream/pygobject2.git] / tests / test_signal.py
1 # -*- Mode: Python -*-
2
3 import gc
4 import unittest
5 import sys
6
7 from gi.repository import GObject, GLib
8 from gi._gobject import signalhelper
9 import testhelper
10 from compathelper import _long
11
12
13 class C(GObject.GObject):
14     __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
15                                   (GObject.TYPE_INT,))}
16
17     def do_my_signal(self, arg):
18         self.arg = arg
19
20
21 class D(C):
22     def do_my_signal(self, arg2):
23         self.arg2 = arg2
24         C.do_my_signal(self, arg2)
25
26
27 class TestSignalCreation(unittest.TestCase):
28     # Bug 540376.
29     def test_illegals(self):
30         self.assertRaises(TypeError, lambda: GObject.signal_new('test',
31                                                                 None,
32                                                                 0,
33                                                                 None,
34                                                                 (GObject.TYPE_LONG,)))
35
36
37 class TestChaining(unittest.TestCase):
38     def setUp(self):
39         self.inst = C()
40         self.inst.connect("my_signal", self.my_signal_handler_cb, 1, 2, 3)
41
42     def my_signal_handler_cb(self, *args):
43         assert len(args) == 5
44         assert isinstance(args[0], C)
45         assert args[0] == self.inst
46
47         assert isinstance(args[1], int)
48         assert args[1] == 42
49
50         assert args[2:] == (1, 2, 3)
51
52     def test_chaining(self):
53         self.inst.emit("my_signal", 42)
54         assert self.inst.arg == 42
55
56     def test_chaining2(self):
57         inst2 = D()
58         inst2.emit("my_signal", 44)
59         assert inst2.arg == 44
60         assert inst2.arg2 == 44
61
62 # This is for bug 153718
63
64
65 class TestGSignalsError(unittest.TestCase):
66     def test_invalid_type(self, *args):
67         def foo():
68             class Foo(GObject.GObject):
69                 __gsignals__ = None
70         self.assertRaises(TypeError, foo)
71         gc.collect()
72
73     def test_invalid_name(self, *args):
74         def foo():
75             class Foo(GObject.GObject):
76                 __gsignals__ = {'not-exists': 'override'}
77         # do not stumble over the warning thrown by GLib
78         old_mask = GLib.log_set_always_fatal(GLib.LogLevelFlags.LEVEL_CRITICAL |
79                                              GLib.LogLevelFlags.LEVEL_ERROR)
80         try:
81             self.assertRaises(TypeError, foo)
82         finally:
83             GLib.log_set_always_fatal(old_mask)
84         gc.collect()
85
86
87 class TestGPropertyError(unittest.TestCase):
88     def test_invalid_type(self, *args):
89         def foo():
90             class Foo(GObject.GObject):
91                 __gproperties__ = None
92         self.assertRaises(TypeError, foo)
93         gc.collect()
94
95     def test_invalid_name(self, *args):
96         def foo():
97             class Foo(GObject.GObject):
98                 __gproperties__ = {None: None}
99
100         self.assertRaises(TypeError, foo)
101         gc.collect()
102
103
104 class TestList(unittest.TestCase):
105     def test_list_names(self):
106         self.assertEqual(GObject.signal_list_names(C), ('my-signal',))
107
108
109 def my_accumulator(ihint, return_accu, handler_return, user_data):
110     """An accumulator that stops emission when the sum of handler
111     returned values reaches 3"""
112     assert user_data == "accum data"
113     if return_accu >= 3:
114         return False, return_accu
115     return True, return_accu + handler_return
116
117
118 class Foo(GObject.GObject):
119     __gsignals__ = {
120         'my-acc-signal': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_INT,
121                           (), my_accumulator, "accum data"),
122         'my-other-acc-signal': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_BOOLEAN,
123                                 (), GObject.signal_accumulator_true_handled),
124         'my-acc-first-wins': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_BOOLEAN,
125                               (), GObject.signal_accumulator_first_wins)
126         }
127
128
129 class TestAccumulator(unittest.TestCase):
130
131     def test_accumulator(self):
132         inst = Foo()
133         inst.connect("my-acc-signal", lambda obj: 1)
134         inst.connect("my-acc-signal", lambda obj: 2)
135         ## the value returned in the following handler will not be
136         ## considered, because at this point the accumulator already
137         ## reached its limit.
138         inst.connect("my-acc-signal", lambda obj: 3)
139         retval = inst.emit("my-acc-signal")
140         self.assertEqual(retval, 3)
141
142     def test_accumulator_true_handled(self):
143         inst = Foo()
144         inst.connect("my-other-acc-signal", self._true_handler1)
145         inst.connect("my-other-acc-signal", self._true_handler2)
146         ## the following handler will not be called because handler2
147         ## returns True, so it should stop the emission.
148         inst.connect("my-other-acc-signal", self._true_handler3)
149         self.__true_val = None
150         inst.emit("my-other-acc-signal")
151         self.assertEqual(self.__true_val, 2)
152
153     def test_accumulator_first_wins(self):
154         # First signal hit will always win
155         inst = Foo()
156         inst.connect("my-acc-first-wins", self._true_handler3)
157         inst.connect("my-acc-first-wins", self._true_handler1)
158         inst.connect("my-acc-first-wins", self._true_handler2)
159         self.__true_val = None
160         inst.emit("my-acc-first-wins")
161         self.assertEqual(self.__true_val, 3)
162
163     def _true_handler1(self, obj):
164         self.__true_val = 1
165         return False
166
167     def _true_handler2(self, obj):
168         self.__true_val = 2
169         return True
170
171     def _true_handler3(self, obj):
172         self.__true_val = 3
173         return False
174
175
176 class E(GObject.GObject):
177     __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
178                                ())}
179
180     # Property used to test detailed signal
181     prop = GObject.Property(type=int, default=0)
182
183     def __init__(self):
184         GObject.GObject.__init__(self)
185         self.status = 0
186
187     def do_signal(self):
188         assert self.status == 0
189         self.status = 1
190
191
192 class F(GObject.GObject):
193     __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
194                                ())}
195
196     def __init__(self):
197         GObject.GObject.__init__(self)
198         self.status = 0
199
200     def do_signal(self):
201         self.status += 1
202
203
204 class TestEmissionHook(unittest.TestCase):
205     def test_add(self):
206         self.hook = True
207         e = E()
208         e.connect('signal', self._callback)
209         GObject.add_emission_hook(E, "signal", self._emission_hook)
210         e.emit('signal')
211         self.assertEqual(e.status, 3)
212
213     def test_remove(self):
214         self.hook = False
215         e = E()
216         e.connect('signal', self._callback)
217         hook_id = GObject.add_emission_hook(E, "signal", self._emission_hook)
218         GObject.remove_emission_hook(E, "signal", hook_id)
219         e.emit('signal')
220         self.assertEqual(e.status, 3)
221
222     def _emission_hook(self, e):
223         self.assertEqual(e.status, 1)
224         e.status = 2
225
226     def _callback(self, e):
227         if self.hook:
228             self.assertEqual(e.status, 2)
229         else:
230             self.assertEqual(e.status, 1)
231         e.status = 3
232
233     def test_callback_return_false(self):
234         self.hook = False
235         obj = F()
236
237         def _emission_hook(obj):
238             obj.status += 1
239             return False
240         GObject.add_emission_hook(obj, "signal", _emission_hook)
241         obj.emit('signal')
242         obj.emit('signal')
243         self.assertEqual(obj.status, 3)
244
245     def test_callback_return_true(self):
246         self.hook = False
247         obj = F()
248
249         def _emission_hook(obj):
250             obj.status += 1
251             return True
252         hook_id = GObject.add_emission_hook(obj, "signal", _emission_hook)
253         obj.emit('signal')
254         obj.emit('signal')
255         GObject.remove_emission_hook(obj, "signal", hook_id)
256         self.assertEqual(obj.status, 4)
257
258     def test_callback_return_true_but_remove(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         GObject.remove_emission_hook(obj, "signal", hook_id)
268         obj.emit('signal')
269         self.assertEqual(obj.status, 3)
270
271
272 class TestMatching(unittest.TestCase):
273     class Object(GObject.Object):
274         status = 0
275         prop = GObject.Property(type=int, default=0)
276
277         @GObject.Signal()
278         def my_signal(self):
279             pass
280
281     @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=692918
282     def test_signal_handler_block_matching(self):
283         def dummy(*args):
284             "Hack to work around: "
285
286         def foo(obj):
287             obj.status += 1
288
289         obj = self.Object()
290         handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)
291         handler_id
292
293         self.assertEqual(obj.status, 0)
294         obj.emit('my-signal')
295         self.assertEqual(obj.status, 1)
296
297         # Blocking by match criteria disables the foo callback
298         signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
299         count = GObject.signal_handlers_block_matched(obj,
300                                                       GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
301                                                       signal_id=signal_id, detail=detail,
302                                                       closure=foo, func=dummy, data=dummy)
303         self.assertEqual(count, 1)
304         obj.emit('my-signal')
305         self.assertEqual(obj.status, 1)
306
307         # Unblocking by the same match criteria allows callback to work again
308         count = GObject.signal_handlers_unblock_matched(obj,
309                                                         GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
310                                                         signal_id=signal_id, detail=detail,
311                                                         closure=foo, func=dummy, data=dummy)
312         self.assertEqual(count, 1)
313         obj.emit('my-signal')
314         self.assertEqual(obj.status, 2)
315
316         # Disconnecting by match criteria completely removes the handler
317         count = GObject.signal_handlers_disconnect_matched(obj,
318                                                            GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
319                                                            signal_id=signal_id, detail=detail,
320                                                            closure=foo, func=dummy, data=dummy)
321         self.assertEqual(count, 1)
322         obj.emit('my-signal')
323         self.assertEqual(obj.status, 2)
324
325     def test_signal_handler_find(self):
326         def dummy(*args):
327             "Hack to work around: "
328
329         def foo(obj):
330             obj.status += 1
331
332         obj = self.Object()
333         handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)
334
335         signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
336         found_id = GObject.signal_handler_find(obj,
337                                                GObject.SignalMatchType.ID,
338                                                signal_id=signal_id, detail=detail,
339                                                closure=None, func=dummy, data=dummy)
340         self.assertEqual(handler_id, found_id)
341
342
343 class TestClosures(unittest.TestCase):
344     def setUp(self):
345         self.count = 0
346         self.emission_stopped = False
347         self.emission_error = False
348         self.handler_pending = False
349
350     def _callback_handler_pending(self, e):
351         signal_id, detail = GObject.signal_parse_name('signal', e, True)
352         self.handler_pending = GObject.signal_has_handler_pending(e, signal_id, detail,
353                                                                   may_be_blocked=False)
354
355     def _callback(self, e):
356         self.count += 1
357
358     def _callback_stop_emission(self, obj, prop, stop_it):
359         if stop_it:
360             obj.stop_emission_by_name('notify::prop')
361             self.emission_stopped = True
362         else:
363             self.count += 1
364
365     def _callback_invalid_stop_emission_name(self, obj, prop):
366         # We expect a GLib warning but there currently is no way to test that
367         # This can at least make sure we don't crash
368         old_mask = GLib.log_set_always_fatal(GLib.LogLevelFlags.LEVEL_CRITICAL |
369                                              GLib.LogLevelFlags.LEVEL_ERROR)
370         try:
371             obj.stop_emission_by_name('notasignal::baddetail')
372         finally:
373             GLib.log_set_always_fatal(old_mask)
374             self.emission_error = True
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         e.set_property('prop', 1234)
410         self.assertTrue(self.emission_error)
411
412     def test_handler_block(self):
413         e = E()
414         e.connect('signal', self._callback)
415         e.handler_block_by_func(self._callback)
416         e.emit('signal')
417         self.assertEqual(self.count, 0)
418
419     def test_handler_unblock(self):
420         e = E()
421         handler_id = e.connect('signal', self._callback)
422         e.handler_block(handler_id)
423         e.handler_unblock_by_func(self._callback)
424         e.emit('signal')
425         self.assertEqual(self.count, 1)
426
427     def test_handler_block_method(self):
428         # Filed as #375589
429         class A:
430             def __init__(self):
431                 self.a = 0
432
433             def callback(self, o):
434                 self.a = 1
435                 o.handler_block_by_func(self.callback)
436
437         inst = A()
438         e = E()
439         e.connect("signal", inst.callback)
440         e.emit('signal')
441         self.assertEqual(inst.a, 1)
442         gc.collect()
443
444     def test_gstring(self):
445         class C(GObject.GObject):
446             __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_GSTRING,
447                                           (GObject.TYPE_GSTRING,))}
448
449             def __init__(self, test):
450                 GObject.GObject.__init__(self)
451                 self.test = test
452
453             def do_my_signal(self, data):
454                 self.data = data
455                 self.test.assertEqual(len(data), 3)
456                 return ''.join([data[2], data[1], data[0]])
457         c = C(self)
458         data = c.emit("my_signal", "\01\00\02")
459         self.assertEqual(data, "\02\00\01")
460
461     def test_handler_pending(self):
462         obj = F()
463         obj.connect('signal', self._callback_handler_pending)
464         obj.connect('signal', self._callback)
465
466         self.assertEqual(self.count, 0)
467         self.assertEqual(self.handler_pending, False)
468
469         obj.emit('signal')
470         self.assertEqual(self.count, 1)
471         self.assertEqual(self.handler_pending, True)
472
473     def test_signal_handlers_destroy(self):
474         obj = F()
475         obj.connect('signal', self._callback)
476         obj.connect('signal', self._callback)
477         obj.connect('signal', self._callback)
478
479         obj.emit('signal')
480         self.assertEqual(self.count, 3)
481
482         # count should remain at 3 after all handlers are destroyed
483         GObject.signal_handlers_destroy(obj)
484         obj.emit('signal')
485         self.assertEqual(self.count, 3)
486
487
488 class SigPropClass(GObject.GObject):
489     __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
490                                   (GObject.TYPE_INT,))}
491
492     __gproperties__ = {
493         'foo': (str, None, None, '', GObject.PARAM_WRITABLE | GObject.PARAM_CONSTRUCT),
494         }
495
496     signal_emission_failed = False
497
498     def do_my_signal(self, arg):
499         self.arg = arg
500
501     def do_set_property(self, pspec, value):
502         if pspec.name == 'foo':
503             self._foo = value
504         else:
505             raise AttributeError('unknown property %s' % pspec.name)
506         try:
507             self.emit("my-signal", 1)
508         except TypeError:
509             self.signal_emission_failed = True
510
511
512 class TestSigProp(unittest.TestCase):
513     def test_emit_in_property_setter(self):
514         obj = SigPropClass()
515         self.assertFalse(obj.signal_emission_failed)
516
517
518 class CM(GObject.GObject):
519     __gsignals__ = dict(
520         test1=(GObject.SignalFlags.RUN_FIRST, None, ()),
521         test2=(GObject.SignalFlags.RUN_LAST, None, (str,)),
522         test3=(GObject.SignalFlags.RUN_LAST, int, (GObject.TYPE_DOUBLE,)),
523         test4=(GObject.SignalFlags.RUN_FIRST, None,
524                (bool, _long, GObject.TYPE_FLOAT, GObject.TYPE_DOUBLE, int,
525                 GObject.TYPE_UINT, GObject.TYPE_ULONG)),
526         test_float=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_FLOAT, (GObject.TYPE_FLOAT,)),
527         test_double=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_DOUBLE, (GObject.TYPE_DOUBLE,)),
528         test_int64=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_INT64, (GObject.TYPE_INT64,)),
529         test_string=(GObject.SignalFlags.RUN_LAST, str, (str,)),
530         test_object=(GObject.SignalFlags.RUN_LAST, object, (object,)),
531         test_paramspec=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, ()),
532         test_paramspec_in=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, (GObject.ParamSpec, )),
533         test_gvalue=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.Value,)),
534         test_gvalue_ret=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.TYPE_GTYPE,)),
535     )
536
537     testprop = GObject.Property(type=int)
538
539
540 class _TestCMarshaller:
541     def setUp(self):
542         self.obj = CM()
543         testhelper.connectcallbacks(self.obj)
544
545     def test_test1(self):
546         self.obj.emit("test1")
547
548     def test_test2(self):
549         self.obj.emit("test2", "string")
550
551     def test_test3(self):
552         rv = self.obj.emit("test3", 42.0)
553         self.assertEqual(rv, 20)
554
555     def test_test4(self):
556         self.obj.emit("test4", True, _long(10), 3.14, 1.78, 20, _long(30), _long(31))
557
558     def test_float(self):
559         rv = self.obj.emit("test-float", 1.234)
560         self.assertTrue(rv >= 1.233999 and rv <= 1.2400001, rv)
561
562     def test_double(self):
563         rv = self.obj.emit("test-double", 1.234)
564         self.assertEqual(rv, 1.234)
565
566     def test_int64(self):
567         rv = self.obj.emit("test-int64", 102030405)
568         self.assertEqual(rv, 102030405)
569
570         rv = self.obj.emit("test-int64", GObject.G_MAXINT64)
571         self.assertEqual(rv, GObject.G_MAXINT64 - 1)
572
573         rv = self.obj.emit("test-int64", GObject.G_MININT64)
574         self.assertEqual(rv, GObject.G_MININT64)
575
576     def test_string(self):
577         rv = self.obj.emit("test-string", "str")
578         self.assertEqual(rv, "str")
579
580     def test_object(self):
581         rv = self.obj.emit("test-object", self)
582         self.assertEqual(rv, self)
583
584     def test_paramspec(self):
585         rv = self.obj.emit("test-paramspec")
586         self.assertEqual(rv.name, "test-param")
587         self.assertEqual(rv.nick, "test")
588
589     @unittest.skipUnless(hasattr(GObject, 'param_spec_boolean'),
590                          'too old gobject-introspection')
591     def test_paramspec_in(self):
592         rv = GObject.param_spec_boolean('mybool', 'test-bool', 'do something',
593                                         True, GObject.ParamFlags.READABLE)
594
595         rv2 = self.obj.emit("test-paramspec-in", rv)
596         self.assertEqual(type(rv), type(rv2))
597         self.assertEqual(rv2.name, "mybool")
598         self.assertEqual(rv2.nick, "test-bool")
599
600     def test_C_paramspec(self):
601         self.notify_called = False
602
603         def cb_notify(obj, prop):
604             self.notify_called = True
605             self.assertEqual(obj, self.obj)
606             self.assertEqual(prop.name, "testprop")
607
608         self.obj.connect("notify", cb_notify)
609         self.obj.set_property("testprop", 42)
610         self.assertTrue(self.notify_called)
611
612     def test_gvalue(self):
613         # implicit int
614         rv = self.obj.emit("test-gvalue", 42)
615         self.assertEqual(rv, 42)
616
617         # explicit float
618         v = GObject.Value(GObject.TYPE_FLOAT, 1.234)
619         rv = self.obj.emit("test-gvalue", v)
620         self.assertAlmostEqual(rv, 1.234, 4)
621
622         # implicit float
623         rv = self.obj.emit("test-gvalue", 1.234)
624         self.assertAlmostEqual(rv, 1.234, 4)
625
626         # explicit int64
627         v = GObject.Value(GObject.TYPE_INT64, GObject.G_MAXINT64)
628         rv = self.obj.emit("test-gvalue", v)
629         self.assertEqual(rv, GObject.G_MAXINT64)
630
631         # implicit int64
632         # does not work, see https://bugzilla.gnome.org/show_bug.cgi?id=683775
633         #rv = self.obj.emit("test-gvalue", GObject.G_MAXINT64)
634         #self.assertEqual(rv, GObject.G_MAXINT64)
635
636         # explicit uint64
637         v = GObject.Value(GObject.TYPE_UINT64, GObject.G_MAXUINT64)
638         rv = self.obj.emit("test-gvalue", v)
639         self.assertEqual(rv, GObject.G_MAXUINT64)
640
641         # implicit uint64
642         # does not work, see https://bugzilla.gnome.org/show_bug.cgi?id=683775
643         #rv = self.obj.emit("test-gvalue", GObject.G_MAXUINT64)
644         #self.assertEqual(rv, GObject.G_MAXUINT64)
645
646     def test_gvalue_ret(self):
647         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT),
648                          GObject.G_MAXINT)
649         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT),
650                          GObject.G_MAXUINT)
651         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT64),
652                          GObject.G_MAXINT64)
653         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT64),
654                          GObject.G_MAXUINT64)
655         self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_STRING),
656                          "hello")
657
658 if 'generic-c-marshaller' in GObject.features:
659     class TestCMarshaller(_TestCMarshaller, unittest.TestCase):
660         pass
661 else:
662     print()
663     print('** WARNING: LIBFFI disabled, not testing')
664     print()
665
666 # Test for 374653
667
668
669 class TestPyGValue(unittest.TestCase):
670     def test_none_null_boxed_conversion(self):
671         class C(GObject.GObject):
672             __gsignals__ = dict(my_boxed_signal=(
673                 GObject.SignalFlags.RUN_LAST,
674                 GObject.TYPE_STRV, ()))
675
676         obj = C()
677         obj.connect('my-boxed-signal', lambda obj: None)
678         sys.last_type = None
679         obj.emit('my-boxed-signal')
680         assert not sys.last_type
681
682
683 class TestSignalDecorator(unittest.TestCase):
684     class Decorated(GObject.GObject):
685         value = 0
686
687         @GObject.Signal
688         def pushed(self):
689             """this will push"""
690             self.value += 1
691
692         @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
693         def pulled(self):
694             self.value -= 1
695
696         stomped = GObject.Signal('stomped', arg_types=(int,), doc='this will stomp')
697         unnamed = GObject.Signal()
698
699     class DecoratedOverride(GObject.GObject):
700         overridden_closure_called = False
701         notify_called = False
702         value = GObject.Property(type=int, default=0)
703
704         @GObject.SignalOverride
705         def notify(self, *args, **kargs):
706             self.overridden_closure_called = True
707             #GObject.GObject.notify(self, *args, **kargs)
708
709         def on_notify(self, obj, prop):
710             self.notify_called = True
711
712     def setUp(self):
713         self.unnamedCalled = False
714
715     def onUnnamed(self, obj):
716         self.unnamedCalled = True
717
718     def test_get_signal_args(self):
719         self.assertEqual(self.Decorated.pushed.get_signal_args(),
720                          (GObject.SignalFlags.RUN_FIRST, None, tuple()))
721         self.assertEqual(self.Decorated.pulled.get_signal_args(),
722                          (GObject.SignalFlags.RUN_LAST, None, tuple()))
723         self.assertEqual(self.Decorated.stomped.get_signal_args(),
724                          (GObject.SignalFlags.RUN_FIRST, None, (int,)))
725
726     def test_closures_called(self):
727         decorated = self.Decorated()
728         self.assertEqual(decorated.value, 0)
729         decorated.pushed.emit()
730         self.assertEqual(decorated.value, 1)
731         decorated.pulled.emit()
732         self.assertEqual(decorated.value, 0)
733
734     def test_signal_copy(self):
735         blah = self.Decorated.stomped.copy('blah')
736         self.assertEqual(str(blah), blah)
737         self.assertEqual(blah.func, self.Decorated.stomped.func)
738         self.assertEqual(blah.flags, self.Decorated.stomped.flags)
739         self.assertEqual(blah.return_type, self.Decorated.stomped.return_type)
740         self.assertEqual(blah.arg_types, self.Decorated.stomped.arg_types)
741         self.assertEqual(blah.__doc__, self.Decorated.stomped.__doc__)
742
743     def test_doc_string(self):
744         # Test the two techniques for setting doc strings on the signals
745         # class variables, through the "doc" keyword or as the getter doc string.
746         self.assertEqual(self.Decorated.stomped.__doc__, 'this will stomp')
747         self.assertEqual(self.Decorated.pushed.__doc__, 'this will push')
748
749     def test_unnamed_signal_gets_named(self):
750         self.assertEqual(str(self.Decorated.unnamed), 'unnamed')
751
752     def test_unnamed_signal_gets_called(self):
753         obj = self.Decorated()
754         obj.connect('unnamed', self.onUnnamed)
755         self.assertEqual(self.unnamedCalled, False)
756         obj.emit('unnamed')
757         self.assertEqual(self.unnamedCalled, True)
758
759     def NOtest_overridden_signal(self):
760         # Test that the pushed signal is called in with super and the override
761         # which should both increment the "value" to 3
762         obj = self.DecoratedOverride()
763         obj.connect("notify", obj.on_notify)
764         self.assertEqual(obj.value, 0)
765         #obj.notify.emit()
766         obj.value = 1
767         self.assertEqual(obj.value, 1)
768         self.assertTrue(obj.overridden_closure_called)
769         self.assertTrue(obj.notify_called)
770
771
772 class TestSignalConnectors(unittest.TestCase):
773     class CustomButton(GObject.GObject):
774         on_notify_called = False
775         value = GObject.Property(type=int)
776
777         @GObject.Signal(arg_types=(int,))
778         def clicked(self, value):
779             self.value = value
780
781     def setUp(self):
782         self.obj = None
783         self.value = None
784
785     def on_clicked(self, obj, value):
786         self.obj = obj
787         self.value = value
788
789     def test_signal_notify(self):
790         def on_notify(obj, param):
791             obj.on_notify_called = True
792
793         obj = self.CustomButton()
794         obj.connect('notify', on_notify)
795         self.assertFalse(obj.on_notify_called)
796         obj.notify('value')
797         self.assertTrue(obj.on_notify_called)
798
799     def test_signal_emit(self):
800         # standard callback connection with different forms of emit.
801         obj = self.CustomButton()
802         obj.connect('clicked', self.on_clicked)
803
804         # vanilla
805         obj.emit('clicked', 1)
806         self.assertEqual(obj.value, 1)
807         self.assertEqual(obj, self.obj)
808         self.assertEqual(self.value, 1)
809
810         # using class signal as param
811         self.obj = None
812         self.value = None
813         obj.emit(self.CustomButton.clicked, 1)
814         self.assertEqual(obj, self.obj)
815         self.assertEqual(self.value, 1)
816
817         # using bound signal as param
818         self.obj = None
819         self.value = None
820         obj.emit(obj.clicked, 1)
821         self.assertEqual(obj, self.obj)
822         self.assertEqual(self.value, 1)
823
824         # using bound signal with emit
825         self.obj = None
826         self.value = None
827         obj.clicked.emit(1)
828         self.assertEqual(obj, self.obj)
829         self.assertEqual(self.value, 1)
830
831     def test_signal_class_connect(self):
832         obj = self.CustomButton()
833         obj.connect(self.CustomButton.clicked, self.on_clicked)
834         obj.emit('clicked', 2)
835         self.assertEqual(obj, self.obj)
836         self.assertEqual(self.value, 2)
837
838     def test_signal_bound_connect(self):
839         obj = self.CustomButton()
840         obj.clicked.connect(self.on_clicked)
841         obj.emit('clicked', 3)
842         self.assertEqual(obj, self.obj)
843         self.assertEqual(self.value, 3)
844
845
846 class TestInstallSignals(unittest.TestCase):
847     # These tests only test how signalhelper.install_signals works
848     # with the __gsignals__ dict and therefore does not need to use
849     # GObject as a base class because that would automatically call
850     # install_signals within the meta-class.
851     class Base(object):
852         __gsignals__ = {'test': (0, None, tuple())}
853
854     class Sub1(Base):
855         pass
856
857     class Sub2(Base):
858         @GObject.Signal
859         def sub2test(self):
860             pass
861
862     def setUp(self):
863         self.assertEqual(len(self.Base.__gsignals__), 1)
864         signalhelper.install_signals(self.Base)
865         self.assertEqual(len(self.Base.__gsignals__), 1)
866
867     def test_subclass_gets_empty_gsignals_dict(self):
868         # Installing signals will add the __gsignals__ dict to a class
869         # if it doesn't already exists.
870         self.assertFalse('__gsignals__' in self.Sub1.__dict__)
871         signalhelper.install_signals(self.Sub1)
872         self.assertTrue('__gsignals__' in self.Sub1.__dict__)
873         # Sub1 should only contain an empty signals dict, this tests:
874         # https://bugzilla.gnome.org/show_bug.cgi?id=686496
875         self.assertEqual(self.Sub1.__dict__['__gsignals__'], {})
876
877     def test_subclass_with_decorator_gets_gsignals_dict(self):
878         self.assertFalse('__gsignals__' in self.Sub2.__dict__)
879         signalhelper.install_signals(self.Sub2)
880         self.assertTrue('__gsignals__' in self.Sub2.__dict__)
881         self.assertEqual(len(self.Base.__gsignals__), 1)
882         self.assertEqual(len(self.Sub2.__gsignals__), 1)
883         self.assertTrue('sub2test' in self.Sub2.__gsignals__)
884
885         # Make sure the vfunc was added
886         self.assertTrue(hasattr(self.Sub2, 'do_sub2test'))
887
888
889 # For this test to work with both python2 and 3 we need to dynamically
890 # exec the given code due to the new syntax causing an error in python 2.
891 annotated_class_code = """
892 class AnnotatedSignalClass(GObject.GObject):
893     @GObject.Signal
894     def sig1(self, a:int, b:float):
895         pass
896
897     @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
898     def sig2_with_return(self, a:int, b:float) -> str:
899         return "test"
900 """
901
902
903 class TestPython3Signals(unittest.TestCase):
904     AnnotatedClass = None
905
906     def setUp(self):
907         if sys.version_info >= (3, 0):
908             exec(annotated_class_code, globals(), globals())
909             self.assertTrue('AnnotatedSignalClass' in globals())
910             self.AnnotatedClass = globals()['AnnotatedSignalClass']
911
912     def test_annotations(self):
913         if self.AnnotatedClass:
914             self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig1.func),
915                              (None, (int, float)))
916             self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig2_with_return.func),
917                              (str, (int, float)))
918
919             self.assertEqual(self.AnnotatedClass.sig2_with_return.get_signal_args(),
920                              (GObject.SignalFlags.RUN_LAST, str, (int, float)))
921             self.assertEqual(self.AnnotatedClass.sig2_with_return.arg_types,
922                              (int, float))
923             self.assertEqual(self.AnnotatedClass.sig2_with_return.return_type,
924                              str)
925
926
927 class TestSignalModuleLevelFunctions(unittest.TestCase):
928     @unittest.skipIf(sys.version_info < (2, 7), 'Requires Python >= 2.7')
929     def test_signal_list_ids_with_invalid_type(self):
930         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
931             GObject.signal_list_ids(GObject.TYPE_INVALID)
932
933     @unittest.skipIf(sys.version_info < (2, 7), 'Requires Python >= 2.7')
934     def test_signal_list_ids(self):
935         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
936             GObject.signal_list_ids(GObject.TYPE_INT)
937
938         ids = GObject.signal_list_ids(C)
939         self.assertEqual(len(ids), 1)
940         # Note canonicalized names
941         self.assertEqual(GObject.signal_name(ids[0]), 'my-signal')
942         # There is no signal 0 in gobject
943         self.assertEqual(GObject.signal_name(0), None)
944
945     @unittest.skipIf(sys.version_info < (2, 7), 'Requires Python >= 2.7')
946     def test_signal_lookup_with_invalid_type(self):
947         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
948             GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INVALID)
949
950     @unittest.skipIf(sys.version_info < (2, 7), 'Requires Python >= 2.7')
951     def test_signal_lookup(self):
952         ids = GObject.signal_list_ids(C)
953         self.assertEqual(ids[0], GObject.signal_lookup('my_signal', C))
954         self.assertEqual(ids[0], GObject.signal_lookup('my-signal', C))
955
956         with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
957             GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INT)
958
959         # Invalid signal names return 0 instead of raising
960         self.assertEqual(GObject.signal_lookup('NOT_A_SIGNAL_NAME', C),
961                          0)
962
963     def test_signal_query(self):
964         my_signal_id, = GObject.signal_list_ids(C)
965
966         # Form is: (id, name, gtype, arg_count, return_type, (arg_type1, ...))
967         my_signal_expected_query_result = [my_signal_id, 'my-signal', C.__gtype__,
968                                            1, GObject.TYPE_NONE, (GObject.TYPE_INT,)]
969         # signal_query(name, type)
970         self.assertEqual(list(GObject.signal_query('my-signal', C)), my_signal_expected_query_result)
971         # signal_query(signal_id)
972         self.assertEqual(list(GObject.signal_query(my_signal_id)), my_signal_expected_query_result)
973         # invalid query returns None instead of raising
974         self.assertEqual(GObject.signal_query(0), None)
975         self.assertEqual(GObject.signal_query('NOT_A_SIGNAL', C),
976                          None)
977
978
979 if __name__ == '__main__':
980     unittest.main()