(self)
| 425 | self.pipeline.set_state(gst.STATE_NULL) |
| 426 | |
| 427 | def testFakeSrcProbeMany(self): |
| 428 | self.fakesrc.set_property('num-buffers', 1000) |
| 429 | |
| 430 | pad = self.fakesrc.get_pad('src') |
| 431 | id = pad.add_buffer_probe(self._probe_callback_fakesrc) |
| 432 | self._got_fakesrc_buffer = 0 |
| 433 | self.pipeline.set_state(gst.STATE_PLAYING) |
| 434 | while not self._got_fakesrc_buffer == 1000: |
| 435 | import time |
| 436 | # allow for context switching; a busy loop here locks up the |
| 437 | # streaming thread too much |
| 438 | time.sleep(.0001) |
| 439 | pad.remove_buffer_probe(id) |
| 440 | |
| 441 | self.pipeline.set_state(gst.STATE_NULL) |
| 442 | |
| 443 | def _probe_callback_fakesrc(self, pad, buffer): |
| 444 | self.failUnless(isinstance(pad, gst.Pad)) |
nothing calls this directly
no test coverage detected