testsuite/common.py: add a common.TestCase class that has methods to track and verify garbage collection of GstObject

Original commit message from CVS:

* testsuite/common.py:
add a common.TestCase class that has methods to track and verify
garbage collection of GstObject
* testsuite/test_pad.py:
use it

ChangeLog

@@ -1,3 +1,11 @@
+2005-09-28  Thomas Vander Stichele  <thomas at apestaart dot org>
+
+	* testsuite/common.py:
+	  add a common.TestCase class that has methods to track and verify
+	  garbage collection of GstObject
+	* testsuite/test_pad.py:
+	  use it
+
 2005-09-28  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* gst/Makefile.am:

testsuite/common.py

@@ -30,6 +30,7 @@ except ImportError:
         pass
 import os
 import sys
+import gc
 import unittest
 
 import pygtk
@@ -109,3 +110,31 @@ def run_silent(function, *args, **kwargs):
    output = enable_stderr()
 
    return output
+
+class TestCase(unittest.TestCase):
+    def gccollect(self):
+        # run the garbage collector
+        gst.debug('garbage collecting')
+        gc.collect()
+        gst.debug('done garbage collecting')
+
+    def gctrack(self):
+        # store all gst objects in the gc in a tracking dict
+        # call before doing any allocation in your test, from setUp
+        gst.debug('tracking gc GstObjects')
+        self.gccollect()
+        self._tracked = {}
+        for c in [gst.Element, gst.Pad]:
+            self._tracked[c] = [o for o in gc.get_objects() if isinstance(o, c)]
+
+    def gcverify(self):
+        # verify no new gst objects got added to the gc
+        # call after doing all cleanup in your test, from tearDown
+        gst.debug('verifying gc GstObjects')
+        new = []
+        for c in [gst.Element, gst.Pad]:
+            objs = [o for o in gc.get_objects() if isinstance(o, c)]
+            new.extend([o for o in objs if o not in self._tracked[c]])
+
+        self.failIf(new, new)
+        del self._tracked

testsuite/test_pad.py

@@ -20,15 +20,17 @@
 # License along with this library; if not, write to the Free Software
 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
 
-from common import gst, unittest
+from common import gst, unittest, TestCase
 import sys
 import gc
 
-class PadTemplateTest(unittest.TestCase):
+class PadTemplateTest(TestCase):
+    def setUp(self):
+        self.gctrack()
+
     def tearDown(self):
-        gst.debug('garbage collecting')
+        self.gccollect()
-        gc.collect()
+        self.gcverify()
-        gst.debug('done garbage collecting')
         
     def testConstructor(self):
         template = gst.PadTemplate("template", gst.PAD_SINK,
@@ -37,27 +39,22 @@ class PadTemplateTest(unittest.TestCase):
         self.assertEquals(sys.getrefcount(template), 3)
         #self.assertEquals(template.__gstrefcount__, 1)
 
-class PadTest(unittest.TestCase):
+class PadTest(TestCase):
+    def setUp(self):
+        self.gctrack()
+
     def tearDown(self):
-        gst.debug('garbage collecting')
+        self.gccollect()
-        gc.collect()
+        self.gcverify()
-        gst.debug('done garbage collecting')
+       
-        #elements = [o for o in gc.get_objects() if isinstance(o, gst.Element)]
-        #self.failIf(elements, elements)
-        #pads = [o for o in gc.get_objects() if isinstance(o, gst.Pad)]
-        #self.failIf(pads, pads)
-        
     def testConstructor(self):
         # first style uses gst_pad_new
         gst.debug('creating pad with name src')
-        print "creating pad with src"
         pad = gst.Pad("src", gst.PAD_SRC)
-        print pad
         self.failUnless(pad)
         self.assertEquals(sys.getrefcount(pad), 3)
         self.assertEquals(pad.__gstrefcount__, 1)
 
-
         gst.debug('creating pad with no name')
         self.failUnless(gst.Pad(None, gst.PAD_SRC))
         self.failUnless(gst.Pad(name=None, direction=gst.PAD_SRC))
@@ -67,20 +64,19 @@ class PadTest(unittest.TestCase):
         # second uses gst_pad_new_from_template
         #template = gst.PadTemplate()
 
-class PadPipelineTest(unittest.TestCase):
+class PadPipelineTest(TestCase):
     def setUp(self):
         self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
         src = self.pipeline.get_by_name('source')
         self.srcpad = src.get_pad('src')
+        self.gctrack()
 
     def tearDown(self):
         del self.pipeline
         del self.srcpad
-        gst.debug('garbage collecting')
+        self.gccollect()
-        gc.collect()
+        self.gcverify()
-        gst.debug('done garbage collecting')
+
-        #elements = [o for o in gc.get_objects() if isinstance(o, gst.Element)]
-        #self.failIf(elements, elements)
         
 # FIXME: now that GstQuery is a miniobject with various _new_ factory
 # functions, we need to figure out a way to deal with them in python
@@ -89,33 +85,61 @@ class PadPipelineTest(unittest.TestCase):
 #        assert self.srcpad.query(gst.QUERY_POSITION, gst.FORMAT_BYTES) == 0
 #        assert self.srcpad.query(gst.QUERY_POSITION, gst.FORMAT_TIME) == 0
 
-class PadProbeTest(unittest.TestCase):
+
+class PadProbeTest(TestCase):
     def setUp(self):
+        self.gctrack()
         self.pipeline = gst.Pipeline()
+        self.assertEquals(self.pipeline.__gstrefcount__, 1)
+        self.assertEquals(sys.getrefcount(self.pipeline), 3)
+
         self.fakesrc = gst.element_factory_make('fakesrc')
         self.fakesink = gst.element_factory_make('fakesink')
+        self.assertEquals(self.fakesrc.__gstrefcount__, 1)
+        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+
         self.pipeline.add_many(self.fakesrc, self.fakesink)
+        self.assertEquals(self.fakesrc.__gstrefcount__, 2) # added
+        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+
         self.fakesrc.link(self.fakesink)
 
+        self.assertEquals(self.pipeline.__gstrefcount__, 1)
+        self.assertEquals(sys.getrefcount(self.pipeline), 3)
+        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
+        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+
     def tearDown(self):
+        self.assertEquals(self.pipeline.__gstrefcount__, 1)
+        self.assertEquals(sys.getrefcount(self.pipeline), 3)
+        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
+        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        gst.debug('deleting pipeline')
         del self.pipeline
+        self.gccollect()
+
+        self.assertEquals(self.fakesrc.__gstrefcount__, 1) # parent gone
+        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
+        gst.debug('deleting fakesrc')
         del self.fakesrc
+        self.gccollect()
         del self.fakesink
-        gst.debug('garbage collecting')
+        self.gccollect()
-        gc.collect()
+
-        gst.debug('done garbage collecting')
+        self.gcverify()
         
     def testFakeSrcProbeOnce(self):
         self.fakesrc.set_property('num-buffers', 1)
 
         pad = self.fakesrc.get_pad('src')
-        pad.add_buffer_probe(self._probe_callback_fakesrc)
+        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
         self._got_fakesrc_buffer = 0
         self.pipeline.set_state(gst.STATE_PLAYING)
         while not self._got_fakesrc_buffer:
             pass
 
         self.pipeline.set_state(gst.STATE_NULL)
+        pad.remove_buffer_probe (id)
 
     def testFakeSrcProbeMany(self):
         self.fakesrc.set_property('num-buffers', 1000)
@@ -129,7 +153,6 @@ class PadProbeTest(unittest.TestCase):
 
         self.pipeline.set_state(gst.STATE_NULL)
 
-
     def _probe_callback_fakesrc(self, pad, buffer):
         self.failUnless(isinstance(pad, gst.Pad))
         self.failUnless(isinstance(buffer, gst.Buffer))
@@ -152,25 +175,50 @@ class PadProbeTest(unittest.TestCase):
         assert m
         assert self._num_times_called == 1
         self.pipeline.set_state(gst.STATE_NULL)
+        # FIXME: having m going out of scope doesn't seem to be enough
+        # to get it gc collected, and it keeps a ref to the pipeline.
+        # Look for a way to not have to do this explicitly
+        del m
+        self.gccollect()
+
+class PadRefCountTest(TestCase):
+    def setUp(self):
+        self.gctrack()
 
-class PadRefCountTest(unittest.TestCase):
     def tearDown(self):
-        gst.debug('garbage collecting')
+        self.gccollect()
-        gc.collect()
+        self.gcverify()
-        gst.debug('done garbage collecting')
         
     def testAddPad(self):
         # add a pad to an element
         e = gst.element_factory_make('fakesrc')
-        gst.debug('creating pad with name mpypad')
+        self.assertEquals(sys.getrefcount(e), 3)
-        pad = gst.Pad("mpypad", gst.PAD_SRC)
+        self.assertEquals(e.__gstrefcount__, 1)
+
+        gst.debug('creating pad with name mypad')
+        pad = gst.Pad("mypad", gst.PAD_SRC)
         self.failUnless(pad)
+        self.assertEquals(sys.getrefcount(pad), 3)
+        self.assertEquals(pad.__gstrefcount__, 1)
+
+        gst.debug('adding pad to element')
         e.add_pad(pad)
-        gst.debug('deleting element')
+        self.assertEquals(sys.getrefcount(e), 3)
-        del e
+        self.assertEquals(e.__gstrefcount__, 1)
-        gst.debug('garbage collecting')
+        self.assertEquals(sys.getrefcount(pad), 3)
+        self.assertEquals(pad.__gstrefcount__, 2) # added to element
+
+        gst.debug('deleting element and collecting')
         gc.collect()
-        gst.debug('done garbage collecting')
+        del e
+        self.assertEquals(gc.collect(), 1) # collected the element
+        self.assertEquals(sys.getrefcount(pad), 3)
+        self.assertEquals(pad.__gstrefcount__, 1) # removed from element
+
+        gst.debug('deleting pad and collecting')
+        del pad
+        self.assertEquals(gc.collect(), 1) # collected the pad
+        gst.debug('going into teardown')
 
 if __name__ == "__main__":
     unittest.main()