# -*- Mode: Python -*-

import unittest
import gc
import weakref

from gi.repository import GLib, GObject

class Idle(GLib.Idle):
    def __init__(self, loop):
        GLib.Idle.__init__(self)
        self.count = 0
        self.set_callback(self.callback, loop)

    def callback(self, loop):
        self.count += 1
        return True


class MySource(GLib.Source):
    def __init__(self):
        GLib.Source.__init__(self)

    def prepare(self):
        return True, 0

    def check(self):
        return True

    def dispatch(self, callback, args):
        return callback(*args)


# update a list on weakref callback
def on_weakref_dead(d, i = 0):
    def callback(r):
        d[i] = True

    return callback

# the tests themselves
class TestSource(unittest.TestCase):
    def timeout_callback(self, loop):
        loop.quit()

    def my_callback(self, loop):
        self.pos += 1
        return True

    def setup_timeout(self, loop):
        timeout = GLib.Timeout(500)
        timeout.set_callback(self.timeout_callback, loop)
        timeout.attach()

    def testSources(self):
        loop = GLib.MainLoop()

        self.setup_timeout(loop)

        idle = Idle(loop)
        idle.attach()

        self.pos = 0

        m = MySource()
        m.set_callback(self.my_callback, loop)
        m.attach()

        loop.run()

        assert self.pos >= 0 and idle.count >= 0

    def testSourcePrepare(self):
        # this test may not terminate if prepare() is wrapped incorrectly
        dispatched = [False]
        loop = GLib.MainLoop()

        class CustomTimeout(GLib.Source):
            def prepare(self):
                return (False, 10)

            def check(self):
                return True

            def dispatch(self, callback, args):
                dispatched[0] = True

                loop.quit()

                return False

        source = CustomTimeout()

        source.attach()
        source.set_callback(dir)

        loop.run()

        assert dispatched[0]


class TestTimeout(unittest.TestCase):
    def test504337(self):
        timeout_source = GLib.Timeout(20)
        idle_source = GLib.Idle()


    # will likely segfault without a fix for bug #504337
    def testTimeoutBug504337(self):
        def f():
            timeout_source = GObject.Timeout(20)

        f()

        if gc.isenabled():
            gc.collect()

    # will likely segfault without a fix for bug #504337
    def testIdleBug504337(self):
        def f():
            timeout_source = GObject.Idle()

        f()

        if gc.isenabled():
            gc.collect()

    # unattached, should be dealloced after return and gc
    def testDeallocAfterReturnAndGC(self):
        if not gc.isenabled():
            return

        d = [False]

        def f():
            s = GObject.Source()
            r = weakref.ref(s, on_weakref_dead(d))

            return r

        r = f()

        gc.collect()

        assert d[0]

    # destroyed without an attach(), should be dealloced
    def testDestroyWithoutAttach(self):
        d = [False]

        def f():
            s = GObject.Source()
            r = weakref.ref(s, on_weakref_dead(d))

            s.destroy()

            if gc.isenabled():
                # a flawed implementation might have a zero refcount here
                gc.collect()

            return r

        r = f()

        assert d[0]

    # attached, should be dealloced after destroy() and return
    def testDeallocAfterDestroy(self):
        d = [False]
        c = GObject.MainContext()

        class DerivedSource(GObject.Source):
            # despite finalize
            def finalize(self):
                pass

        def f():
            s = DerivedSource()
            r = weakref.ref(s, on_weakref_dead(d))

            s.attach(c)
            s.destroy()

            return (r, c)

        (r, c) = f()

        assert d[0]

    # gc should dealloc after context is destroyed on return
    def testDeallocAfterContextDeathAndGC(self):
        if not gc.isenabled():
            return

        d = [False]

        def f():
            s = GObject.Source()
            c = GObject.MainContext()
            r = weakref.ref(s, on_weakref_dead(d))

            s.attach(c)

            return r

        r = f()

        gc.collect()

        assert d[0]

    # should not be dealloced; still attached
    def testStillAttached(self):
        d = [False]

        def f():
            c = GObject.MainContext()
            s = GObject.Source()
            r = weakref.ref(s, on_weakref_dead(d))

            s.attach(c)

            return (r, c)

        (r, c) = f()

        assert not d[0]

    # finalizers can cause undefined behavior?
    def testFinalizerSelfOperation(self):
        # self ops
        c = GObject.MainContext()
        d = [False]

        class FinalizingSource(GObject.Source):
            def finalize(self):
                try:
                    self.attach(c)
                    self.destroy()
                except ValueError:
                    d[0] = True

        def f():
            s = FinalizingSource()

            s.attach(c)
            s.destroy()

            return None

        f()

        assert d[0]

    # finalizer code can resurrect a dealloced instance
    def testFinalizerResurrection(self):
        d = [None]

        class FinalizingSource(GObject.Source):
            def finalize(self):
                d[0] = self

        def f():
            c = GObject.MainContext()
            s = FinalizingSource()

            s.attach(c)
            s.destroy()

        f()

        # is the runtime is still sane?
        assert d[0] != None

        try:
            d[0].set_callback(lambda a: None)
        except:
            pass
        else:
            assert False

    # do we fail to collect if we self-ref via a callback?
    def testLoopClosureCycle(self):
        # be aware this test will probably fail; no
        # implementation handles this case correctly yet
        d = [False]

        def f():
            loop = GObject.MainLoop()
            tos = GObject.Timeout(1)
            r = weakref.ref(tos, on_weakref_dead(d))

            def callback():
                loop.quit()

                return True

            tos.set_callback(callback)
            tos.attach(loop.get_context())

            loop.run()

            return r

        r = f()

        assert d[0]

if __name__ == '__main__':
    unittest.main()

