Index: tests/test_source.py
===================================================================
--- tests/test_source.py	(revision 746)
+++ tests/test_source.py	(working copy)
@@ -5,7 +5,10 @@
 import sys
 import select
 import unittest
+import gc
+import weakref
 
+from trampoline import Trampoline
 from common import gobject
 
 class Idle(gobject.Idle):
@@ -31,6 +34,14 @@
     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()
@@ -62,5 +73,226 @@
 
         assert self.pos >= 0 and idle.count >= 0
 
+    # 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]
+
+    # attached, should be dealloced after destroy() and return
+    def testDeallocAfterDestroy(self):
+        d = [False]
+        c = gobject.MainContext()
+
+        class DerivedSource(gobject.Source):
+            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
+
+    # wrapper implementation stress test
+    def testCoroutinesApp(self):
+        # set up context
+        loop = gobject.MainLoop()
+        context = loop.get_context()
+        trampoline = Trampoline()
+        pauses = 200
+        d = [False] * pauses
+        refs = []
+
+        # simple "wait a while" coroutine
+        def wait_y(trampoline, context, timeout_ms, i):
+            tos = gobject.Timeout(timeout_ms)
+            tt = trampoline.this_thread
+
+            refs.append(weakref.ref(tos, on_weakref_dead(d, i)))
+
+            def callback():
+                trampoline.run([tt])
+
+            tos.set_callback(callback)
+            tos.attach(context)
+
+            yield trampoline.freeze
+
+            tos.destroy()
+
+        # use the coroutine above
+        def pause():
+            for i in xrange(pauses):
+                yield wait_y(trampoline, context, 20, i)
+
+            loop.quit()
+
+        # run it
+        trampoline.add(pause())
+        trampoline.run()
+        loop.run()
+
+        assert False not in d
+
+    # 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()
