| # -*- Mode: Python -*- | |
| # vi:si:et:sw=4:sts=4:ts=4 | |
| # | |
| # gst-python - Python bindings for GStreamer | |
| # Copyright (C) 2002 David I. Lehn | |
| # Copyright (C) 2004 Johan Dahlin | |
| # Copyright (C) 2005 Edward Hervey | |
| # | |
| # This library is free software; you can redistribute it and/or | |
| # modify it under the terms of the GNU Lesser General Public | |
| # License as published by the Free Software Foundation; either | |
| # version 2.1 of the License, or (at your option) any later version. | |
| # | |
| # This library is distributed in the hope that it will be useful, | |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU | |
| # Lesser General Public License for more details. | |
| # | |
| # You should have received a copy of the GNU Lesser General Public | |
| # License along with this library; if not, write to the Free Software | |
| # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA | |
| from common import gst, unittest, TestCase, pygobject_2_13 | |
| import sys | |
| import gc | |
| import gobject | |
| class SrcBin(gst.Bin): | |
| def prepare(self): | |
| src = gst.element_factory_make('fakesrc') | |
| self.add(src) | |
| pad = src.get_pad("src") | |
| ghostpad = gst.GhostPad("src", pad) | |
| self.add_pad(ghostpad) | |
| gobject.type_register(SrcBin) | |
| class SinkBin(gst.Bin): | |
| def prepare(self): | |
| sink = gst.element_factory_make('fakesink') | |
| self.add(sink) | |
| pad = sink.get_pad("sink") | |
| ghostpad = gst.GhostPad("sink", pad) | |
| self.add_pad(ghostpad) | |
| self.sink = sink | |
| def connect_handoff(self, cb, *args, **kwargs): | |
| self.sink.set_property('signal-handoffs', True) | |
| self.sink.connect('handoff', cb, *args, **kwargs) | |
| gobject.type_register(SinkBin) | |
| class PipeTest(TestCase): | |
| def setUp(self): | |
| gst.info("setUp") | |
| TestCase.setUp(self) | |
| self.pipeline = gst.Pipeline() | |
| self.assertEquals(self.pipeline.__gstrefcount__, 1) | |
| self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3) | |
| self.src = SrcBin() | |
| self.src.prepare() | |
| self.sink = SinkBin() | |
| self.sink.prepare() | |
| self.assertEquals(self.src.__gstrefcount__, 1) | |
| self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3) | |
| self.assertEquals(self.sink.__gstrefcount__, 1) | |
| self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3) | |
| gst.info("end of SetUp") | |
| def tearDown(self): | |
| gst.info("tearDown") | |
| self.assertTrue (self.pipeline.__gstrefcount__ >= 1 and self.pipeline.__gstrefcount__ <= 2) | |
| self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3) | |
| self.assertEquals(self.src.__gstrefcount__, 2) | |
| self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3) | |
| self.assertEquals(self.sink.__gstrefcount__, 2) | |
| self.assertEquals(sys.getrefcount(self.sink), 3) | |
| gst.debug('deleting pipeline') | |
| del self.pipeline | |
| self.gccollect() | |
| self.assertEquals(self.src.__gstrefcount__, 1) # parent gone | |
| self.assertEquals(self.sink.__gstrefcount__, 1) # parent gone | |
| self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3) | |
| self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3) | |
| gst.debug('deleting src') | |
| del self.src | |
| self.gccollect() | |
| gst.debug('deleting sink') | |
| del self.sink | |
| self.gccollect() | |
| TestCase.tearDown(self) | |
| def testBinState(self): | |
| self.pipeline.add(self.src, self.sink) | |
| self.src.link(self.sink) | |
| self.sink.connect_handoff(self._sink_handoff_cb) | |
| self._handoffs = 0 | |
| self.assertTrue(self.pipeline.set_state(gst.STATE_PLAYING) != gst.STATE_CHANGE_FAILURE) | |
| while True: | |
| (ret, cur, pen) = self.pipeline.get_state() | |
| if ret == gst.STATE_CHANGE_SUCCESS and cur == gst.STATE_PLAYING: | |
| break | |
| while self._handoffs < 10: | |
| pass | |
| self.assertEquals(self.pipeline.set_state(gst.STATE_NULL), gst.STATE_CHANGE_SUCCESS) | |
| while True: | |
| (ret, cur, pen) = self.pipeline.get_state() | |
| if ret == gst.STATE_CHANGE_SUCCESS and cur == gst.STATE_NULL: | |
| break | |
| ## def testProbedLink(self): | |
| ## self.pipeline.add(self.src) | |
| ## pad = self.src.get_pad("src") | |
| ## self.sink.connect_handoff(self._sink_handoff_cb) | |
| ## self._handoffs = 0 | |
| ## # FIXME: adding a probe to the ghost pad does not work atm | |
| ## # id = pad.add_buffer_probe(self._src_buffer_probe_cb) | |
| ## realpad = pad.get_target() | |
| ## self._probe_id = realpad.add_buffer_probe(self._src_buffer_probe_cb) | |
| ## self._probed = False | |
| ## while True: | |
| ## (ret, cur, pen) = self.pipeline.get_state() | |
| ## if ret == gst.STATE_CHANGE_SUCCESS and cur == gst.STATE_PLAYING: | |
| ## break | |
| ## while not self._probed: | |
| ## pass | |
| ## while self._handoffs < 10: | |
| ## pass | |
| ## self.pipeline.set_state(gst.STATE_NULL) | |
| ## while True: | |
| ## (ret, cur, pen) = self.pipeline.get_state() | |
| ## if ret == gst.STATE_CHANGE_SUCCESS and cur == gst.STATE_NULL: | |
| ## break | |
| def _src_buffer_probe_cb(self, pad, buffer): | |
| gst.debug("received probe on pad %r" % pad) | |
| self._probed = True | |
| gst.debug('adding sink bin') | |
| self.pipeline.add(self.sink) | |
| # this seems to get rid of the warnings about pushing on an unactivated | |
| # pad | |
| gst.debug('setting sink state') | |
| # FIXME: attempt one: sync to current pending state of bin | |
| (res, cur, pen) = self.pipeline.get_state(timeout=0) | |
| target = pen | |
| if target == gst.STATE_VOID_PENDING: | |
| target = cur | |
| gst.debug("setting sink state to %r" % target) | |
| # FIXME: the following print can cause a lock-up; why ? | |
| # print target | |
| # if we don't set async, it will possibly end up in PAUSED | |
| self.sink.set_state(target) | |
| gst.debug('linking') | |
| self.src.link(self.sink) | |
| gst.debug('removing buffer probe id %r' % self._probe_id) | |
| pad.remove_buffer_probe(self._probe_id) | |
| self._probe_id = None | |
| gst.debug('done') | |
| def _sink_handoff_cb(self, sink, buffer, pad): | |
| gst.debug('received handoff on pad %r' % pad) | |
| self._handoffs += 1 | |
| class TargetTest(TestCase): | |
| def test_target(self): | |
| src = gst.Pad("src", gst.PAD_SRC) | |
| ghost = gst.GhostPad("ghost_src", src) | |
| self.failUnless(ghost.get_target() is src) | |
| ghost.set_target(None) | |
| self.failUnless(ghost.get_target() is None) | |
| ghost.set_target(src) | |
| self.failUnless(ghost.get_target() is src) | |
| if __name__ == "__main__": | |
| unittest.main() |