testsuite/old/test_element.py

   1 # -*- Mode: Python -*-
   2 # vi:si:et:sw=4:sts=4:ts=4
   3 #
   4 # gst-python - Python bindings for GStreamer
   5 # Copyright (C) 2002 David I. Lehn
   6 # Copyright (C) 2004 Johan Dahlin
   7 # Copyright (C) 2005 Edward Hervey
   8 #
   9 # This library is free software; you can redistribute it and/or
  10 # modify it under the terms of the GNU Lesser General Public
  11 # License as published by the Free Software Foundation; either
  12 # version 2.1 of the License, or (at your option) any later version.
  13 #
  14 # This library is distributed in the hope that it will be useful,
  15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17 # Lesser General Public License for more details.
  18 #
  19 # You should have received a copy of the GNU Lesser General Public
  20 # License along with this library; if not, write to the Free Software
  21 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
  22
  23 from common import gst, unittest, TestCase, pygobject_2_13
  24
  25 import sys
  26
  27 # since I can't subclass gst.Element for some reason, I use a bin here
  28 # it don't matter to Jesus
  29 class TestElement(gst.Bin):
  30     def break_it_down(self):
  31         self.debug('Hammer Time')
  32
  33 class ElementTest(TestCase):
  34     name = 'fakesink'
  35     alias = 'sink'
  36
  37     def testGoodConstructor(self):
  38         element = gst.element_factory_make(self.name, self.alias)
  39         assert element is not None, 'element is None'
  40         assert isinstance(element, gst.Element)
  41         assert element.get_name() == self.alias
  42
  43 ## FIXME : Make a new test for state changes, using bus signals
  44
  45 ## class FakeSinkTest(ElementTest):
  46 ##     FAKESINK_STATE_ERROR_NONE           = "0"
  47 ##     FAKESINK_STATE_ERROR_NULL_READY,    = "1"
  48 ##     FAKESINK_STATE_ERROR_READY_PAUSED,  = "2"
  49 ##     FAKESINK_STATE_ERROR_PAUSED_PLAYING = "3"
  50 ##     FAKESINK_STATE_ERROR_PLAYING_PAUSED = "4"
  51 ##     FAKESINK_STATE_ERROR_PAUSED_READY   = "5"
  52 ##     FAKESINK_STATE_ERROR_READY_NULL     = "6"
  53
  54 ##     name = 'fakesink'
  55 ##     alias = 'sink'
  56 ##     def setUp(self):
  57 ##         ElementTest.setUp(self)
  58 ##         self.element = gst.element_factory_make('fakesink', 'sink')
  59
  60 ##     def tearDown(self):
  61 ##         self.element.set_state(gst.STATE_NULL)
  62 ##         del self.element
  63 ##         ElementTest.tearDown(self)
  64
  65 ##     def checkError(self, old_state, state, name):
  66 ##         assert self.element.get_state() == gst.STATE_NULL
  67 ##         assert self.element.set_state(old_state)
  68 ##         assert self.element.get_state() == old_state
  69 ##         self.element.set_property('state-error', name)
  70 ##         self.error = False
  71 ##         def error_cb(element, source, gerror, debug):
  72 ##             assert isinstance(element, gst.Element)
  73 ##             assert element == self.element
  74 ##             assert isinstance(source, gst.Element)
  75 ##             assert source == self.element
  76 ##             assert isinstance(gerror, gst.GError)
  77 ##             self.error = True
  78
  79 ##         self.element.connect('error', error_cb)
  80 ##         self.element.set_state (state)
  81 ##         assert self.error, 'error not set'
  82 ##         #assert error_message.find('ERROR') != -1
  83
  84 ##         self.element.get_state() == old_state, 'state changed'
  85
  86 ##     def testStateErrorNullReady(self):
  87 ##         self.checkError(gst.STATE_NULL, gst.STATE_READY,
  88 ##                         self.FAKESINK_STATE_ERROR_NULL_READY)
  89
  90 ##     def testStateErrorReadyPaused(self):
  91 ##         self.checkError(gst.STATE_READY, gst.STATE_PAUSED,
  92 ##                         self.FAKESINK_STATE_ERROR_READY_PAUSED)
  93
  94 ##     def testStateErrorPausedPlaying(self):
  95 ##         self.checkError(gst.STATE_PAUSED, gst.STATE_PLAYING,
  96 ##                         self.FAKESINK_STATE_ERROR_PAUSED_PLAYING)
  97
  98 ##     def testStateErrorPlayingPaused(self):
  99 ##         self.checkError(gst.STATE_PLAYING, gst.STATE_PAUSED,
 100 ##                         self.FAKESINK_STATE_ERROR_PLAYING_PAUSED)
 101
 102 ##     def testStateErrorPausedReady(self):
 103 ##         self.checkError(gst.STATE_PAUSED, gst.STATE_READY,
 104 ##                         self.FAKESINK_STATE_ERROR_PAUSED_READY)
 105
 106 ##     def testStateErrorReadyNull(self):
 107 ##         self.checkError(gst.STATE_READY, gst.STATE_NULL,
 108 ##                         self.FAKESINK_STATE_ERROR_READY_NULL)
 109
 110 ##     def checkStateChange(self, old, new):
 111 ##         def state_change_cb(element, old_s, new_s):
 112 ##             assert isinstance(element, gst.Element)
 113 ##             assert element == self.element
 114 ##             assert old_s == old
 115 ##             assert new_s == new
 116
 117 ##         assert self.element.set_state(old)
 118 ##         assert self.element.get_state(0.0)[1] == old
 119
 120 ## # FIXME: replace with messages
 121 ## #        self.element.connect('state-change', state_change_cb)
 122
 123 ##         assert self.element.set_state(new)
 124 ##         assert self.element.get_state(0.0)[1] == new
 125
 126 ##     def testStateChangeNullReady(self):
 127 ##         self.checkStateChange(gst.STATE_NULL, gst.STATE_READY)
 128
 129 ##     def testStateChangeReadyPaused(self):
 130 ##         self.checkStateChange(gst.STATE_READY, gst.STATE_PAUSED)
 131
 132 ##     def testStateChangePausedPlaying(self):
 133 ##         self.checkStateChange(gst.STATE_PAUSED, gst.STATE_PLAYING)
 134
 135 ##     def testStateChangePlayingPaused(self):
 136 ##         self.checkStateChange(gst.STATE_PLAYING, gst.STATE_PAUSED)
 137
 138 ##     def testStateChangePausedReady(self):
 139 ##         self.checkStateChange(gst.STATE_PAUSED, gst.STATE_READY)
 140
 141 ##     def testStateChangeReadyNull(self):
 142 ##         self.checkStateChange(gst.STATE_READY, gst.STATE_NULL)
 143
 144 class NonExistentTest(ElementTest):
 145     name = 'this-element-does-not-exist'
 146     alias = 'no-alias'
 147
 148     testGoodConstructor = lambda s: None
 149     testGoodConstructor2 = lambda s: None
 150
 151 class FileSrcTest(ElementTest):
 152     name = 'filesrc'
 153     alias = 'source'
 154
 155 class FileSinkTest(ElementTest):
 156     name = 'filesink'
 157     alias = 'sink'
 158
 159 class ElementName(TestCase):
 160     def testElementStateGetName(self):
 161         get_name = gst.element_state_get_name
 162         for state in ('NULL',
 163                       'READY',
 164                       'PLAYING',
 165                       'PAUSED'):
 166             name = 'STATE_' + state
 167             assert hasattr(gst, name)
 168             attr = getattr(gst, name)
 169             assert get_name(attr) == state
 170
 171         assert get_name(gst.STATE_VOID_PENDING) == 'VOID_PENDING'
 172         assert get_name(-1) == 'UNKNOWN!(-1)'
 173         self.assertRaises(TypeError, get_name, '')
 174
 175 class QueryTest(TestCase):
 176     def setUp(self):
 177         TestCase.setUp(self)
 178         self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
 179         self.assertEquals(self.pipeline.__gstrefcount__, 1)
 180
 181         self.element = self.pipeline.get_by_name('source')
 182         self.assertEquals(self.pipeline.__gstrefcount__, 1)
 183         self.assertEquals(self.element.__gstrefcount__, 2)
 184         self.assertEquals(sys.getrefcount(self.element), pygobject_2_13 and 2 or 3)
 185
 186     def tearDown(self):
 187         del self.pipeline
 188         del self.element
 189         TestCase.tearDown(self)
 190
 191     def testQuery(self):
 192         gst.debug('querying fakesrc in FORMAT_BYTES')
 193         res = self.element.query_position(gst.FORMAT_BYTES)
 194         self.assertEquals(self.pipeline.__gstrefcount__, 1)
 195         self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3)
 196         self.assertEquals(self.element.__gstrefcount__, 2)
 197         self.assertEquals(sys.getrefcount(self.element), pygobject_2_13 and 2 or 3)
 198         assert res
 199         assert res[0] == 0
 200         self.assertRaises(gst.QueryError, self.element.query_position,
 201             gst.FORMAT_TIME)
 202         self.gccollect()
 203
 204 class QueueTest(TestCase):
 205     def testConstruct(self):
 206         queue = gst.element_factory_make('queue')
 207         assert queue.get_name() == 'queue0'
 208         self.assertEquals(queue.__gstrefcount__, 1)
 209
 210 class DebugTest(TestCase):
 211     def testDebug(self):
 212         e = gst.element_factory_make('fakesrc')
 213         e.error('I am an error string')
 214         e.warning('I am a warning string')
 215         e.info('I am an info string')
 216         e.debug('I am a debug string')
 217         e.log('I am a log string')
 218         e.debug('I am a formatted %s %s' % ('log', 'string'))
 219
 220     def testElementDebug(self):
 221         e = TestElement("testelement")
 222         e.set_property("name", "testelement")
 223         e.break_it_down()
 224
 225 class LinkTest(TestCase):
 226     def testLinkNoPads(self):
 227         src = gst.Bin()
 228         sink = gst.Bin()
 229         self.assertRaises(gst.LinkError, src.link, sink)
 230
 231     def testLink(self):
 232         src = gst.element_factory_make('fakesrc')
 233         sink = gst.element_factory_make('fakesink')
 234         self.failUnless(src.link(sink))
 235         # FIXME: this unlink leaks, no idea why
 236         # src.unlink(sink)
 237         # print src.__gstrefcount__
 238
 239     def testLinkPads(self):
 240         src = gst.element_factory_make('fakesrc')
 241         sink = gst.element_factory_make('fakesink')
 242         # print src.__gstrefcount__
 243         self.failUnless(src.link_pads("src", sink, "sink"))
 244         src.unlink_pads("src", sink, "sink")
 245
 246     def testLinkFiltered(self):
 247         # a filtered link uses capsfilter and thus needs a bin
 248         bin = gst.Bin()
 249         src = gst.element_factory_make('fakesrc')
 250         sink = gst.element_factory_make('fakesink')
 251         bin.add(src, sink)
 252         caps = gst.caps_from_string("audio/x-raw-int")
 253
 254         self.failUnless(src.link(sink, caps))
 255
 256         # DANGER WILL.  src is not actually connected to sink, since
 257         # there's a capsfilter in the way.  What a leaky abstraction.
 258         # FIXME
 259         # src.unlink(sink)
 260
 261         # instead, mess with pads directly
 262         pad = src.get_pad('src')
 263         pad.unlink(pad.get_peer())
 264         pad = sink.get_pad('sink')
 265         pad.get_peer().unlink(pad)
 266
 267 if __name__ == "__main__":
 268     unittest.main()