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:
* gst/gstbuffer.override:
* gst/pygstminiobject.c: (pygstminiobject_new_noref):
test_interface.py \
test_pad.py \
test_pipeline.py \
- test_probe.py \
test_registry.py \
test_struct.py \
test_xml.py
class CapsTest(unittest.TestCase):
def setUp(self):
- self.caps = gst.caps_from_string('video/x-raw-yuv,width=10,framerate=5.0;video/x-raw-rgb,width=15,framerate=10.0')
- self.structure = self.caps[0]
- self.any = gst.Caps("ANY")
- self.empty = gst.Caps()
+ self.caps = gst.caps_from_string('video/x-raw-yuv,width=10,framerate=5.0;video/x-raw-rgb,width=15,framerate=10.0')
+ self.structure = self.caps[0]
+ self.any = gst.Caps("ANY")
+ self.empty = gst.Caps()
def testCapsMime(self):
- mime = self.structure.get_name()
- assert mime == 'video/x-raw-yuv'
+ mime = self.structure.get_name()
+ assert mime == 'video/x-raw-yuv'
def testCapsList(self):
- 'check if we can access Caps as a list'
- structure = self.caps[0]
- mime = structure.get_name()
- assert mime == 'video/x-raw-yuv'
- structure = self.caps[1]
- mime = structure.get_name()
- assert mime == 'video/x-raw-rgb'
+ 'check if we can access Caps as a list'
+ structure = self.caps[0]
+ mime = structure.get_name()
+ assert mime == 'video/x-raw-yuv'
+ structure = self.caps[1]
+ mime = structure.get_name()
+ assert mime == 'video/x-raw-rgb'
def testCapsConstructEmpty(self):
caps = gst.Caps()
- assert isinstance(caps, gst.Caps)
+ assert isinstance(caps, gst.Caps)
def testCapsConstructFromString(self):
caps = gst.Caps('video/x-raw-yuv,width=10')
- assert isinstance(caps, gst.Caps)
+ assert isinstance(caps, gst.Caps)
assert len(caps) == 1
- assert isinstance(caps[0], gst.Structure)
+ assert isinstance(caps[0], gst.Structure)
assert caps[0].get_name() == 'video/x-raw-yuv'
- assert isinstance(caps[0]['width'], int)
+ assert isinstance(caps[0]['width'], int)
assert caps[0]['width'] == 10
def testCapsConstructFromStructure(self):
struct = gst.structure_from_string('video/x-raw-yuv,width=10')
caps = gst.Caps(struct)
- assert isinstance(caps, gst.Caps)
+ assert isinstance(caps, gst.Caps)
assert len(caps) == 1
- assert isinstance(caps[0], gst.Structure)
+ assert isinstance(caps[0], gst.Structure)
assert caps[0].get_name() == 'video/x-raw-yuv'
- assert isinstance(caps[0]['width'], int)
+ assert isinstance(caps[0]['width'], int)
assert caps[0]['width'] == 10
def testCapsConstructFromStructures(self):
struct1 = gst.structure_from_string('video/x-raw-yuv,width=10')
struct2 = gst.structure_from_string('video/x-raw-rgb,height=20.0')
caps = gst.Caps(struct1, struct2)
- assert isinstance(caps, gst.Caps)
+ assert isinstance(caps, gst.Caps)
assert len(caps) == 2
struct = caps[0]
- assert isinstance(struct, gst.Structure), struct
+ assert isinstance(struct, gst.Structure), struct
assert struct.get_name() == 'video/x-raw-yuv', struct.get_name()
assert struct.has_key('width')
- assert isinstance(struct['width'], int)
+ assert isinstance(struct['width'], int)
assert struct['width'] == 10
struct = caps[1]
- assert isinstance(struct, gst.Structure), struct
+ assert isinstance(struct, gst.Structure), struct
assert struct.get_name() == 'video/x-raw-rgb', struct.get_name()
assert struct.has_key('height')
- assert isinstance(struct['height'], float)
+ assert isinstance(struct['height'], float)
assert struct['height'] == 20.0
def testCapsRefernceStructs(self):
'test that shows why it\'s not a good idea to use structures by reference'
- caps = gst.Caps('hi/mom,width=0')
- structure = caps[0]
- del caps
- assert structure['width'] == 0
-
+ caps = gst.Caps('hi/mom,width=0')
+ structure = caps[0]
+ del caps
+ assert structure['width'] == 0
+
def testCapsStructureChange(self):
- 'test if changing the structure of the caps works by reference'
- assert self.structure['width'] == 10
+ 'test if changing the structure of the caps works by reference'
+ assert self.structure['width'] == 10
self.structure['width'] = 5
- assert self.structure['width'] == 5.0
- # check if we changed the caps as well
- structure = self.caps[0]
- assert structure['width'] == 5.0
+ assert self.structure['width'] == 5.0
+ # check if we changed the caps as well
+ structure = self.caps[0]
+ assert structure['width'] == 5.0
def testCapsBadConstructor(self):
struct = gst.structure_from_string('video/x-raw-yuv,width=10')
def testTrueFalse(self):
'test that comparisons using caps work the intended way'
- assert self.any # not empty even though it has no structures
- assert not self.empty
- assert not gst.Caps('EMPTY') # also empty
- assert gst.Caps('your/mom')
+ assert self.any # not empty even though it has no structures
+ assert not self.empty
+ assert not gst.Caps('EMPTY') # also empty
+ assert gst.Caps('your/mom')
def testComparisons(self):
- assert self.empty < self.any
- assert self.empty < self.structure
- assert self.empty < self.caps
- assert self.caps < self.any
- assert self.empty <= self.empty
- assert self.caps <= self.caps
- assert self.caps <= self.any
- assert self.empty == "EMPTY"
- assert self.caps != self.any
- assert self.empty != self.any
- assert self.any > self.empty
- assert self.any >= self.empty
+ assert self.empty < self.any
+ assert self.empty < self.structure
+ assert self.empty < self.caps
+ assert self.caps < self.any
+ assert self.empty <= self.empty
+ assert self.caps <= self.caps
+ assert self.caps <= self.any
+ assert self.empty == "EMPTY"
+ assert self.caps != self.any
+ assert self.empty != self.any
+ assert self.any > self.empty
+ assert self.any >= self.empty
def testFilters(self):
- name = 'video/x-raw-yuv'
- filtercaps = gst.Caps(*[struct for struct in self.caps if struct.get_name() == name])
- intersection = self.caps & 'video/x-raw-yuv'
- assert filtercaps == intersection
+ name = 'video/x-raw-yuv'
+ filtercaps = gst.Caps(*[struct for struct in self.caps if struct.get_name() == name])
+ intersection = self.caps & 'video/x-raw-yuv'
+ assert filtercaps == intersection
def doSubtract(self, set, subset):
- '''mimic the test in GStreamer core's testsuite/caps/subtract.c'''
- assert not set - set
- assert not subset - subset
- assert not subset - set
- test = set - subset
- assert test
- test2 = test | subset
- test = test2 - set
- assert not test
- #our own extensions foolow here
- assert subset == set & subset
- assert set == set | subset
- assert set - subset == set ^ subset
+ '''mimic the test in GStreamer core's testsuite/caps/subtract.c'''
+ assert not set - set
+ assert not subset - subset
+ assert not subset - set
+ test = set - subset
+ assert test
+ test2 = test | subset
+ test = test2 - set
+ assert not test
+ #our own extensions foolow here
+ assert subset == set & subset
+ assert set == set | subset
+ assert set - subset == set ^ subset
def testSubtract(self):
- self.doSubtract(
- gst.Caps ("some/mime, _int = [ 1, 2 ], list = { \"A\", \"B\", \"C\" }"),
- gst.Caps ("some/mime, _int = 1, list = \"A\""))
- self.doSubtract(
- gst.Caps ("some/mime, _double = (double) 1.0; other/mime, _int = { 1, 2 }"),
- gst.Caps ("some/mime, _double = (double) 1.0"))
+ self.doSubtract(
+ gst.Caps ("some/mime, _int = [ 1, 2 ], list = { \"A\", \"B\", \"C\" }"),
+ gst.Caps ("some/mime, _int = 1, list = \"A\""))
+ self.doSubtract(
+ gst.Caps ("some/mime, _double = (double) 1.0; other/mime, _int = { 1, 2 }"),
+ gst.Caps ("some/mime, _double = (double) 1.0"))
if __name__ == "__main__":
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
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
+# 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()
- 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):
object = testhelper.get_object()
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)
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()
-
# 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):
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')
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)
+++ /dev/null
-# -*- 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()
def testInt(self):
assert self.struct.has_key('width')
assert isinstance(self.struct['width'], int)
- assert self.struct['width'] == 10, self.struct['width']
- self.struct['width'] = 5
+ assert self.struct['width'] == 10, self.struct['width']
+ self.struct['width'] = 5
assert self.struct.has_key('width')
assert isinstance(self.struct['width'], int)
- assert self.struct['width'] == 5, self.struct['width']
+ assert self.struct['width'] == 5, self.struct['width']
def testString(self):
assert self.struct.has_key('foo')
assert isinstance(self.struct['foo'], str)
- assert self.struct['foo'] == 'bar', self.struct['foo']
- self.struct['foo'] = 'baz'
+ assert self.struct['foo'] == 'bar', self.struct['foo']
+ self.struct['foo'] = 'baz'
assert self.struct.has_key('foo')
assert isinstance(self.struct['foo'], str)
- assert self.struct['foo'] == 'baz', self.struct['foo']
+ assert self.struct['foo'] == 'baz', self.struct['foo']
def testCreateInt(self):
- self.struct['integer'] = 5
+ self.struct['integer'] = 5
assert self.struct.has_key('integer')
assert isinstance(self.struct['integer'], int)
- assert self.struct['integer'] == 5, self.struct['integer']
+ assert self.struct['integer'] == 5, self.struct['integer']
def testGstValue(self):
s = self.struct
assert s['rlist'] == [([(['a', 'b'], ['c', 'd']),'e'], ['f', 'g']), 'h']
def testStructureChange(self):
- assert self.struct['framerate'] == 5.0
- self.struct['framerate'] = 10.0
- assert self.struct['framerate'] == 10.0
+ assert self.struct['framerate'] == 5.0
+ self.struct['framerate'] = 10.0
+ assert self.struct['framerate'] == 10.0
self.struct['pixel-aspect-ratio'] = gst.Fraction(4, 2)
assert self.struct['pixel-aspect-ratio'].num == 2
assert self.struct['pixel-aspect-ratio'].denom == 1
projects / platform / upstream / gstreamer.git / commitdiff