diff --git a/ChangeLog b/ChangeLog index 28d508498d..6fb3640059 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,11 @@ +2005-09-28 Thomas Vander Stichele + + * 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 * gst/Makefile.am: diff --git a/testsuite/common.py b/testsuite/common.py index 7a350bab47..4e519f470d 100644 --- a/testsuite/common.py +++ b/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 diff --git a/testsuite/test_pad.py b/testsuite/test_pad.py index 481d8a792e..79bf87794c 100644 --- a/testsuite/test_pad.py +++ b/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') - gc.collect() - gst.debug('done garbage collecting') + self.gccollect() + self.gcverify() 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') - gc.collect() - 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) - + self.gccollect() + self.gcverify() + 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') - gc.collect() - gst.debug('done garbage collecting') - #elements = [o for o in gc.get_objects() if isinstance(o, gst.Element)] - #self.failIf(elements, elements) + self.gccollect() + self.gcverify() + # 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') - gc.collect() - gst.debug('done garbage collecting') + self.gccollect() + + 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') - gc.collect() - gst.debug('done garbage collecting') + self.gccollect() + self.gcverify() def testAddPad(self): # add a pad to an element e = gst.element_factory_make('fakesrc') - gst.debug('creating pad with name mpypad') - pad = gst.Pad("mpypad", gst.PAD_SRC) + self.assertEquals(sys.getrefcount(e), 3) + 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') - del e - gst.debug('garbage collecting') + self.assertEquals(sys.getrefcount(e), 3) + self.assertEquals(e.__gstrefcount__, 1) + 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()