mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-06-03 05:59:44 +00:00
269 lines
9.7 KiB
Python
269 lines
9.7 KiB
Python
# -*- 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
|
|
|
|
# since I can't subclass gst.Element for some reason, I use a bin here
|
|
# it don't matter to Jesus
|
|
class TestElement(gst.Bin):
|
|
def break_it_down(self):
|
|
self.debug('Hammer Time')
|
|
|
|
class ElementTest(TestCase):
|
|
name = 'fakesink'
|
|
alias = 'sink'
|
|
|
|
def testGoodConstructor(self):
|
|
element = gst.element_factory_make(self.name, self.alias)
|
|
assert element is not None, 'element is None'
|
|
assert isinstance(element, gst.Element)
|
|
assert element.get_name() == self.alias
|
|
|
|
## FIXME : Make a new test for state changes, using bus signals
|
|
|
|
## class FakeSinkTest(ElementTest):
|
|
## FAKESINK_STATE_ERROR_NONE = "0"
|
|
## FAKESINK_STATE_ERROR_NULL_READY, = "1"
|
|
## FAKESINK_STATE_ERROR_READY_PAUSED, = "2"
|
|
## FAKESINK_STATE_ERROR_PAUSED_PLAYING = "3"
|
|
## FAKESINK_STATE_ERROR_PLAYING_PAUSED = "4"
|
|
## FAKESINK_STATE_ERROR_PAUSED_READY = "5"
|
|
## FAKESINK_STATE_ERROR_READY_NULL = "6"
|
|
|
|
## name = 'fakesink'
|
|
## alias = 'sink'
|
|
## def setUp(self):
|
|
## ElementTest.setUp(self)
|
|
## self.element = gst.element_factory_make('fakesink', 'sink')
|
|
|
|
## def tearDown(self):
|
|
## self.element.set_state(gst.STATE_NULL)
|
|
## del self.element
|
|
## ElementTest.tearDown(self)
|
|
|
|
## def checkError(self, old_state, state, name):
|
|
## assert self.element.get_state() == gst.STATE_NULL
|
|
## assert self.element.set_state(old_state)
|
|
## assert self.element.get_state() == old_state
|
|
## self.element.set_property('state-error', name)
|
|
## self.error = False
|
|
## def error_cb(element, source, gerror, debug):
|
|
## assert isinstance(element, gst.Element)
|
|
## assert element == self.element
|
|
## assert isinstance(source, gst.Element)
|
|
## assert source == self.element
|
|
## assert isinstance(gerror, gst.GError)
|
|
## self.error = True
|
|
|
|
## self.element.connect('error', error_cb)
|
|
## self.element.set_state (state)
|
|
## assert self.error, 'error not set'
|
|
## #assert error_message.find('ERROR') != -1
|
|
|
|
## self.element.get_state() == old_state, 'state changed'
|
|
|
|
## def testStateErrorNullReady(self):
|
|
## self.checkError(gst.STATE_NULL, gst.STATE_READY,
|
|
## self.FAKESINK_STATE_ERROR_NULL_READY)
|
|
|
|
## def testStateErrorReadyPaused(self):
|
|
## self.checkError(gst.STATE_READY, gst.STATE_PAUSED,
|
|
## self.FAKESINK_STATE_ERROR_READY_PAUSED)
|
|
|
|
## def testStateErrorPausedPlaying(self):
|
|
## self.checkError(gst.STATE_PAUSED, gst.STATE_PLAYING,
|
|
## self.FAKESINK_STATE_ERROR_PAUSED_PLAYING)
|
|
|
|
## def testStateErrorPlayingPaused(self):
|
|
## self.checkError(gst.STATE_PLAYING, gst.STATE_PAUSED,
|
|
## self.FAKESINK_STATE_ERROR_PLAYING_PAUSED)
|
|
|
|
## def testStateErrorPausedReady(self):
|
|
## self.checkError(gst.STATE_PAUSED, gst.STATE_READY,
|
|
## self.FAKESINK_STATE_ERROR_PAUSED_READY)
|
|
|
|
## def testStateErrorReadyNull(self):
|
|
## self.checkError(gst.STATE_READY, gst.STATE_NULL,
|
|
## self.FAKESINK_STATE_ERROR_READY_NULL)
|
|
|
|
## def checkStateChange(self, old, new):
|
|
## def state_change_cb(element, old_s, new_s):
|
|
## assert isinstance(element, gst.Element)
|
|
## assert element == self.element
|
|
## assert old_s == old
|
|
## assert new_s == new
|
|
|
|
## assert self.element.set_state(old)
|
|
## assert self.element.get_state(0.0)[1] == old
|
|
|
|
## # FIXME: replace with messages
|
|
## # self.element.connect('state-change', state_change_cb)
|
|
|
|
## assert self.element.set_state(new)
|
|
## assert self.element.get_state(0.0)[1] == new
|
|
|
|
## def testStateChangeNullReady(self):
|
|
## self.checkStateChange(gst.STATE_NULL, gst.STATE_READY)
|
|
|
|
## def testStateChangeReadyPaused(self):
|
|
## self.checkStateChange(gst.STATE_READY, gst.STATE_PAUSED)
|
|
|
|
## def testStateChangePausedPlaying(self):
|
|
## self.checkStateChange(gst.STATE_PAUSED, gst.STATE_PLAYING)
|
|
|
|
## def testStateChangePlayingPaused(self):
|
|
## self.checkStateChange(gst.STATE_PLAYING, gst.STATE_PAUSED)
|
|
|
|
## def testStateChangePausedReady(self):
|
|
## self.checkStateChange(gst.STATE_PAUSED, gst.STATE_READY)
|
|
|
|
## def testStateChangeReadyNull(self):
|
|
## self.checkStateChange(gst.STATE_READY, gst.STATE_NULL)
|
|
|
|
class NonExistentTest(ElementTest):
|
|
name = 'this-element-does-not-exist'
|
|
alias = 'no-alias'
|
|
|
|
testGoodConstructor = lambda s: None
|
|
testGoodConstructor2 = lambda s: None
|
|
|
|
class FileSrcTest(ElementTest):
|
|
name = 'filesrc'
|
|
alias = 'source'
|
|
|
|
class FileSinkTest(ElementTest):
|
|
name = 'filesink'
|
|
alias = 'sink'
|
|
|
|
class ElementName(TestCase):
|
|
def testElementStateGetName(self):
|
|
get_name = gst.element_state_get_name
|
|
for state in ('NULL',
|
|
'READY',
|
|
'PLAYING',
|
|
'PAUSED'):
|
|
name = 'STATE_' + state
|
|
assert hasattr(gst, name)
|
|
attr = getattr(gst, name)
|
|
assert get_name(attr) == state
|
|
|
|
assert get_name(gst.STATE_VOID_PENDING) == 'VOID_PENDING'
|
|
assert get_name(-1) == 'UNKNOWN!(-1)'
|
|
self.assertRaises(TypeError, get_name, '')
|
|
|
|
class QueryTest(TestCase):
|
|
def setUp(self):
|
|
TestCase.setUp(self)
|
|
self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
|
|
self.assertEquals(self.pipeline.__gstrefcount__, 1)
|
|
|
|
self.element = self.pipeline.get_by_name('source')
|
|
self.assertEquals(self.pipeline.__gstrefcount__, 1)
|
|
self.assertEquals(self.element.__gstrefcount__, 2)
|
|
self.assertEquals(sys.getrefcount(self.element), pygobject_2_13 and 2 or 3)
|
|
|
|
def tearDown(self):
|
|
del self.pipeline
|
|
del self.element
|
|
TestCase.tearDown(self)
|
|
|
|
def testQuery(self):
|
|
gst.debug('querying fakesrc in FORMAT_BYTES')
|
|
res = self.element.query_position(gst.FORMAT_BYTES)
|
|
self.assertEquals(self.pipeline.__gstrefcount__, 1)
|
|
self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3)
|
|
self.assertEquals(self.element.__gstrefcount__, 2)
|
|
self.assertEquals(sys.getrefcount(self.element), pygobject_2_13 and 2 or 3)
|
|
assert res
|
|
assert res[0] == 0
|
|
self.assertRaises(gst.QueryError, self.element.query_position,
|
|
gst.FORMAT_TIME)
|
|
self.gccollect()
|
|
|
|
class QueueTest(TestCase):
|
|
def testConstruct(self):
|
|
queue = gst.element_factory_make('queue')
|
|
assert queue.get_name() == 'queue0'
|
|
self.assertEquals(queue.__gstrefcount__, 1)
|
|
|
|
class DebugTest(TestCase):
|
|
def testDebug(self):
|
|
e = gst.element_factory_make('fakesrc')
|
|
e.error('I am an error string')
|
|
e.warning('I am a warning string')
|
|
e.info('I am an info string')
|
|
e.debug('I am a debug string')
|
|
e.log('I am a log string')
|
|
e.debug('I am a formatted %s %s' % ('log', 'string'))
|
|
|
|
def testElementDebug(self):
|
|
e = TestElement("testelement")
|
|
e.set_property("name", "testelement")
|
|
e.break_it_down()
|
|
|
|
class LinkTest(TestCase):
|
|
def testLinkNoPads(self):
|
|
src = gst.Bin()
|
|
sink = gst.Bin()
|
|
self.assertRaises(gst.LinkError, src.link, sink)
|
|
|
|
def testLink(self):
|
|
src = gst.element_factory_make('fakesrc')
|
|
sink = gst.element_factory_make('fakesink')
|
|
self.failUnless(src.link(sink))
|
|
# FIXME: this unlink leaks, no idea why
|
|
# src.unlink(sink)
|
|
# print src.__gstrefcount__
|
|
|
|
def testLinkPads(self):
|
|
src = gst.element_factory_make('fakesrc')
|
|
sink = gst.element_factory_make('fakesink')
|
|
# print src.__gstrefcount__
|
|
self.failUnless(src.link_pads("src", sink, "sink"))
|
|
src.unlink_pads("src", sink, "sink")
|
|
|
|
def testLinkFiltered(self):
|
|
# a filtered link uses capsfilter and thus needs a bin
|
|
bin = gst.Bin()
|
|
src = gst.element_factory_make('fakesrc')
|
|
sink = gst.element_factory_make('fakesink')
|
|
bin.add(src, sink)
|
|
caps = gst.caps_from_string("audio/x-raw-int")
|
|
|
|
self.failUnless(src.link(sink, caps))
|
|
|
|
# DANGER WILL. src is not actually connected to sink, since
|
|
# there's a capsfilter in the way. What a leaky abstraction.
|
|
# FIXME
|
|
# src.unlink(sink)
|
|
|
|
# instead, mess with pads directly
|
|
pad = src.get_pad('src')
|
|
pad.unlink(pad.get_peer())
|
|
pad = sink.get_pad('sink')
|
|
pad.get_peer().unlink(pad)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|