mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-04-28 16:04:54 +00:00
Code Issues 1 Projects Releases Wiki Activity

55 working tests now; some disabled

Browse source 
Original commit message from CVS:
55 working tests now; some disabled
This commit is contained in:
Thomas Vander Stichele 2005-09-01 15:50:46 +00:00
parent 23b365140e
commit 9ee7ac4ee3
9 changed files with 192 additions and 262 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
ChangeLog
View file

@ -1,3 +1,16 @@
2005-09-01 Thomas Vander Stichele <thomas at apestaart dot org>
* testsuite/Makefile.am:
* testsuite/test_caps.py:
* testsuite/test_element.py:
* testsuite/test_event.py:
* testsuite/test_pad.py:
* testsuite/test_pipeline.py:
* testsuite/test_struct.py:
updated/bits commented out
* testsuite/test_probe.py:
removed
2005-09-01 Thomas Vander Stichele <thomas at apestaart dot org>
* gst/gst-types.defs:

1
testsuite/Makefile.am
View file

@ -25,7 +25,6 @@ tests = \
test_interface.py \
test_pad.py \
test_pipeline.py \
test_probe.py \
test_registry.py \
test_struct.py \
test_xml.py

3
testsuite/test_element.py
View file

@ -101,7 +101,8 @@ class FakeSinkTest(ElementTest):
assert self.element.set_state(old)
assert self.element.get_state() == old
self.element.connect('state-change', state_change_cb)
# FIXME: replace with messages
# self.element.connect('state-change', state_change_cb)
assert self.element.set_state(new)
assert self.element.get_state() == new

116
testsuite/test_event.py
View file

@ -30,64 +30,66 @@ class EventTest(unittest.TestCase):
self.sink = pipeline.get_by_name('sink')
pipeline.set_state(gst.STATE_PLAYING)
## def testEventEmpty(self):
## event = gst.Event(gst.EVENT_EMPTY)
## self.sink.send_event(event)
# def testEventEmpty(self):
# event = gst.Event(gst.EVENT_EMPTY)
# self.sink.send_event(event)
def testEventSeek(self):
event = gst.event_new_seek(gst.SEEK_METHOD_CUR, 0)
assert event
self.sink.send_event(event)
# def testEventSeek(self):
# event = gst.event_new_seek(gst.SEEK_METHOD_CUR, 0)
# assert event
# self.sink.send_event(event)
class EventFileSrcTest(unittest.TestCase):
filename = '/tmp/gst-python-test-file'
def setUp(self):
if os.path.exists(self.filename):
os.remove(self.filename)
open(self.filename, 'w').write(''.join(map(str, range(10))))
self.pipeline = gst.parse_launch('filesrc name=source location=%s blocksize=1 ! fakesink signal-handoffs=1 name=sink' % self.filename)
self.source = self.pipeline.get_by_name('source')
self.sink = self.pipeline.get_by_name('sink')
self.sink.connect('handoff', self.handoff_cb)
self.bus = self.pipeline.get_bus()
self.pipeline.set_state(gst.STATE_PLAYING)
def tearDown(self):
assert self.pipeline.set_state(gst.STATE_PLAYING)
if os.path.exists(self.filename):
os.remove(self.filename)
def handoff_cb(self, element, buffer, pad):
self.handoffs.append(str(buffer))
def playAndIter(self):
self.handoffs = []
assert self.pipeline.set_state(gst.STATE_PLAYING)
while 42:
msg = self.bus.pop()
if msg and msg.type == gst.MESSAGE_EOS:
break
assert self.pipeline.set_state(gst.STATE_PAUSED)
handoffs = self.handoffs
self.handoffs = []
return handoffs
def sink_seek(self, offset, method=gst.SEEK_METHOD_SET):
method |= (gst.SEEK_FLAG_FLUSH | gst.FORMAT_BYTES)
self.source.send_event(gst.event_new_seek(method, offset))
self.source.send_event(gst.Event(gst.EVENT_FLUSH))
self.sink.send_event(gst.event_new_seek(method, offset))
self.sink.send_event(gst.Event(gst.EVENT_FLUSH))
def testSimple(self):
handoffs = self.playAndIter()
assert handoffs == map(str, range(10))
def testSeekCur(self):
self.sink_seek(8)
#print self.playAndIter()
# FIXME: fix these tests
#class EventFileSrcTest(unittest.TestCase):
# # FIXME: properly create temp files
# filename = '/tmp/gst-python-test-file'
# def setUp(self):
# if os.path.exists(self.filename):
# os.remove(self.filename)
# open(self.filename, 'w').write(''.join(map(str, range(10))))
#
# self.pipeline = gst.parse_launch('filesrc name=source location=%s blocksize=1 ! fakesink signal-handoffs=1 name=sink' % self.filename)
# self.source = self.pipeline.get_by_name('source')
# self.sink = self.pipeline.get_by_name('sink')
# self.sink.connect('handoff', self.handoff_cb)
# self.bus = self.pipeline.get_bus()
# self.pipeline.set_state(gst.STATE_PLAYING)
#
# def tearDown(self):
# assert self.pipeline.set_state(gst.STATE_PLAYING)
# if os.path.exists(self.filename):
# os.remove(self.filename)
#
# def handoff_cb(self, element, buffer, pad):
# self.handoffs.append(str(buffer))
#
# def playAndIter(self):
# self.handoffs = []
# assert self.pipeline.set_state(gst.STATE_PLAYING)
# while 42:
# msg = self.bus.pop()
# if msg and msg.type == gst.MESSAGE_EOS:
# break
# assert self.pipeline.set_state(gst.STATE_PAUSED)
# handoffs = self.handoffs
# self.handoffs = []
# return handoffs
#
# def sink_seek(self, offset, method=gst.SEEK_METHOD_SET):
# method |= (gst.SEEK_FLAG_FLUSH | gst.FORMAT_BYTES)
# self.source.send_event(gst.event_new_seek(method, offset))
# self.source.send_event(gst.Event(gst.EVENT_FLUSH))
# self.sink.send_event(gst.event_new_seek(method, offset))
# self.sink.send_event(gst.Event(gst.EVENT_FLUSH))
#
# def testSimple(self):
# handoffs = self.playAndIter()
# assert handoffs == map(str, range(10))
#
# def testSeekCur(self):
# self.sink_seek(8)
#
# #print self.playAndIter()
class TestEmit(unittest.TestCase):
def testEmit(self):
@ -98,7 +100,7 @@ class TestEmit(unittest.TestCase):
testhelper.emit_event(object)
# Then emit from Python
object.emit('event', gst.Event(gst.EVENT_UNKNOWN))
object.emit('event', gst.event_new_eos())
def _event_cb(self, obj, event):
assert isinstance(event, gst.Event)

34
testsuite/test_pad.py
View file

@ -26,13 +26,35 @@ class PadTest(unittest.TestCase):
def setUp(self):
self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
src = self.pipeline.get_by_name('source')
self.sink = src.get_pad('src')
self.srcpad = src.get_pad('src')
def testQuery(self):
assert self.sink.query(gst.QUERY_TOTAL, gst.FORMAT_BYTES) == -1
assert self.sink.query(gst.QUERY_POSITION, gst.FORMAT_BYTES) == 0
assert self.sink.query(gst.QUERY_POSITION, gst.FORMAT_TIME) == 0
# 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
# def testQuery(self):
# assert self.sink.query(gst.QUERY_TOTAL, gst.FORMAT_BYTES) == -1
# 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):
def testFakeSrcProbe(self):
pipeline = gst.Pipeline()
fakesrc = gst.element_factory_make('fakesrc')
fakesrc.set_property('num-buffers', 1)
fakesink = gst.element_factory_make('fakesink')
pipeline.add_many(fakesrc, fakesink)
fakesrc.link(fakesink)
pad = fakesrc.get_pad('src')
pad.add_buffer_probe(self._probe_callback_fakesrc)
self._got_fakesrc_buffer = False
pipeline.set_state(gst.STATE_PLAYING)
while not self._got_fakesrc_buffer:
pass
def _probe_callback_fakesrc(self, pad, buffer):
self.failUnless(isinstance(pad, gst.Pad))
self.failUnless(isinstance(buffer, gst.Buffer))
self._got_fakesrc_buffer = True
if __name__ == "__main__":
unittest.main()

13
testsuite/test_pipeline.py
View file

@ -20,6 +20,8 @@
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import time
from common import gst, unittest
class PipelineConstructor(unittest.TestCase):
@ -30,12 +32,6 @@ class PipelineConstructor(unittest.TestCase):
assert isinstance(pipeline, gst.Pipeline), 'pipeline is not a GstPipline'
assert pipeline.get_name() == name, 'pipelines name is wrong'
## class ThreadConstructor(unittest.TestCase):
## def testCreate(self):
## thread = gst.Thread('test-thread')
## assert thread is not None, 'thread is None'
## assert isinstance(thread, gst.Thread)
class Pipeline(unittest.TestCase):
def setUp(self):
self.pipeline = gst.Pipeline('test-pipeline')
@ -50,10 +46,9 @@ class Pipeline(unittest.TestCase):
self.pipeline.set_state(gst.STATE_PLAYING)
self.assertEqual(self.pipeline.get_state(), gst.STATE_PLAYING)
while self.pipeline.iterate():
pass
time.sleep(1)
self.assertEqual(self.pipeline.get_state(), gst.STATE_PAUSED)
self.assertEqual(self.pipeline.get_state(), gst.STATE_PLAYING)
self.pipeline.set_state(gst.STATE_NULL)
self.assertEqual(self.pipeline.get_state(), gst.STATE_NULL)

102
testsuite/test_probe.py
View file

@ -1,102 +0,0 @@
# -*- Mode: Python; test-case-name: testsuite.test_probe -*-
# 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import sys
from common import gst, unittest
class ProbeTest(unittest.TestCase):
def testWrongNumber(self):
self.assertRaises(TypeError, gst.Probe, True)
def testWrongType(self):
# bool is int type
self.assertRaises(TypeError, gst.Probe, "noint", lambda x: "x")
# second arg should be callable
self.assertRaises(TypeError, gst.Probe, True, "nocallable")
def testPerformNoData(self):
probe = gst.Probe(True, self._probe_callback, "yeeha")
self.assertRaises(TypeError, probe.perform, None)
self.assertRaises(TypeError, probe.perform, "nodata")
def testPerformNoArg(self):
probe = gst.Probe(True, self._probe_callback_no_arg)
buffer = gst.Buffer()
probe.perform(buffer)
self.assertEqual(self._no_arg, None)
def _probe_callback_no_arg(self, probe, data):
self._no_arg = None
def testPerformOneArg(self):
probe = gst.Probe(True, self._probe_callback, "yeeha")
buffer = gst.Buffer()
probe.perform(buffer)
self.assertEqual(self._probe_result, "yeeha")
def _probe_callback(self, probe, data, result):
self._probe_result = result
return True
def testPerformTwoArgs(self):
probe = gst.Probe(True, self._probe_callback_two, "yeeha", "works")
buffer = gst.Buffer()
probe.perform(buffer)
self.assertEqual(self._probe_result1, "yeeha")
self.assertEqual(self._probe_result2, "works")
def _probe_callback_two(self, probe, data, result1, result2):
self._probe_result1 = result1
self._probe_result2 = result2
return True
# this test checks if the probe can replace the probed GstData with
# another, FIXME: use return values on probe callback for this
def notestPerformChangeBuffer(self):
probe = gst.Probe(True, self._probe_callback_change_buffer)
buffer = gst.Buffer('changeme')
probe.perform(buffer)
self.assertEqual(str(buffer), 'changed')
def _probe_callback_change_buffer(self, probe, data):
data = gst.Buffer('changed')
def testFakeSrcProbe(self):
pipeline = gst.Pipeline()
fakesrc = gst.element_factory_make('fakesrc')
fakesrc.set_property('num-buffers', 1)
fakesink = gst.element_factory_make('fakesink')
pipeline.add_many(fakesrc, fakesink)
fakesrc.link(fakesink)
pad = fakesrc.get_pad('src')
probe = gst.Probe(True, self._probe_callback_fakesrc)
pad.add_probe(probe)
pipeline.set_state(gst.STATE_PLAYING)
while pipeline.iterate(): pass
self.assertEqual(self._got_fakesrc_buffer, True)
def _probe_callback_fakesrc(self, probe, data):
self._got_fakesrc_buffer = True
if __name__ == "__main__":
unittest.main()