← Back to branch summary

~ubuntu-branches/ubuntu/precise/gst0.10-python/precise

« back to all changes in this revision

Viewing changes to testsuite/test_pad.py

  • Committer: Bazaar Package Importer
  • Author(s): Loic Minier
  • Date: 2006-06-25 19:37:45 UTC
  • Revision ID: james.westby@ubuntu.com-20060625193745-9yeg0wq56r24n57x
Tags: upstream-0.10.4
Import upstream version 0.10.4

Show diffs side-by-side

added added

removed removed

Lines of Context:
testsuite/test_pad.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., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
 
22
 
 
23
from common import gst, unittest, TestCase
 
24
 
 
25
import sys
 
26
import time
 
27
 
 
28
class PadTemplateTest(TestCase):
 
29
    def testConstructor(self):
 
30
        template = gst.PadTemplate("template", gst.PAD_SINK,
 
31
            gst.PAD_ALWAYS, gst.caps_from_string("audio/x-raw-int"))
 
32
        self.failUnless(template)
 
33
        self.assertEquals(sys.getrefcount(template), 3)
 
34
        #self.assertEquals(template.__gstrefcount__, 1)
 
35
 
 
36
class PadPushUnlinkedTest(TestCase):
 
37
    def setUp(self):
 
38
        TestCase.setUp(self)
 
39
        self.src = gst.Pad("src", gst.PAD_SRC)
 
40
        self.sink = gst.Pad("sink", gst.PAD_SINK)
 
41
 
 
42
    def tearDown(self):
 
43
        self.assertEquals(sys.getrefcount(self.src), 3)
 
44
        self.assertEquals(self.src.__gstrefcount__, 1)
 
45
        del self.src
 
46
        self.assertEquals(sys.getrefcount(self.sink), 3)
 
47
        self.assertEquals(self.sink.__gstrefcount__, 1)
 
48
        del self.sink
 
49
        TestCase.tearDown(self)
 
50
 
 
51
    def testNoProbe(self):
 
52
        self.buffer = gst.Buffer()
 
53
        self.assertEquals(self.buffer.__grefcount__, 1)
 
54
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
 
55
        # pushing it takes a ref in the python wrapper to keep buffer
 
56
        # alive afterwards; but the core unrefs the ref it receives
 
57
        self.assertEquals(self.buffer.__grefcount__, 1)
 
58
 
 
59
    def testFalseProbe(self):
 
60
        id = self.src.add_buffer_probe(self._probe_handler, False)
 
61
        self.buffer = gst.Buffer()
 
62
        self.assertEquals(self.buffer.__grefcount__, 1)
 
63
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
 
64
        self.assertEquals(self.buffer.__grefcount__, 1)
 
65
        self.src.remove_buffer_probe(id)
 
66
 
 
67
    def testTrueProbe(self):
 
68
        id = self.src.add_buffer_probe(self._probe_handler, True)
 
69
        self.buffer = gst.Buffer()
 
70
        self.assertEquals(self.buffer.__grefcount__, 1)
 
71
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
 
72
        self.assertEquals(self.buffer.__grefcount__, 1)
 
73
        self.src.remove_buffer_probe(id)
 
74
 
 
75
    def _probe_handler(self, pad, buffer, ret):
 
76
        return ret
 
77
 
 
78
class PadPushLinkedTest(TestCase):
 
79
    def setUp(self):
 
80
        TestCase.setUp(self)
 
81
        self.src = gst.Pad("src", gst.PAD_SRC)
 
82
        self.sink = gst.Pad("sink", gst.PAD_SINK)
 
83
        caps = gst.caps_from_string("foo/bar")
 
84
        self.src.set_caps(caps)
 
85
        self.sink.set_caps(caps)
 
86
        self.sink.set_chain_function(self._chain_func)
 
87
        self.src.link(self.sink)
 
88
        self.buffers = []
 
89
 
 
90
    def tearDown(self):
 
91
        self.assertEquals(sys.getrefcount(self.src), 3)
 
92
        self.assertEquals(self.src.__gstrefcount__, 1)
 
93
        del self.src
 
94
        self.assertEquals(sys.getrefcount(self.sink), 3)
 
95
        self.assertEquals(self.sink.__gstrefcount__, 1)
 
96
        del self.sink
 
97
        TestCase.tearDown(self)
 
98
 
 
99
    def _chain_func(self, pad, buffer):
 
100
        gst.debug('got buffer %r, id %x, with GMO rc %d'% (
 
101
            buffer, id(buffer), buffer.__grefcount__))
 
102
        self.buffers.append(buffer)
 
103
 
 
104
        return gst.FLOW_OK
 
105
 
 
106
    def testNoProbe(self):
 
107
        self.buffer = gst.Buffer()
 
108
        gst.debug('created new buffer %r, id %x' % (
 
109
            self.buffer, id(self.buffer)))
 
110
        self.assertEquals(self.buffer.__grefcount__, 1)
 
111
        gst.debug('pushing buffer on linked pad, no probe')
 
112
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
 
113
        gst.debug('pushed buffer on linked pad, no probe')
 
114
        # one refcount is held by our scope, another is held on
 
115
        # self.buffers through _chain_func
 
116
        self.assertEquals(self.buffer.__grefcount__, 2)
 
117
        self.assertEquals(len(self.buffers), 1)
 
118
        self.buffers = None
 
119
        self.assertEquals(self.buffer.__grefcount__, 1)
 
120
 
 
121
    def testFalseProbe(self):
 
122
        id = self.src.add_buffer_probe(self._probe_handler, False)
 
123
        self.buffer = gst.Buffer()
 
124
        self.assertEquals(self.buffer.__grefcount__, 1)
 
125
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
 
126
        self.assertEquals(self.buffer.__grefcount__, 1)
 
127
        self.src.remove_buffer_probe(id)
 
128
        self.assertEquals(len(self.buffers), 0)
 
129
 
 
130
    def testTrueProbe(self):
 
131
        probe_id = self.src.add_buffer_probe(self._probe_handler, True)
 
132
        self.buffer = gst.Buffer()
 
133
        self.assertEquals(self.buffer.__grefcount__, 1)
 
134
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
 
135
        # one refcount is held by our scope, another is held on
 
136
        # self.buffers through _chain_func
 
137
        self.assertEquals(self.buffer.__grefcount__, 2)
 
138
 
 
139
        # they are not the same Python object ...
 
140
        self.failIf(self.buffer is self.buffers[0])
 
141
        self.failIf(id(self.buffer) == id(self.buffers[0]))
 
142
        # ... but they wrap the same GstBuffer
 
143
        self.failUnless(self.buffer == self.buffers[0])
 
144
        self.assertEquals(repr(self.buffer), repr(self.buffers[0]))
 
145
        
 
146
        self.src.remove_buffer_probe(probe_id)
 
147
        self.assertEquals(len(self.buffers), 1)
 
148
        self.buffers = None
 
149
        self.assertEquals(self.buffer.__grefcount__, 1)
 
150
 
 
151
    def _probe_handler(self, pad, buffer, ret):
 
152
        return ret
 
153
 
 
154
# a test to show that we can link a pad from the probe handler
 
155
 
 
156
class PadPushProbeLinkTest(TestCase):
 
157
    def setUp(self):
 
158
        TestCase.setUp(self)
 
159
        self.src = gst.Pad("src", gst.PAD_SRC)
 
160
        self.sink = gst.Pad("sink", gst.PAD_SINK)
 
161
        caps = gst.caps_from_string("foo/bar")
 
162
        self.src.set_caps(caps)
 
163
        self.sink.set_caps(caps)
 
164
        self.sink.set_chain_function(self._chain_func)
 
165
        self.buffers = []
 
166
 
 
167
    def tearDown(self):
 
168
        self.assertEquals(sys.getrefcount(self.src), 3)
 
169
        self.assertEquals(self.src.__gstrefcount__, 1)
 
170
        del self.src
 
171
        self.assertEquals(sys.getrefcount(self.sink), 3)
 
172
        self.assertEquals(self.sink.__gstrefcount__, 1)
 
173
        del self.sink
 
174
        TestCase.tearDown(self)
 
175
 
 
176
    def _chain_func(self, pad, buffer):
 
177
        self.buffers.append(buffer)
 
178
 
 
179
        return gst.FLOW_OK
 
180
 
 
181
    def testProbeLink(self):
 
182
        id = self.src.add_buffer_probe(self._probe_handler)
 
183
        self.buffer = gst.Buffer()
 
184
        self.assertEquals(self.buffer.__grefcount__, 1)
 
185
        gst.debug('pushing buffer on linked pad, no probe')
 
186
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
 
187
        gst.debug('pushed buffer on linked pad, no probe')
 
188
        # one refcount is held by our scope, another is held on
 
189
        # self.buffers through _chain_func
 
190
        self.assertEquals(self.buffer.__grefcount__, 2)
 
191
        self.assertEquals(len(self.buffers), 1)
 
192
        self.buffers = None
 
193
        self.assertEquals(self.buffer.__grefcount__, 1)
 
194
 
 
195
 
 
196
    def _probe_handler(self, pad, buffer):
 
197
        self.src.link(self.sink)
 
198
        return True
 
199
      
 
200
 
 
201
class PadTest(TestCase):
 
202
    def testConstructor(self):
 
203
        # first style uses gst_pad_new
 
204
        gst.debug('creating pad with name src')
 
205
        pad = gst.Pad("src", gst.PAD_SRC)
 
206
        self.failUnless(pad)
 
207
        self.assertEquals(sys.getrefcount(pad), 3)
 
208
        self.assertEquals(pad.__gstrefcount__, 1)
 
209
 
 
210
        gst.debug('creating pad with no name')
 
211
        self.failUnless(gst.Pad(None, gst.PAD_SRC))
 
212
        self.failUnless(gst.Pad(name=None, direction=gst.PAD_SRC))
 
213
        self.failUnless(gst.Pad(direction=gst.PAD_SRC, name=None))
 
214
        self.failUnless(gst.Pad(direction=gst.PAD_SRC, name="src"))
 
215
 
 
216
        # second uses gst_pad_new_from_template
 
217
        #template = gst.PadTemplate()
 
218
 
 
219
class PadPipelineTest(TestCase):
 
220
    def setUp(self):
 
221
        TestCase.setUp(self)
 
222
        self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
 
223
        src = self.pipeline.get_by_name('source')
 
224
        self.srcpad = src.get_pad('src')
 
225
 
 
226
    def tearDown(self):
 
227
        del self.pipeline
 
228
        del self.srcpad
 
229
        TestCase.tearDown(self)
 
230
        
 
231
# FIXME: now that GstQuery is a miniobject with various _new_ factory
 
232
# functions, we need to figure out a way to deal with them in python
 
233
#    def testQuery(self):
 
234
#        assert self.sink.query(gst.QUERY_TOTAL, gst.FORMAT_BYTES) == -1
 
235
#        assert self.srcpad.query(gst.QUERY_POSITION, gst.FORMAT_BYTES) == 0
 
236
#        assert self.srcpad.query(gst.QUERY_POSITION, gst.FORMAT_TIME) == 0
 
237
 
 
238
 
 
239
class PadProbePipeTest(TestCase):
 
240
    def setUp(self):
 
241
        TestCase.setUp(self)
 
242
        self.pipeline = gst.Pipeline()
 
243
        self.assertEquals(self.pipeline.__gstrefcount__, 1)
 
244
        self.assertEquals(sys.getrefcount(self.pipeline), 3)
 
245
 
 
246
        self.fakesrc = gst.element_factory_make('fakesrc')
 
247
        self.fakesink = gst.element_factory_make('fakesink')
 
248
        self.assertEquals(self.fakesrc.__gstrefcount__, 1)
 
249
        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
 
250
 
 
251
        self.pipeline.add(self.fakesrc, self.fakesink)
 
252
        self.assertEquals(self.fakesrc.__gstrefcount__, 2) # added
 
253
        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
 
254
        self.assertEquals(self.fakesink.__gstrefcount__, 2) # added
 
255
        self.assertEquals(sys.getrefcount(self.fakesink), 3)
 
256
 
 
257
        self.fakesrc.link(self.fakesink)
 
258
 
 
259
        self.assertEquals(self.pipeline.__gstrefcount__, 1)
 
260
        self.assertEquals(sys.getrefcount(self.pipeline), 3)
 
261
        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
 
262
        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
 
263
        self.assertEquals(self.fakesink.__gstrefcount__, 2)
 
264
        self.assertEquals(sys.getrefcount(self.fakesink), 3)
 
265
 
 
266
    def tearDown(self):
 
267
        # Refcount must be either 1 or 2, to allow for a possibly still running
 
268
        # state-recalculation thread
 
269
        self.assertTrue (self.pipeline.__gstrefcount__ >= 1 and self.pipeline.__gstrefcount__ <= 2)
 
270
 
 
271
        self.assertEquals(sys.getrefcount(self.pipeline), 3)
 
272
        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
 
273
        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
 
274
        gst.debug('deleting pipeline')
 
275
        del self.pipeline
 
276
        self.gccollect()
 
277
 
 
278
        self.assertEquals(self.fakesrc.__gstrefcount__, 1) # parent gone
 
279
        self.assertEquals(self.fakesink.__gstrefcount__, 1) # parent gone
 
280
        self.assertEquals(sys.getrefcount(self.fakesrc), 3)
 
281
        self.assertEquals(sys.getrefcount(self.fakesink), 3)
 
282
        gst.debug('deleting fakesrc')
 
283
        del self.fakesrc
 
284
        self.gccollect()
 
285
        gst.debug('deleting fakesink')
 
286
        del self.fakesink
 
287
        self.gccollect()
 
288
 
 
289
        TestCase.tearDown(self)
 
290
        
 
291
    def testFakeSrcProbeOnceKeep(self):
 
292
        self.fakesrc.set_property('num-buffers', 1)
 
293
 
 
294
        self.fakesink.set_property('signal-handoffs', True)
 
295
        self.fakesink.connect('handoff', self._handoff_callback_fakesink)
 
296
 
 
297
        pad = self.fakesrc.get_pad('src')
 
298
        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
 
299
        self._got_fakesrc_buffer = 0
 
300
        self._got_fakesink_buffer = 0
 
301
        self.pipeline.set_state(gst.STATE_PLAYING)
 
302
        while not self._got_fakesrc_buffer:
 
303
            gst.debug('waiting for fakesrc buffer')
 
304
            pass
 
305
        while not self._got_fakesink_buffer:
 
306
            gst.debug('waiting for fakesink buffer')
 
307
            pass
 
308
 
 
309
        gst.debug('got buffers from fakesrc and fakesink')
 
310
        self.assertEquals(self._got_fakesink_buffer, 1)
 
311
        pad.remove_buffer_probe(id)
 
312
 
 
313
        self.pipeline.set_state(gst.STATE_NULL)
 
314
 
 
315
    def testFakeSrcProbeMany(self):
 
316
        self.fakesrc.set_property('num-buffers', 1000)
 
317
 
 
318
        pad = self.fakesrc.get_pad('src')
 
319
        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
 
320
        self._got_fakesrc_buffer = 0
 
321
        self.pipeline.set_state(gst.STATE_PLAYING)
 
322
        while not self._got_fakesrc_buffer == 1000:
 
323
            import time
 
324
            # allow for context switching; a busy loop here locks up the
 
325
            # streaming thread too much
 
326
            time.sleep(.0001)
 
327
        pad.remove_buffer_probe(id)
 
328
 
 
329
        self.pipeline.set_state(gst.STATE_NULL)
 
330
 
 
331
    def _probe_callback_fakesrc(self, pad, buffer):
 
332
        self.failUnless(isinstance(pad, gst.Pad))
 
333
        self.failUnless(isinstance(buffer, gst.Buffer))
 
334
        self._got_fakesrc_buffer += 1
 
335
        gst.debug('fakesrc sent buffer %r, %d total sent' % (
 
336
            buffer, self._got_fakesrc_buffer))
 
337
        return True
 
338
 
 
339
    def _handoff_callback_fakesink(self, sink, buffer, pad):
 
340
        self.failUnless(isinstance(buffer, gst.Buffer))
 
341
        self.failUnless(isinstance(pad, gst.Pad))
 
342
        self._got_fakesink_buffer += 1
 
343
        gst.debug('fakesink got buffer %r, %d total received' % (
 
344
            buffer, self._got_fakesrc_buffer))
 
345
        gst.debug('pad %r, py refcount %d, go rc %d, gst rc %d' % (
 
346
            pad, sys.getrefcount(pad), pad.__grefcount__, pad.__gstrefcount__))
 
347
        return True
 
348
 
 
349
    def testRemovingProbe(self):
 
350
        self.fakesrc.set_property('num-buffers', 10)
 
351
 
 
352
        handle = None
 
353
        self._num_times_called = 0
 
354
        def buffer_probe(pad, buffer):
 
355
            self._num_times_called += 1
 
356
            pad.remove_buffer_probe(handle)
 
357
            return True
 
358
 
 
359
        pad = self.fakesrc.get_pad('src')
 
360
        handle = pad.add_buffer_probe(buffer_probe)
 
361
        self.pipeline.set_state(gst.STATE_PLAYING)
 
362
        m = self.pipeline.get_bus().poll(gst.MESSAGE_EOS, -1)
 
363
        assert m
 
364
        assert self._num_times_called == 1
 
365
        self.pipeline.set_state(gst.STATE_NULL)
 
366
        # FIXME: having m going out of scope doesn't seem to be enough
 
367
        # to get it gc collected, and it keeps a ref to the pipeline.
 
368
        # Look for a way to not have to do this explicitly
 
369
        del m
 
370
        self.gccollect()
 
371
 
 
372
class PadRefCountTest(TestCase):
 
373
    def testAddPad(self):
 
374
        # add a pad to an element
 
375
        e = gst.element_factory_make('fakesrc')
 
376
        self.assertEquals(sys.getrefcount(e), 3)
 
377
        self.assertEquals(e.__gstrefcount__, 1)
 
378
 
 
379
        gst.debug('creating pad with name mypad')
 
380
        pad = gst.Pad("mypad", gst.PAD_SRC)
 
381
        self.failUnless(pad)
 
382
        self.assertEquals(sys.getrefcount(pad), 3)
 
383
        self.assertEquals(pad.__gstrefcount__, 1)
 
384
 
 
385
        gst.debug('adding pad to element')
 
386
        e.add_pad(pad)
 
387
        self.assertEquals(sys.getrefcount(e), 3)
 
388
        self.assertEquals(e.__gstrefcount__, 1)
 
389
        self.assertEquals(sys.getrefcount(pad), 3)
 
390
        self.assertEquals(pad.__gstrefcount__, 2) # added to element
 
391
 
 
392
        gst.debug('deleting element and collecting')
 
393
        self.gccollect()
 
394
        del e
 
395
        self.assertEquals(self.gccollect(), 1) # collected the element
 
396
        self.assertEquals(sys.getrefcount(pad), 3)
 
397
        self.assertEquals(pad.__gstrefcount__, 1) # removed from element
 
398
 
 
399
        gst.debug('deleting pad and collecting')
 
400
        del pad
 
401
        self.assertEquals(self.gccollect(), 1) # collected the pad
 
402
        gst.debug('going into teardown')
 
403
 
 
404
if __name__ == "__main__":
 
405
    unittest.main()