path: root/tests/test_signal.py

# -*- Mode: Python -*-

import gc
import unittest
import sys
import weakref
import threading
import time

from gi.repository import GObject, GLib, Regress, Gio
from gi import _signalhelper as signalhelper
from gi.module import repository as repo

import testhelper
from .helper import capture_glib_warnings, capture_gi_deprecation_warnings


class C(GObject.GObject):
    __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
                                  (GObject.TYPE_INT,))}

    def do_my_signal(self, arg):
        self.arg = arg


class D(C):
    def do_my_signal(self, arg2):
        self.arg2 = arg2
        C.do_my_signal(self, arg2)


class TestSignalCreation(unittest.TestCase):
    # Bug 540376.
    def test_illegals(self):
        self.assertRaises(TypeError, lambda: GObject.signal_new('test',
                                                                None,
                                                                0,
                                                                None,
                                                                (GObject.TYPE_LONG,)))


class TestChaining(unittest.TestCase):
    def setUp(self):
        self.inst = C()
        self.inst.connect("my_signal", self.my_signal_handler_cb, 1, 2, 3)

    def my_signal_handler_cb(self, *args):
        assert len(args) == 5
        assert isinstance(args[0], C)
        assert args[0] == self.inst

        assert isinstance(args[1], int)
        assert args[1] == 42

        assert args[2:] == (1, 2, 3)

    def test_chaining(self):
        self.inst.emit("my_signal", 42)
        assert self.inst.arg == 42

    def test_chaining2(self):
        inst2 = D()
        inst2.emit("my_signal", 44)
        assert inst2.arg == 44
        assert inst2.arg2 == 44

# This is for bug 153718


class TestGSignalsError(unittest.TestCase):
    def test_invalid_type(self, *args):
        def foo():
            class Foo(GObject.GObject):
                __gsignals__ = None
        self.assertRaises(TypeError, foo)
        gc.collect()

    def test_invalid_name(self, *args):
        def foo():
            class Foo(GObject.GObject):
                __gsignals__ = {'not-exists': 'override'}

        with capture_glib_warnings(allow_warnings=True):
            self.assertRaises(TypeError, foo)
        gc.collect()


class TestGPropertyError(unittest.TestCase):
    def test_invalid_type(self, *args):
        def foo():
            class Foo(GObject.GObject):
                __gproperties__ = None
        self.assertRaises(TypeError, foo)
        gc.collect()

    def test_invalid_name(self, *args):
        def foo():
            class Foo(GObject.GObject):
                __gproperties__ = {None: None}

        self.assertRaises(TypeError, foo)
        gc.collect()


class TestList(unittest.TestCase):
    def test_list_names(self):
        self.assertEqual(GObject.signal_list_names(C), ('my-signal',))


def my_accumulator(ihint, return_accu, handler_return, user_data):
    """An accumulator that stops emission when the sum of handler
    returned values reaches 3"""
    assert user_data == "accum data"
    if return_accu >= 3:
        return False, return_accu
    return True, return_accu + handler_return


class Foo(GObject.GObject):
    my_acc_signal = GObject.Signal(return_type=GObject.TYPE_INT,
                                   flags=GObject.SignalFlags.RUN_LAST,
                                   accumulator=my_accumulator,
                                   accu_data="accum data")

    my_other_acc_signal = GObject.Signal(return_type=GObject.TYPE_BOOLEAN,
                                         flags=GObject.SignalFlags.RUN_LAST,
                                         accumulator=GObject.signal_accumulator_true_handled)

    my_acc_first_wins = GObject.Signal(return_type=GObject.TYPE_BOOLEAN,
                                       flags=GObject.SignalFlags.RUN_LAST,
                                       accumulator=GObject.signal_accumulator_first_wins)


class TestAccumulator(unittest.TestCase):

    def test_accumulator(self):
        inst = Foo()
        inst.my_acc_signal.connect(lambda obj: 1)
        inst.my_acc_signal.connect(lambda obj: 2)
        # the value returned in the following handler will not be
        # considered, because at this point the accumulator already
        # reached its limit.
        inst.my_acc_signal.connect(lambda obj: 3)
        retval = inst.my_acc_signal.emit()
        self.assertEqual(retval, 3)

    def test_accumulator_true_handled(self):
        inst = Foo()
        inst.my_other_acc_signal.connect(self._true_handler1)
        inst.my_other_acc_signal.connect(self._true_handler2)
        # the following handler will not be called because handler2
        # returns True, so it should stop the emission.
        inst.my_other_acc_signal.connect(self._true_handler3)
        self.__true_val = None
        inst.my_other_acc_signal.emit()
        self.assertEqual(self.__true_val, 2)

    def test_accumulator_first_wins(self):
        # First signal hit will always win
        inst = Foo()
        inst.my_acc_first_wins.connect(self._true_handler3)
        inst.my_acc_first_wins.connect(self._true_handler1)
        inst.my_acc_first_wins.connect(self._true_handler2)
        self.__true_val = None
        inst.my_acc_first_wins.emit()
        self.assertEqual(self.__true_val, 3)

    def _true_handler1(self, obj):
        self.__true_val = 1
        return False

    def _true_handler2(self, obj):
        self.__true_val = 2
        return True

    def _true_handler3(self, obj):
        self.__true_val = 3
        return False


class E(GObject.GObject):
    __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
                               ())}

    # Property used to test detailed signal
    prop = GObject.Property(type=int, default=0)

    def __init__(self):
        GObject.GObject.__init__(self)
        self.status = 0

    def do_signal(self):
        assert self.status == 0
        self.status = 1


class F(GObject.GObject):
    __gsignals__ = {'signal': (GObject.SignalFlags.RUN_FIRST, None,
                               ())}

    def __init__(self):
        GObject.GObject.__init__(self)
        self.status = 0

    def do_signal(self):
        self.status += 1


class TestEmissionHook(unittest.TestCase):
    def test_add(self):
        self.hook = True
        e = E()
        e.connect('signal', self._callback)
        GObject.add_emission_hook(E, "signal", self._emission_hook)
        e.emit('signal')
        self.assertEqual(e.status, 3)

    def test_remove(self):
        self.hook = False
        e = E()
        e.connect('signal', self._callback)
        hook_id = GObject.add_emission_hook(E, "signal", self._emission_hook)
        GObject.remove_emission_hook(E, "signal", hook_id)
        e.emit('signal')
        self.assertEqual(e.status, 3)

    def _emission_hook(self, e):
        self.assertEqual(e.status, 1)
        e.status = 2

    def _callback(self, e):
        if self.hook:
            self.assertEqual(e.status, 2)
        else:
            self.assertEqual(e.status, 1)
        e.status = 3

    def test_callback_return_false(self):
        self.hook = False
        obj = F()

        def _emission_hook(obj):
            obj.status += 1
            return False
        GObject.add_emission_hook(obj, "signal", _emission_hook)
        obj.emit('signal')
        obj.emit('signal')
        self.assertEqual(obj.status, 3)

    def test_callback_return_true(self):
        self.hook = False
        obj = F()

        def _emission_hook(obj):
            obj.status += 1
            return True
        hook_id = GObject.add_emission_hook(obj, "signal", _emission_hook)
        obj.emit('signal')
        obj.emit('signal')
        GObject.remove_emission_hook(obj, "signal", hook_id)
        self.assertEqual(obj.status, 4)

    def test_callback_return_true_but_remove(self):
        self.hook = False
        obj = F()

        def _emission_hook(obj):
            obj.status += 1
            return True
        hook_id = GObject.add_emission_hook(obj, "signal", _emission_hook)
        obj.emit('signal')
        GObject.remove_emission_hook(obj, "signal", hook_id)
        obj.emit('signal')
        self.assertEqual(obj.status, 3)


class TestMatching(unittest.TestCase):
    class Object(GObject.Object):
        status = 0
        prop = GObject.Property(type=int, default=0)

        @GObject.Signal()
        def my_signal(self):
            pass

    @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=692918
    def test_signal_handler_block_matching(self):
        def dummy(*args):
            "Hack to work around: "

        def foo(obj):
            obj.status += 1

        obj = self.Object()
        handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)
        handler_id

        self.assertEqual(obj.status, 0)
        obj.emit('my-signal')
        self.assertEqual(obj.status, 1)

        # Blocking by match criteria disables the foo callback
        signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
        count = GObject.signal_handlers_block_matched(obj,
                                                      GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
                                                      signal_id=signal_id, detail=detail,
                                                      closure=foo, func=dummy, data=dummy)
        self.assertEqual(count, 1)
        obj.emit('my-signal')
        self.assertEqual(obj.status, 1)

        # Unblocking by the same match criteria allows callback to work again
        count = GObject.signal_handlers_unblock_matched(obj,
                                                        GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
                                                        signal_id=signal_id, detail=detail,
                                                        closure=foo, func=dummy, data=dummy)
        self.assertEqual(count, 1)
        obj.emit('my-signal')
        self.assertEqual(obj.status, 2)

        # Disconnecting by match criteria completely removes the handler
        count = GObject.signal_handlers_disconnect_matched(obj,
                                                           GObject.SignalMatchType.ID | GObject.SignalMatchType.CLOSURE,
                                                           signal_id=signal_id, detail=detail,
                                                           closure=foo, func=dummy, data=dummy)
        self.assertEqual(count, 1)
        obj.emit('my-signal')
        self.assertEqual(obj.status, 2)

    def test_signal_handler_find(self):
        def foo(obj):
            obj.status += 1

        obj = self.Object()
        handler_id = GObject.signal_connect_closure(obj, 'my-signal', foo, after=False)

        signal_id, detail = GObject.signal_parse_name('my-signal', obj, True)
        found_id = GObject.signal_handler_find(obj,
                                               GObject.SignalMatchType.ID,
                                               signal_id=signal_id, detail=detail,
                                               closure=None, func=0, data=0)
        self.assertEqual(handler_id, found_id)


class TestClosures(unittest.TestCase):
    def setUp(self):
        self.count = 0
        self.emission_stopped = False
        self.emission_error = False
        self.handler_pending = False

    def _callback_handler_pending(self, e):
        signal_id, detail = GObject.signal_parse_name('signal', e, True)
        self.handler_pending = GObject.signal_has_handler_pending(e, signal_id, detail,
                                                                  may_be_blocked=False)

    def _callback(self, e):
        self.count += 1

    def _callback_stop_emission(self, obj, prop, stop_it):
        if stop_it:
            obj.stop_emission_by_name('notify::prop')
            self.emission_stopped = True
        else:
            self.count += 1

    def _callback_invalid_stop_emission_name(self, obj, prop):
        with capture_glib_warnings(allow_warnings=True) as warn:
            obj.stop_emission_by_name('notasignal::baddetail')
            self.emission_error = True
            self.assertTrue(warn)

    def test_disconnect_by_func(self):
        e = E()
        e.connect('signal', self._callback)
        e.disconnect_by_func(self._callback)
        e.emit('signal')
        self.assertEqual(self.count, 0)

    def test_disconnect(self):
        e = E()
        handler_id = e.connect('signal', self._callback)
        self.assertTrue(e.handler_is_connected(handler_id))
        e.disconnect(handler_id)
        e.emit('signal')
        self.assertEqual(self.count, 0)
        self.assertFalse(e.handler_is_connected(handler_id))

    def test_stop_emission_by_name(self):
        e = E()

        # Sandwich a callback that stops emission in between a callback that increments
        e.connect('notify::prop', self._callback_stop_emission, False)
        e.connect('notify::prop', self._callback_stop_emission, True)
        e.connect('notify::prop', self._callback_stop_emission, False)

        e.set_property('prop', 1234)
        self.assertEqual(e.get_property('prop'), 1234)
        self.assertEqual(self.count, 1)
        self.assertTrue(self.emission_stopped)

    def test_stop_emission_by_name_error(self):
        e = E()

        e.connect('notify::prop', self._callback_invalid_stop_emission_name)
        with capture_glib_warnings():
            e.set_property('prop', 1234)
        self.assertTrue(self.emission_error)

    def test_handler_block(self):
        e = E()
        e.connect('signal', self._callback)
        e.handler_block_by_func(self._callback)
        e.emit('signal')
        self.assertEqual(self.count, 0)

    def test_handler_unblock(self):
        e = E()
        handler_id = e.connect('signal', self._callback)
        e.handler_block(handler_id)
        e.handler_unblock_by_func(self._callback)
        e.emit('signal')
        self.assertEqual(self.count, 1)

    def test_handler_block_method(self):
        # Filed as #375589
        class A:
            def __init__(self):
                self.a = 0

            def callback(self, o):
                self.a = 1
                o.handler_block_by_func(self.callback)

        inst = A()
        e = E()
        e.connect("signal", inst.callback)
        e.emit('signal')
        self.assertEqual(inst.a, 1)
        gc.collect()

    def test_gstring(self):
        class C(GObject.GObject):
            __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_LAST, GObject.TYPE_GSTRING,
                                          (GObject.TYPE_GSTRING,))}

            def __init__(self, test):
                GObject.GObject.__init__(self)
                self.test = test

            def do_my_signal(self, data):
                self.data = data
                self.test.assertEqual(len(data), 3)
                return ''.join([data[2], data[1], data[0]])
        c = C(self)
        data = c.emit("my_signal", "\01\00\02")
        self.assertEqual(data, "\02\00\01")

    def test_handler_pending(self):
        obj = F()
        obj.connect('signal', self._callback_handler_pending)
        obj.connect('signal', self._callback)

        self.assertEqual(self.count, 0)
        self.assertEqual(self.handler_pending, False)

        obj.emit('signal')
        self.assertEqual(self.count, 1)
        self.assertEqual(self.handler_pending, True)

    def test_signal_handlers_destroy(self):
        obj = F()
        obj.connect('signal', self._callback)
        obj.connect('signal', self._callback)
        obj.connect('signal', self._callback)

        obj.emit('signal')
        self.assertEqual(self.count, 3)

        # count should remain at 3 after all handlers are destroyed
        GObject.signal_handlers_destroy(obj)
        obj.emit('signal')
        self.assertEqual(self.count, 3)


class SigPropClass(GObject.GObject):
    __gsignals__ = {'my_signal': (GObject.SignalFlags.RUN_FIRST, None,
                                  (GObject.TYPE_INT,))}

    __gproperties__ = {
        'foo': (str, None, None, '',
                GObject.ParamFlags.WRITABLE | GObject.ParamFlags.CONSTRUCT),
        }

    signal_emission_failed = False

    def do_my_signal(self, arg):
        self.arg = arg

    def do_set_property(self, pspec, value):
        if pspec.name == 'foo':
            self._foo = value
        else:
            raise AttributeError('unknown property %s' % pspec.name)
        try:
            self.emit("my-signal", 1)
        except TypeError:
            self.signal_emission_failed = True


class TestSigProp(unittest.TestCase):
    def test_emit_in_property_setter(self):
        obj = SigPropClass()
        self.assertFalse(obj.signal_emission_failed)


class CM(GObject.GObject):
    __gsignals__ = dict(
        test1=(GObject.SignalFlags.RUN_FIRST, None, ()),
        test2=(GObject.SignalFlags.RUN_LAST, None, (str,)),
        test3=(GObject.SignalFlags.RUN_LAST, int, (GObject.TYPE_DOUBLE,)),
        test4=(GObject.SignalFlags.RUN_FIRST, None,
               (bool, int, GObject.TYPE_FLOAT, GObject.TYPE_DOUBLE, int,
                GObject.TYPE_UINT, GObject.TYPE_ULONG)),
        test_float=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_FLOAT, (GObject.TYPE_FLOAT,)),
        test_double=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_DOUBLE, (GObject.TYPE_DOUBLE,)),
        test_int64=(GObject.SignalFlags.RUN_LAST, GObject.TYPE_INT64, (GObject.TYPE_INT64,)),
        test_string=(GObject.SignalFlags.RUN_LAST, str, (str,)),
        test_object=(GObject.SignalFlags.RUN_LAST, object, (object,)),
        test_paramspec=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, ()),
        test_paramspec_in=(GObject.SignalFlags.RUN_LAST, GObject.ParamSpec, (GObject.ParamSpec, )),
        test_gvalue=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.Value,)),
        test_gvalue_ret=(GObject.SignalFlags.RUN_LAST, GObject.Value, (GObject.TYPE_GTYPE,)),
    )

    testprop = GObject.Property(type=int)


class _TestCMarshaller:
    def setUp(self):
        self.obj = CM()
        testhelper.connectcallbacks(self.obj)

    def test_test1(self):
        self.obj.emit("test1")

    def test_test2(self):
        self.obj.emit("test2", "string")

    def test_test3(self):
        rv = self.obj.emit("test3", 42.0)
        self.assertEqual(rv, 20)

    def test_test4(self):
        self.obj.emit("test4", True, 10, 3.14, 1.78, 20, 30, 31)

    def test_float(self):
        rv = self.obj.emit("test-float", 1.234)
        self.assertTrue(rv >= 1.233999 and rv <= 1.2400001, rv)

    def test_double(self):
        rv = self.obj.emit("test-double", 1.234)
        self.assertEqual(rv, 1.234)

    def test_int64(self):
        rv = self.obj.emit("test-int64", 102030405)
        self.assertEqual(rv, 102030405)

        rv = self.obj.emit("test-int64", GLib.MAXINT64)
        self.assertEqual(rv, GLib.MAXINT64 - 1)

        rv = self.obj.emit("test-int64", GLib.MININT64)
        self.assertEqual(rv, GLib.MININT64)

    def test_string(self):
        rv = self.obj.emit("test-string", "str")
        self.assertEqual(rv, "str")

    def test_object(self):
        rv = self.obj.emit("test-object", self)
        self.assertEqual(rv, self)

    def test_paramspec(self):
        rv = self.obj.emit("test-paramspec")
        self.assertEqual(rv.name, "test-param")
        self.assertEqual(rv.nick, "test")

    @unittest.skipUnless(hasattr(GObject, 'param_spec_boolean'),
                         'too old gobject-introspection')
    def test_paramspec_in(self):
        rv = GObject.param_spec_boolean('mybool', 'test-bool', 'do something',
                                        True, GObject.ParamFlags.READABLE)

        rv2 = self.obj.emit("test-paramspec-in", rv)
        self.assertEqual(type(rv), type(rv2))
        self.assertEqual(rv2.name, "mybool")
        self.assertEqual(rv2.nick, "test-bool")

    def test_C_paramspec(self):
        self.notify_called = False

        def cb_notify(obj, prop):
            self.notify_called = True
            self.assertEqual(obj, self.obj)
            self.assertEqual(prop.name, "testprop")

        self.obj.connect("notify", cb_notify)
        self.obj.set_property("testprop", 42)
        self.assertTrue(self.notify_called)

    def test_gvalue(self):
        # implicit int
        rv = self.obj.emit("test-gvalue", 42)
        self.assertEqual(rv, 42)

        # explicit float
        v = GObject.Value(GObject.TYPE_FLOAT, 1.234)
        rv = self.obj.emit("test-gvalue", v)
        self.assertAlmostEqual(rv, 1.234, places=4)

        # implicit float
        rv = self.obj.emit("test-gvalue", 1.234)
        self.assertAlmostEqual(rv, 1.234, places=4)

        # explicit int64
        v = GObject.Value(GObject.TYPE_INT64, GLib.MAXINT64)
        rv = self.obj.emit("test-gvalue", v)
        self.assertEqual(rv, GLib.MAXINT64)

        # explicit uint64
        v = GObject.Value(GObject.TYPE_UINT64, GLib.MAXUINT64)
        rv = self.obj.emit("test-gvalue", v)
        self.assertEqual(rv, GLib.MAXUINT64)

    @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=705291
    def test_gvalue_implicit_int64(self):
        # implicit int64
        rv = self.obj.emit("test-gvalue", GLib.MAXINT64)
        self.assertEqual(rv, GLib.MAXINT64)

        # implicit uint64
        rv = self.obj.emit("test-gvalue", GLib.MAXUINT64)
        self.assertEqual(rv, GLib.MAXUINT64)

    def test_gvalue_ret(self):
        self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT),
                         GLib.MAXINT)
        self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT),
                         GLib.MAXUINT)
        self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_INT64),
                         GLib.MAXINT64)
        self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_UINT64),
                         GLib.MAXUINT64)
        self.assertEqual(self.obj.emit("test-gvalue-ret", GObject.TYPE_STRING),
                         "hello")


class TestCMarshaller(_TestCMarshaller, unittest.TestCase):
    pass


# Test for 374653


class TestPyGValue(unittest.TestCase):
    def test_none_null_boxed_conversion(self):
        class C(GObject.GObject):
            __gsignals__ = dict(my_boxed_signal=(
                GObject.SignalFlags.RUN_LAST,
                GObject.TYPE_STRV, ()))

        obj = C()
        obj.connect('my-boxed-signal', lambda obj: None)
        sys.last_type = None
        obj.emit('my-boxed-signal')
        assert not sys.last_type


class TestSignalDecorator(unittest.TestCase):
    class Decorated(GObject.GObject):
        value = 0

        @GObject.Signal
        def pushed(self):
            """this will push"""
            self.value += 1

        @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
        def pulled(self):
            self.value -= 1

        @GObject.Signal(flags=GObject.SignalFlags.DETAILED)
        def detailed(self):
            self.value -= 1

        stomped = GObject.Signal('stomped', arg_types=(int,), doc='this will stomp')
        unnamed = GObject.Signal()

    class DecoratedOverride(GObject.GObject):
        overridden_closure_called = False
        notify_called = False
        value = GObject.Property(type=int, default=0)

        @GObject.SignalOverride
        def notify(self, *args, **kargs):
            self.overridden_closure_called = True

        def on_notify(self, obj, prop):
            self.notify_called = True

    def setUp(self):
        self.unnamedCalled = False

    def onUnnamed(self, obj):
        self.unnamedCalled = True

    def test_disconnect(self):
        decorated = self.Decorated()
        id_ = decorated.pushed.connect(lambda *args: None)
        decorated.pushed.disconnect(id_)

    def test_signal_repr(self):
        decorated = self.Decorated()
        assert repr(decorated.pushed) == 'BoundSignal("pushed")'

    def test_signal_call(self):
        decorated = self.Decorated()
        assert decorated.value == 0
        decorated.pushed()
        assert decorated.value == 1

    def test_connect_detailed(self):
        decorated = self.Decorated()
        id_ = decorated.detailed.connect_detailed(lambda *args: None, "foo")
        decorated.pushed.disconnect(id_)

    def test_get_signal_args(self):
        self.assertEqual(self.Decorated.pushed.get_signal_args(),
                         (GObject.SignalFlags.RUN_FIRST, None, tuple(), None, None))
        self.assertEqual(self.Decorated.pulled.get_signal_args(),
                         (GObject.SignalFlags.RUN_LAST, None, tuple(), None, None))
        self.assertEqual(self.Decorated.stomped.get_signal_args(),
                         (GObject.SignalFlags.RUN_FIRST, None, (int,), None, None))

    def test_closures_called(self):
        decorated = self.Decorated()
        self.assertEqual(decorated.value, 0)
        decorated.pushed.emit()
        self.assertEqual(decorated.value, 1)
        decorated.pulled.emit()
        self.assertEqual(decorated.value, 0)

    def test_signal_copy(self):
        blah = self.Decorated.stomped.copy('blah')
        self.assertEqual(str(blah), blah)
        self.assertEqual(blah.func, self.Decorated.stomped.func)
        self.assertEqual(blah.flags, self.Decorated.stomped.flags)
        self.assertEqual(blah.return_type, self.Decorated.stomped.return_type)
        self.assertEqual(blah.arg_types, self.Decorated.stomped.arg_types)
        self.assertEqual(blah.__doc__, self.Decorated.stomped.__doc__)

    def test_doc_string(self):
        # Test the two techniques for setting doc strings on the signals
        # class variables, through the "doc" keyword or as the getter doc string.
        self.assertEqual(self.Decorated.stomped.__doc__, 'this will stomp')
        self.assertEqual(self.Decorated.pushed.__doc__, 'this will push')

    def test_unnamed_signal_gets_named(self):
        self.assertEqual(str(self.Decorated.unnamed), 'unnamed')

    def test_unnamed_signal_gets_called(self):
        obj = self.Decorated()
        obj.connect('unnamed', self.onUnnamed)
        self.assertEqual(self.unnamedCalled, False)
        obj.emit('unnamed')
        self.assertEqual(self.unnamedCalled, True)

    def test_overridden_signal(self):
        # Test that the pushed signal is called in with super and the override
        # which should both increment the "value" to 3
        obj = self.DecoratedOverride()
        obj.connect("notify", obj.on_notify)
        self.assertEqual(obj.value, 0)
        obj.value = 1
        self.assertEqual(obj.value, 1)
        self.assertTrue(obj.overridden_closure_called)
        self.assertTrue(obj.notify_called)


class TestSignalConnectors(unittest.TestCase):
    class CustomButton(GObject.GObject):
        on_notify_called = False
        value = GObject.Property(type=int)

        @GObject.Signal(arg_types=(int,))
        def clicked(self, value):
            self.value = value

    def setUp(self):
        self.obj = None
        self.value = None

    def on_clicked(self, obj, value):
        self.obj = obj
        self.value = value

    def test_signal_notify(self):
        def on_notify(obj, param):
            obj.on_notify_called = True

        obj = self.CustomButton()
        obj.connect('notify', on_notify)
        self.assertFalse(obj.on_notify_called)
        obj.notify('value')
        self.assertTrue(obj.on_notify_called)

    def test_signal_emit(self):
        # standard callback connection with different forms of emit.
        obj = self.CustomButton()
        obj.connect('clicked', self.on_clicked)

        # vanilla
        obj.emit('clicked', 1)
        self.assertEqual(obj.value, 1)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 1)

        # using class signal as param
        self.obj = None
        self.value = None
        obj.emit(self.CustomButton.clicked, 1)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 1)

        # using bound signal as param
        self.obj = None
        self.value = None
        obj.emit(obj.clicked, 1)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 1)

        # using bound signal with emit
        self.obj = None
        self.value = None
        obj.clicked.emit(1)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 1)

    def test_signal_class_connect(self):
        obj = self.CustomButton()
        obj.connect(self.CustomButton.clicked, self.on_clicked)
        obj.emit('clicked', 2)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 2)

    def test_signal_bound_connect(self):
        obj = self.CustomButton()
        obj.clicked.connect(self.on_clicked)
        obj.emit('clicked', 3)
        self.assertEqual(obj, self.obj)
        self.assertEqual(self.value, 3)


class _ConnectDataTestBase(object):
    # Notes:
    #  - self.Object is overridden in sub-classes.
    #  - Numeric suffixes indicate the number of user data args passed in.
    Object = None

    def run_connect_test(self, emit_args, user_data, flags=0):
        obj = self.Object()
        callback_args = []

        def callback(*args):
            callback_args.append(args)
            return 0

        obj.connect_data('sig-with-int64-prop', callback, connect_flags=flags, *user_data)
        obj.emit('sig-with-int64-prop', *emit_args)
        self.assertEqual(len(callback_args), 1)
        return callback_args[0]

    def test_0(self):
        obj, value = self.run_connect_test([GLib.MAXINT64], user_data=[])
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)

    def test_1(self):
        obj, value, data = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'])
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_after_0(self):
        obj, value = self.run_connect_test([GLib.MAXINT64],
                                           user_data=[],
                                           flags=GObject.ConnectFlags.AFTER)
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)

    def test_after_1(self):
        obj, value, data = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'],
                                                 flags=GObject.ConnectFlags.AFTER)
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_swaped_0(self):
        # Swapped only works with a single user data argument.
        with self.assertRaises(ValueError):
            self.run_connect_test([GLib.MAXINT64],
                                  user_data=[],
                                  flags=GObject.ConnectFlags.SWAPPED)

    def test_swaped_1(self):
        # Notice obj and data are reversed in the return.
        data, value, obj = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'],
                                                 flags=GObject.ConnectFlags.SWAPPED)
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_swaped_2(self):
        # Swapped only works with a single user data argument.
        with self.assertRaises(ValueError):
            self.run_connect_test([GLib.MAXINT64],
                                  user_data=[1, 2],
                                  flags=GObject.ConnectFlags.SWAPPED)

    def test_after_and_swapped_0(self):
        # Swapped only works with a single user data argument.
        with self.assertRaises(ValueError):
            self.run_connect_test([GLib.MAXINT64],
                                  user_data=[],
                                  flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)

    def test_after_and_swapped_1(self):
        # Notice obj and data are reversed in the return.
        data, value, obj = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'],
                                                 flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)
        self.assertIsInstance(obj, self.Object)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_after_and_swapped_2(self):
        # Swapped only works with a single user data argument.
        with self.assertRaises(ValueError):
            self.run_connect_test([GLib.MAXINT64],
                                  user_data=[],
                                  flags=GObject.ConnectFlags.AFTER | GObject.ConnectFlags.SWAPPED)


class TestConnectDataNonIntrospected(unittest.TestCase, _ConnectDataTestBase):
    # This tests connect_data with non-introspected signals
    # (created in Python in this case).
    class Object(GObject.Object):
        test = GObject.Signal()
        sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
                                             arg_types=[GObject.TYPE_INT64],
                                             flags=GObject.SignalFlags.RUN_LAST)


class TestConnectDataIntrospected(unittest.TestCase, _ConnectDataTestBase):
    # This tests connect_data with introspected signals brought in from Regress.
    Object = Regress.TestObj


class TestInstallSignals(unittest.TestCase):
    # These tests only test how signalhelper.install_signals works
    # with the __gsignals__ dict and therefore does not need to use
    # GObject as a base class because that would automatically call
    # install_signals within the meta-class.
    class Base(object):
        __gsignals__ = {'test': (0, None, tuple())}

    class Sub1(Base):
        pass

    class Sub2(Base):
        @GObject.Signal
        def sub2test(self):
            pass

    def setUp(self):
        self.assertEqual(len(self.Base.__gsignals__), 1)
        signalhelper.install_signals(self.Base)
        self.assertEqual(len(self.Base.__gsignals__), 1)

    def test_subclass_gets_empty_gsignals_dict(self):
        # Installing signals will add the __gsignals__ dict to a class
        # if it doesn't already exists.
        self.assertFalse('__gsignals__' in self.Sub1.__dict__)
        signalhelper.install_signals(self.Sub1)
        self.assertTrue('__gsignals__' in self.Sub1.__dict__)
        # Sub1 should only contain an empty signals dict, this tests:
        # https://bugzilla.gnome.org/show_bug.cgi?id=686496
        self.assertEqual(self.Sub1.__dict__['__gsignals__'], {})

    def test_subclass_with_decorator_gets_gsignals_dict(self):
        self.assertFalse('__gsignals__' in self.Sub2.__dict__)
        signalhelper.install_signals(self.Sub2)
        self.assertTrue('__gsignals__' in self.Sub2.__dict__)
        self.assertEqual(len(self.Base.__gsignals__), 1)
        self.assertEqual(len(self.Sub2.__gsignals__), 1)
        self.assertTrue('sub2test' in self.Sub2.__gsignals__)

        # Make sure the vfunc was added
        self.assertTrue(hasattr(self.Sub2, 'do_sub2test'))


class TestPython3Signals(unittest.TestCase):

    class AnnotatedClass(GObject.GObject):
        @GObject.Signal
        def sig1(self, a: int, b: float):
            pass

        @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
        def sig2_with_return(self, a: int, b: float) -> str:
            return "test"

    def test_annotations(self):
        self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig1.func),
                         (None, (int, float)))
        self.assertEqual(signalhelper.get_signal_annotations(self.AnnotatedClass.sig2_with_return.func),
                         (str, (int, float)))

        self.assertEqual(self.AnnotatedClass.sig2_with_return.get_signal_args(),
                         (GObject.SignalFlags.RUN_LAST, str, (int, float), None, None))
        self.assertEqual(self.AnnotatedClass.sig2_with_return.arg_types,
                         (int, float))
        self.assertEqual(self.AnnotatedClass.sig2_with_return.return_type,
                         str)

    def test_emit_return(self):
        obj = self.AnnotatedClass()
        self.assertEqual(obj.sig2_with_return.emit(1, 2.0),
                         'test')


class TestSignalModuleLevelFunctions(unittest.TestCase):
    def test_signal_list_ids_with_invalid_type(self):
        with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
            GObject.signal_list_ids(GObject.TYPE_INVALID)

    def test_signal_list_ids(self):
        with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
            GObject.signal_list_ids(GObject.TYPE_INT)

        ids = GObject.signal_list_ids(C)
        self.assertEqual(len(ids), 1)
        # Note canonicalized names
        self.assertEqual(GObject.signal_name(ids[0]), 'my-signal')
        # There is no signal 0 in gobject
        self.assertEqual(GObject.signal_name(0), None)

    def test_signal_lookup_with_invalid_type(self):
        with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
            GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INVALID)

    def test_signal_lookup(self):
        ids = GObject.signal_list_ids(C)
        self.assertEqual(ids[0], GObject.signal_lookup('my_signal', C))
        self.assertEqual(ids[0], GObject.signal_lookup('my-signal', C))

        with self.assertRaisesRegex(TypeError, 'type must be instantiable or an interface.*'):
            GObject.signal_lookup('NOT_A_SIGNAL_NAME', GObject.TYPE_INT)

        # Invalid signal names return 0 instead of raising
        self.assertEqual(GObject.signal_lookup('NOT_A_SIGNAL_NAME', C),
                         0)

    def test_signal_query(self):
        my_signal_id, = GObject.signal_list_ids(C)

        # Form is: (id, name, gtype, arg_count, return_type, (arg_type1, ...))
        my_signal_expected_query_result = [my_signal_id, 'my-signal', C.__gtype__,
                                           1, GObject.TYPE_NONE, (GObject.TYPE_INT,)]
        # signal_query(name, type)
        self.assertEqual(list(GObject.signal_query('my-signal', C)), my_signal_expected_query_result)
        # signal_query(signal_id)
        self.assertEqual(list(GObject.signal_query(my_signal_id)), my_signal_expected_query_result)
        # invalid query returns None instead of raising
        self.assertEqual(GObject.signal_query(0), None)
        self.assertEqual(GObject.signal_query('NOT_A_SIGNAL', C),
                         None)


class TestIntrospectedSignals(unittest.TestCase):
    def test_object_param_signal(self):
        obj = Regress.TestObj()

        def callback(obj, obj_param):
            self.assertEqual(obj_param.props.int, 3)
            self.assertGreater(obj_param.__grefcount__, 1)
            obj.called = True

        obj.called = False
        obj.connect('sig-with-obj', callback)
        obj.emit_sig_with_obj()
        self.assertTrue(obj.called)

    def test_connect_after(self):
        obj = Regress.TestObj()

        def callback(obj, obj_param):
            obj.called = True

        obj.called = False
        obj.connect_after('sig-with-obj', callback)
        obj.emit_sig_with_obj()
        self.assertTrue(obj.called)

    def test_int64_param_from_py(self):
        obj = Regress.TestObj()

        def callback(obj, i):
            obj.callback_i = i
            return i

        obj.callback_i = None
        obj.connect('sig-with-int64-prop', callback)
        rv = obj.emit('sig-with-int64-prop', GLib.MAXINT64)
        self.assertEqual(rv, GLib.MAXINT64)
        self.assertEqual(obj.callback_i, GLib.MAXINT64)

    def test_uint64_param_from_py(self):
        obj = Regress.TestObj()

        def callback(obj, i):
            obj.callback_i = i
            return i

        obj.callback_i = None
        obj.connect('sig-with-uint64-prop', callback)
        rv = obj.emit('sig-with-uint64-prop', GLib.MAXUINT64)
        self.assertEqual(rv, GLib.MAXUINT64)
        self.assertEqual(obj.callback_i, GLib.MAXUINT64)

    def test_int64_param_from_c(self):
        obj = Regress.TestObj()

        def callback(obj, i):
            obj.callback_i = i
            return i

        obj.callback_i = None

        obj.connect('sig-with-int64-prop', callback)
        obj.emit_sig_with_int64()
        self.assertEqual(obj.callback_i, GLib.MAXINT64)

    def test_uint64_param_from_c(self):
        obj = Regress.TestObj()

        def callback(obj, i):
            obj.callback_i = i
            return i

        obj.callback_i = None

        obj.connect('sig-with-uint64-prop', callback)
        obj.emit_sig_with_uint64()
        self.assertEqual(obj.callback_i, GLib.MAXUINT64)

    def test_intarray_ret(self):
        obj = Regress.TestObj()

        def callback(obj, i):
            obj.callback_i = i
            return [i, i + 1]

        obj.callback_i = None

        try:
            obj.connect('sig-with-intarray-ret', callback)
        except TypeError as e:
            # compat with g-i 1.34.x
            if 'unknown signal' in str(e):
                return
            raise

        rv = obj.emit('sig-with-intarray-ret', 42)
        self.assertEqual(obj.callback_i, 42)
        self.assertEqual(type(rv), GLib.Array)
        self.assertEqual(rv.len, 2)

    @unittest.skip('https://bugzilla.gnome.org/show_bug.cgi?id=669496')
    def test_array_parm(self):
        obj = Regress.TestObj()

        def callback(obj, arr):
            obj.callback_arr = arr

        obj.connect('sig-with-array-prop', callback)
        obj.callback_arr = None
        self.assertEqual(obj.emit('sig-with-array-prop', [1, 2, GLib.MAXUINT]), None)
        self.assertEqual(obj.callback_arr, [1, 2, GLib.MAXUINT])

    def test_held_struct_ref(self):
        held_structs = []

        def callback(obj, struct):
            # The struct held by Python will become a copy after this callback exits.
            struct.some_int = 42
            struct.some_int8 = 42
            held_structs.append(struct)

        struct = Regress.TestSimpleBoxedA()
        obj = Regress.TestObj()

        self.assertEqual(struct.some_int, 0)
        self.assertEqual(struct.some_int8, 0)

        obj.connect('test-with-static-scope-arg', callback)
        obj.emit('test-with-static-scope-arg', struct)

        # The held struct will be a copy of the modified struct.
        self.assertEqual(len(held_structs), 1)
        held_struct = held_structs[0]
        self.assertEqual(held_struct.some_int, 42)
        self.assertEqual(held_struct.some_int8, 42)

        # Boxed equality checks pointers by default.
        self.assertNotEqual(struct, held_struct)

    def test_action(self):
        obj = Regress.TestAction()
        other_obj = obj.emit('action')
        self.assertEqual(other_obj.__grefcount__, 1)
        other_obj2 = obj.emit('action2')
        self.assertIsNone(other_obj2)


class TestIntrospectedSignalsIssue158(unittest.TestCase):
    """
    The test for https://gitlab.gnome.org/GNOME/pygobject/issues/158
    """
    _obj_sig_names = [sig.get_name() for sig in repo.find_by_name('Regress', 'TestObj').get_signals()]

    def __init__(self, *args):
        unittest.TestCase.__init__(self, *args)
        self._gc_thread_stop = False

    def _gc_thread(self):
        while not self._gc_thread_stop:
            gc.collect()
            time.sleep(0.010)

    def _callback(self, *args):
        pass

    def test_run(self):
        """
        Manually trigger GC from a different thread periodicaly
        while the main thread keeps connecting/disconnecting to/from signals.

        It takes a lot of time to reproduce the issue. It is possible to make it
        fail reliably by changing the code of pygobject_unwatch_closure slightly from:
          PyGObjectData *inst_data = data;
          inst_data->closures = g_slist_remove (inst_data->closures, closure);
        to
          PyGObjectData *inst_data = data;
          GSList *tmp = g_slist_remove (inst_data->closures, closure);
          g_usleep(G_USEC_PER_SEC/10);
          inst_data->closures = tmp;
        """
        obj = Regress.TestObj()
        gc_thread = threading.Thread(target=self._gc_thread)
        gc_thread.start()

        for _ in range(8):
            handlers = [obj.connect(sig, self._callback) for sig in self._obj_sig_names]
            time.sleep(0.010)
            while len(handlers) > 0:
                obj.disconnect(handlers.pop())

        self._gc_thread_stop = True
        gc_thread.join()


class _ConnectObjectTestBase(object):
    # Notes:
    #  - self.Object is overridden in sub-classes.
    #  - Numeric suffixes indicate the number of user data args passed in.
    Object = None
    SwapObject = None

    def run_connect_test(self, emit_args, user_data, flags=0):
        obj = self.Object()
        callback_args = []
        swap_obj = self.SwapObject()

        def callback(*args):
            callback_args.append(args)
            return 0

        if flags & GObject.ConnectFlags.AFTER:
            connect_func = obj.connect_object_after
        else:
            connect_func = obj.connect_object

        with capture_gi_deprecation_warnings():
            connect_func('sig-with-int64-prop', callback, swap_obj, *user_data)
        obj.emit('sig-with-int64-prop', *emit_args)
        self.assertEqual(len(callback_args), 1)
        return callback_args[0]

    def test_0(self):
        obj, value = self.run_connect_test([GLib.MAXINT64], user_data=[])
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)

    def test_1(self):
        obj, value, data = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'])
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_2(self):
        obj, value, data1, data2 = self.run_connect_test([GLib.MAXINT64],
                                                         user_data=['mydata1', 'mydata2'])
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data1, 'mydata1')
        self.assertEqual(data2, 'mydata2')

    def test_after_0(self):
        obj, value = self.run_connect_test([GLib.MAXINT64],
                                           user_data=[],
                                           flags=GObject.ConnectFlags.AFTER)
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)

    def test_after_1(self):
        obj, value, data = self.run_connect_test([GLib.MAXINT64],
                                                 user_data=['mydata'],
                                                 flags=GObject.ConnectFlags.AFTER)
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data, 'mydata')

    def test_after_2(self):
        obj, value, data1, data2 = self.run_connect_test([GLib.MAXINT64],
                                                         user_data=['mydata1', 'mydata2'],
                                                         flags=GObject.ConnectFlags.AFTER)
        self.assertIsInstance(obj, self.SwapObject)
        self.assertEqual(value, GLib.MAXINT64)
        self.assertEqual(data1, 'mydata1')
        self.assertEqual(data2, 'mydata2')


class TestConnectGObjectNonIntrospected(unittest.TestCase, _ConnectObjectTestBase):
    # This tests connect_object with non-introspected signals
    # (created in Python in this case).
    class Object(GObject.Object):
        test = GObject.Signal()
        sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
                                             arg_types=[GObject.TYPE_INT64],
                                             flags=GObject.SignalFlags.RUN_LAST)

    # Object passed for swapping is GObject based.
    class SwapObject(GObject.Object):
        pass


class TestConnectGObjectIntrospected(unittest.TestCase, _ConnectObjectTestBase):
    # This tests connect_object with introspected signals brought in from Regress.
    Object = Regress.TestObj

    # Object passed for swapping is GObject based.
    class SwapObject(GObject.Object):
        pass


class TestConnectPyObjectNonIntrospected(unittest.TestCase, _ConnectObjectTestBase):
    # This tests connect_object with non-introspected signals
    # (created in Python in this case).
    class Object(GObject.Object):
        test = GObject.Signal()
        sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
                                             arg_types=[GObject.TYPE_INT64],
                                             flags=GObject.SignalFlags.RUN_LAST)

    # Object passed for swapping is pure Python
    SwapObject = object


class TestConnectPyObjectIntrospected(unittest.TestCase, _ConnectObjectTestBase):
    # This tests connect_object with introspected signals brought in from Regress.
    Object = Regress.TestObj

    # Object passed for swapping is pure Python
    SwapObject = object


class _RefCountTestBase(object):
    # NOTE: ref counts are always one more than expected because the getrefcount()
    # function adds a ref for the input argument.

    # Sub-classes set this
    Object = None

    class PyData(object):
        pass

    @unittest.skipUnless(hasattr(sys, "getrefcount"), "no sys.getrefcount")
    def test_callback_ref_count_del(self):
        def callback(obj, value):
            return value // 2

        callback_ref = weakref.ref(callback)
        self.assertEqual(sys.getrefcount(callback), 2)

        obj = self.Object()
        obj.connect('sig-with-int64-prop', callback)
        self.assertEqual(sys.getrefcount(callback), 3)

        del callback
        self.assertEqual(sys.getrefcount(callback_ref()), 2)

        res = obj.emit('sig-with-int64-prop', 42)
        self.assertEqual(res, 21)
        self.assertEqual(sys.getrefcount(callback_ref), 2)

        del obj
        self.assertIsNone(callback_ref())

    @unittest.skipUnless(hasattr(sys, "getrefcount"), "no sys.getrefcount")
    def test_callback_ref_count_disconnect(self):
        def callback(obj, value):
            return value // 2

        callback_ref = weakref.ref(callback)
        self.assertEqual(sys.getrefcount(callback), 2)

        obj = self.Object()
        handler_id = obj.connect('sig-with-int64-prop', callback)
        self.assertEqual(sys.getrefcount(callback), 3)

        del callback
        self.assertEqual(sys.getrefcount(callback_ref()), 2)

        res = obj.emit('sig-with-int64-prop', 42)
        self.assertEqual(res, 21)
        self.assertEqual(sys.getrefcount(callback_ref), 2)

        obj.disconnect(handler_id)
        self.assertIsNone(callback_ref())

    @unittest.skipUnless(hasattr(sys, "getrefcount"), "no sys.getrefcount")
    def test_callback_ref_count_disconnect_by_func(self):
        def callback(obj, value):
            return value // 2

        callback_ref = weakref.ref(callback)
        self.assertEqual(sys.getrefcount(callback), 2)

        obj = self.Object()
        obj.connect('sig-with-int64-prop', callback)
        self.assertEqual(sys.getrefcount(callback), 3)

        del callback
        self.assertEqual(sys.getrefcount(callback_ref()), 2)

        res = obj.emit('sig-with-int64-prop', 42)
        self.assertEqual(res, 21)
        self.assertEqual(sys.getrefcount(callback_ref), 2)

        obj.disconnect_by_func(callback_ref())
        self.assertIsNone(callback_ref())

    @unittest.skipUnless(hasattr(sys, "getrefcount"), "no sys.getrefcount")
    def test_user_data_ref_count(self):
        def callback(obj, value, data):
            return value // 2

        data = self.PyData()
        data_ref = weakref.ref(data)
        self.assertEqual(sys.getrefcount(data), 2)

        obj = self.Object()
        obj.connect('sig-with-int64-prop', callback, data)
        self.assertEqual(sys.getrefcount(data), 3)

        del data
        self.assertEqual(sys.getrefcount(data_ref()), 2)

        res = obj.emit('sig-with-int64-prop', 42)
        self.assertEqual(res, 21)
        self.assertEqual(sys.getrefcount(data_ref()), 2)

        del obj
        self.assertIsNone(data_ref())

    @unittest.skipUnless(hasattr(sys, "getrefcount"), "no sys.getrefcount")
    @unittest.expectedFailure  # https://bugzilla.gnome.org/show_bug.cgi?id=688064
    def test_object_ref_count(self):
        # connect_object() should only weakly reference the object passed in
        # and auto-disconnect the signal when the object is destroyed.
        def callback(data, value):
            return value // 2

        data = GObject.Object()
        data_ref = weakref.ref(data)
        self.assertEqual(sys.getrefcount(data), 2)

        obj = self.Object()
        handler_id = obj.connect_object('sig-with-int64-prop', callback, data)
        self.assertEqual(sys.getrefcount(data), 2)

        res = obj.emit('sig-with-int64-prop', 42)
        self.assertEqual(res, 21)
        self.assertEqual(sys.getrefcount(data), 2)

        del data

        self.assertIsNone(data_ref())
        self.assertFalse(obj.handler_is_connected(handler_id))


class TestRefCountsNonIntrospected(unittest.TestCase, _RefCountTestBase):
    class Object(GObject.Object):
        sig_with_int64_prop = GObject.Signal(return_type=GObject.TYPE_INT64,
                                             arg_types=[GObject.TYPE_INT64],
                                             flags=GObject.SignalFlags.RUN_LAST)


class TestRefCountsIntrospected(unittest.TestCase, _RefCountTestBase):
    Object = Regress.TestObj


class TestClosureRefCycle(unittest.TestCase):

    def test_closure_ref_cycle_unreachable(self):
        # https://bugzilla.gnome.org/show_bug.cgi?id=731501

        called = []

        def on_add(store, *args):
            called.append(store)

        store = Gio.ListStore()
        store.connect_object('items-changed', on_add, store)

        # Remove all Python references to the object and keep it alive
        # on the C level.
        x = Gio.FileInfo()
        x.set_attribute_object('store', store)
        del store
        gc.collect()

        # get it back and trigger the signal
        x.get_attribute_object('store').append(Gio.FileInfo())

        self.assertEqual(len(called), 1)
        self.assertTrue(called[0].__grefcount__ > 0)