# -*- Mode: Python -*-
# 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA

from common import gst, unittest, TestCase, pygobject_2_13

import sys
import time

class PadTemplateTest(TestCase):
    def testConstructor(self):
        template = gst.PadTemplate("template", gst.PAD_SINK,
            gst.PAD_ALWAYS, gst.caps_from_string("audio/x-raw-int"))
        self.failUnless(template)
        self.assertEquals(sys.getrefcount(template), pygobject_2_13 and 2 or 3)
        #self.assertEquals(template.__gstrefcount__, 1)

class PadPushUnlinkedTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.src = gst.Pad("src", gst.PAD_SRC)
        self.sink = gst.Pad("sink", gst.PAD_SINK)

    def tearDown(self):
        self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.src.__gstrefcount__, 1)
        del self.src
        self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.sink.__gstrefcount__, 1)
        del self.sink
        TestCase.tearDown(self)

    def testNoProbe(self):
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
        # pushing it takes a ref in the python wrapper to keep buffer
        # alive afterwards; but the core unrefs the ref it receives
        self.assertEquals(self.buffer.__grefcount__, 1)

    def testFalseProbe(self):
        id = self.src.add_buffer_probe(self._probe_handler, False)
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.src.remove_buffer_probe(id)

    def testTrueProbe(self):
        id = self.src.add_buffer_probe(self._probe_handler, True)
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_NOT_LINKED)
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.src.remove_buffer_probe(id)

    def _probe_handler(self, pad, buffer, ret):
        return ret

class PadPushLinkedTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.src = gst.Pad("src", gst.PAD_SRC)
        self.sink = gst.Pad("sink", gst.PAD_SINK)
        caps = gst.caps_from_string("foo/bar")
        self.src.set_caps(caps)
        self.sink.set_caps(caps)
        self.sink.set_chain_function(self._chain_func)
        self.src.set_active(True)
        self.sink.set_active(True)
        self.src.link(self.sink)
        self.buffers = []

    def tearDown(self):
        self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.src.__gstrefcount__, 1)
        self.src.set_caps(None)
        del self.src
        self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.sink.__gstrefcount__, 1)
        self.sink.set_caps(None)
        del self.sink
        TestCase.tearDown(self)

    def _chain_func(self, pad, buffer):
        gst.debug('got buffer %r, id %x, with GMO rc %d'% (
            buffer, id(buffer), buffer.__grefcount__))
        self.buffers.append(buffer)

        return gst.FLOW_OK

    def testNoProbe(self):
        self.buffer = gst.Buffer()
        gst.debug('created new buffer %r, id %x' % (
            self.buffer, id(self.buffer)))
        self.assertEquals(self.buffer.__grefcount__, 1)
        gst.debug('pushing buffer on linked pad, no probe')
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
        gst.debug('pushed buffer on linked pad, no probe')
        # one refcount is held by our scope, another is held on
        # self.buffers through _chain_func
        self.assertEquals(self.buffer.__grefcount__, 2)
        self.assertEquals(len(self.buffers), 1)
        self.buffers = None
        self.assertEquals(self.buffer.__grefcount__, 1)

    def testFalseProbe(self):
        id = self.src.add_buffer_probe(self._probe_handler, False)
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.src.remove_buffer_probe(id)
        self.assertEquals(len(self.buffers), 0)

    def testTrueProbe(self):
        probe_id = self.src.add_buffer_probe(self._probe_handler, True)
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
        # one refcount is held by our scope, another is held on
        # self.buffers through _chain_func
        self.assertEquals(self.buffer.__grefcount__, 2)

        # they are not the same Python object ...
        self.failIf(self.buffer is self.buffers[0])
        self.failIf(id(self.buffer) == id(self.buffers[0]))
        # ... but they wrap the same GstBuffer
        self.failUnless(self.buffer == self.buffers[0])
        self.assertEquals(repr(self.buffer), repr(self.buffers[0]))
        
        self.src.remove_buffer_probe(probe_id)
        self.assertEquals(len(self.buffers), 1)
        self.buffers = None
        self.assertEquals(self.buffer.__grefcount__, 1)

    def _probe_handler(self, pad, buffer, ret):
        return ret

# test for event probes with linked pads
class PadPushEventLinkedTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.src = gst.Pad("src", gst.PAD_SRC)
        self.sink = gst.Pad("sink", gst.PAD_SINK)
        caps = gst.caps_from_string("foo/bar")
        self.src.set_caps(caps)
        self.sink.set_caps(caps)
        self.sink.set_chain_function(self._chain_func)
        self.src.set_active(True)
        self.sink.set_active(True)
        self.src.link(self.sink)
        self.events = []

    def tearDown(self):
        self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.src.__gstrefcount__, 1)
        self.src.set_caps(None)
        del self.src
        self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.sink.__gstrefcount__, 1)
        self.sink.set_caps(None)
        del self.sink
        TestCase.tearDown(self)

    def _chain_func(self, pad, buffer):
        gst.debug('got buffer %r, id %x, with GMO rc %d'% (
            buffer, id(buffer), buffer.__grefcount__))
        self.buffers.append(buffer)

        return gst.FLOW_OK

    def testNoProbe(self):
        self.event = gst.event_new_eos()
        gst.debug('created new eos %r, id %x' % (
            self.event, id(self.event)))
        self.assertEquals(self.event.__grefcount__, 1)
        gst.debug('pushing event on linked pad, no probe')
        self.assertEquals(self.src.push_event(self.event), True)
        gst.debug('pushed event on linked pad, no probe')
        # one refcount is held by our scope
        self.assertEquals(self.event.__grefcount__, 1)
        # the event has reffed the src pad as the src of the event
        self.assertEquals(self.src.__grefcount__, 2)
        # clear it
        self.event = None
        self.assertEquals(self.src.__grefcount__, 1)

    def testFalseProbe(self):
        probe_id = self.src.add_event_probe(self._probe_handler, False)
        self.event = gst.event_new_eos()
        gst.debug('created new eos %r, id %x' % (
            self.event, id(self.event)))
        self.assertEquals(self.event.__grefcount__, 1)
        # a false probe return drops the event and returns False
        self.assertEquals(self.src.push_event(self.event), False)
        # one ref in our local scope, another in self.events
        self.assertEquals(self.event.__grefcount__, 2)
        self.assertEquals(self.sink.__grefcount__, 1)
        # the event has reffed the src pad as the src of the event
        self.assertEquals(self.src.__grefcount__, 2)
        # remove the event from existence
        self.event = None
        self.events = None
        self.assertEquals(self.src.__grefcount__, 1)
        self.src.remove_buffer_probe(probe_id)

    def testTrueProbe(self):
        probe_id = self.src.add_event_probe(self._probe_handler, True)
        self.event = gst.event_new_eos()
        gst.debug('created new eos %r, id %x' % (
            self.event, id(self.event)))
        self.assertEquals(self.event.__grefcount__, 1)
        # a True probe lets it pass
        self.assertEquals(self.src.push_event(self.event), True)

        # one refcount is held by our scope, another is held on
        # self.events through _probe
        self.assertEquals(self.event.__grefcount__, 2)

        # they are not the same Python object ...
        self.failIf(self.event is self.events[0])
        self.failIf(id(self.event) == id(self.events[0]))
        # ... but they wrap the same GstEvent
        self.assertEquals(repr(self.event), repr(self.events[0]))
        self.failUnless(self.event == self.events[0])
        
        self.src.remove_buffer_probe(probe_id)
        self.assertEquals(len(self.events), 1)
        self.events = None
        self.assertEquals(self.event.__grefcount__, 1)

        # the event has reffed the src pad as the src of the event
        self.assertEquals(self.src.__grefcount__, 2)
        # clear it
        self.event = None
        self.assertEquals(self.src.__grefcount__, 1)

    def _probe_handler(self, pad, event, ret):
        gst.debug("probed, pad %r, event %r" % (pad, event))
        self.events.append(event)
        return ret

# a test to show that we can link a pad from the probe handler

class PadPushProbeLinkTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.src = gst.Pad("src", gst.PAD_SRC)
        self.sink = gst.Pad("sink", gst.PAD_SINK)
        caps = gst.caps_from_string("foo/bar")
        self.src.set_caps(caps)
        self.sink.set_caps(caps)
        self.src.set_active(True)
        self.sink.set_active(True)
        self.sink.set_chain_function(self._chain_func)
        self.buffers = []

    def tearDown(self):
        self.assertEquals(sys.getrefcount(self.src), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.src.__gstrefcount__, 1)
        self.src.set_caps(None)
        del self.src
        self.assertEquals(sys.getrefcount(self.sink), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.sink.__gstrefcount__, 1)
        self.sink.set_caps(None)
        del self.sink
        TestCase.tearDown(self)

    def _chain_func(self, pad, buffer):
        self.buffers.append(buffer)

        return gst.FLOW_OK

    def testProbeLink(self):
        id = self.src.add_buffer_probe(self._probe_handler)
        self.buffer = gst.Buffer()
        self.assertEquals(self.buffer.__grefcount__, 1)
        gst.debug('pushing buffer on linked pad, no probe')
        self.assertEquals(self.src.push(self.buffer), gst.FLOW_OK)
        gst.debug('pushed buffer on linked pad, no probe')
        # one refcount is held by our scope, another is held on
        # self.buffers through _chain_func
        self.assertEquals(self.buffer.__grefcount__, 2)
        self.assertEquals(len(self.buffers), 1)
        self.buffers = None
        self.assertEquals(self.buffer.__grefcount__, 1)


    def _probe_handler(self, pad, buffer):
        self.src.link(self.sink)
        return True
      
 
class PadTest(TestCase):
    def testConstructor(self):
        # first style uses gst_pad_new
        gst.debug('creating pad with name src')
        pad = gst.Pad("src", gst.PAD_SRC)
        self.failUnless(pad)
        self.assertEquals(sys.getrefcount(pad), pygobject_2_13 and 2 or 3)
        self.assertEquals(pad.__gstrefcount__, 1)

        gst.debug('creating pad with no name')
        self.failUnless(gst.Pad(None, gst.PAD_SRC))
        self.failUnless(gst.Pad(name=None, direction=gst.PAD_SRC))
        self.failUnless(gst.Pad(direction=gst.PAD_SRC, name=None))
        self.failUnless(gst.Pad(direction=gst.PAD_SRC, name="src"))

        # second uses gst_pad_new_from_template
        #template = gst.PadTemplate()

class PadPipelineTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.pipeline = gst.parse_launch('fakesrc name=source ! fakesink')
        src = self.pipeline.get_by_name('source')
        self.srcpad = src.get_pad('src')

    def tearDown(self):
        del self.pipeline
        del self.srcpad
        TestCase.tearDown(self)
        
# 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 PadProbePipeTest(TestCase):
    def setUp(self):
        TestCase.setUp(self)
        self.pipeline = gst.Pipeline()
        self.assertEquals(self.pipeline.__gstrefcount__, 1)
        self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3)

        self.fakesrc = gst.element_factory_make('fakesrc')
        self.fakesink = gst.element_factory_make('fakesink')
        self.assertEquals(self.fakesrc.__gstrefcount__, 1)
        self.assertEquals(sys.getrefcount(self.fakesrc), pygobject_2_13 and 2 or 3)

        self.pipeline.add(self.fakesrc, self.fakesink)
        self.assertEquals(self.fakesrc.__gstrefcount__, 2) # added
        self.assertEquals(sys.getrefcount(self.fakesrc), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.fakesink.__gstrefcount__, 2) # added
        self.assertEquals(sys.getrefcount(self.fakesink), pygobject_2_13 and 2 or 3)

        self.fakesrc.link(self.fakesink)

        self.assertEquals(self.pipeline.__gstrefcount__, 1)
        self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
        self.assertEquals(sys.getrefcount(self.fakesrc), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.fakesink.__gstrefcount__, 2)
        self.assertEquals(sys.getrefcount(self.fakesink), pygobject_2_13 and 2 or 3)

    def tearDown(self):
        # Refcount must be either 1 or 2, to allow for a possibly still running
        # state-recalculation thread
        self.assertTrue (self.pipeline.__gstrefcount__ >= 1 and self.pipeline.__gstrefcount__ <= 2)

        self.assertEquals(sys.getrefcount(self.pipeline), pygobject_2_13 and 2 or 3)
        self.assertEquals(self.fakesrc.__gstrefcount__, 2)
        self.assertEquals(sys.getrefcount(self.fakesrc), pygobject_2_13 and 2 or 3)
        gst.debug('deleting pipeline')
        del self.pipeline
        self.gccollect()

        self.assertEquals(self.fakesrc.__gstrefcount__, 1) # parent gone
        self.assertEquals(self.fakesink.__gstrefcount__, 1) # parent gone
        self.assertEquals(sys.getrefcount(self.fakesrc), pygobject_2_13 and 2 or 3)
        self.assertEquals(sys.getrefcount(self.fakesink), pygobject_2_13 and 2 or 3)
        gst.debug('deleting fakesrc')
        del self.fakesrc
        self.gccollect()
        gst.debug('deleting fakesink')
        del self.fakesink
        self.gccollect()

        TestCase.tearDown(self)
        
    def testFakeSrcProbeOnceKeep(self):
        self.fakesrc.set_property('num-buffers', 1)

        self.fakesink.set_property('signal-handoffs', True)
        self.fakesink.connect('handoff', self._handoff_callback_fakesink)

        pad = self.fakesrc.get_pad('src')
        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
        self._got_fakesrc_buffer = 0
        self._got_fakesink_buffer = 0
        self.pipeline.set_state(gst.STATE_PLAYING)
        while not self._got_fakesrc_buffer:
            gst.debug('waiting for fakesrc buffer')
            pass
        while not self._got_fakesink_buffer:
            gst.debug('waiting for fakesink buffer')
            pass

        gst.debug('got buffers from fakesrc and fakesink')
        self.assertEquals(self._got_fakesink_buffer, 1)
        pad.remove_buffer_probe(id)

        self.pipeline.set_state(gst.STATE_NULL)

    def testFakeSrcProbeMany(self):
        self.fakesrc.set_property('num-buffers', 1000)

        pad = self.fakesrc.get_pad('src')
        id = pad.add_buffer_probe(self._probe_callback_fakesrc)
        self._got_fakesrc_buffer = 0
        self.pipeline.set_state(gst.STATE_PLAYING)
        while not self._got_fakesrc_buffer == 1000:
            import time
            # allow for context switching; a busy loop here locks up the
            # streaming thread too much
            time.sleep(.0001)
        pad.remove_buffer_probe(id)

        self.pipeline.set_state(gst.STATE_NULL)

    def _probe_callback_fakesrc(self, pad, buffer):
        self.failUnless(isinstance(pad, gst.Pad))
        self.failUnless(isinstance(buffer, gst.Buffer))
        self._got_fakesrc_buffer += 1
        gst.debug('fakesrc sent buffer %r, %d total sent' % (
            buffer, self._got_fakesrc_buffer))
        return True

    def _handoff_callback_fakesink(self, sink, buffer, pad):
        self.failUnless(isinstance(buffer, gst.Buffer))
        self.failUnless(isinstance(pad, gst.Pad))
        self._got_fakesink_buffer += 1
        gst.debug('fakesink got buffer %r, %d total received' % (
            buffer, self._got_fakesrc_buffer))
        gst.debug('pad %r, py refcount %d, go rc %d, gst rc %d' % (
            pad, sys.getrefcount(pad), pad.__grefcount__, pad.__gstrefcount__))
        return True

    def testRemovingProbe(self):
        self.fakesrc.set_property('num-buffers', 10)

        handle = None
        self._num_times_called = 0
        def buffer_probe(pad, buffer, data):
            self._num_times_called += 1
            pad.remove_buffer_probe(handle)
            return True

        pad = self.fakesrc.get_pad('src')
        data = []
        handle = pad.add_buffer_probe(buffer_probe, data)
        self.pipeline.set_state(gst.STATE_PLAYING)
        m = self.pipeline.get_bus().poll(gst.MESSAGE_EOS, -1)
        assert m
        assert self._num_times_called == 1
        self.pipeline.set_state(gst.STATE_NULL)
        assert sys.getrefcount(buffer_probe) == 2
        assert sys.getrefcount(data) == 2
        # FIXME: having m going out of scope doesn't seem to be enough
        # to get it gc collected, and it keeps a ref to the pipeline.
        # Look for a way to not have to do this explicitly
        del m
        self.gccollect()

class PadRefCountTest(TestCase):
    def testAddPad(self):
        # add a pad to an element
        e = gst.element_factory_make('fakesrc')
        self.assertEquals(sys.getrefcount(e), pygobject_2_13 and 2 or 3)
        self.assertEquals(e.__gstrefcount__, 1)

        gst.debug('creating pad with name mypad')
        pad = gst.Pad("mypad", gst.PAD_SRC)
        self.failUnless(pad)
        self.assertEquals(sys.getrefcount(pad), pygobject_2_13 and 2 or 3)
        self.assertEquals(pad.__gstrefcount__, 1)

        gst.debug('adding pad to element')
        e.add_pad(pad)
        self.assertEquals(sys.getrefcount(e), pygobject_2_13 and 2 or 3)
        self.assertEquals(e.__gstrefcount__, 1)
        self.assertEquals(sys.getrefcount(pad), pygobject_2_13 and 2 or 3)
        self.assertEquals(pad.__gstrefcount__, 2) # added to element

        gst.debug('deleting element and collecting')
        self.gccollect()
        del e
        if not pygobject_2_13:
            # the element will be collected at 'del e' if we're using
            # pygobject >= 2.13.0
            self.assertEquals(self.gccollect(), 1) # collected the element
        self.assertEquals(sys.getrefcount(pad), pygobject_2_13 and 2 or 3)
        self.assertEquals(pad.__gstrefcount__, 1) # removed from element

        gst.debug('deleting pad and collecting')
        del pad
        if not pygobject_2_13:
            # the pad will be collected at 'del pad' if we're using
            # pygobject >= 2.13.0
            self.assertEquals(self.gccollect(), 1) # collected the pad
        gst.debug('going into teardown')

class PadBlockTest(TestCase):
    def testCallbackFlush(self):
        # check that the same block callback can be called more than once (weird
        # test but it was broken)

        def blocked_cb(pad, blocked):
            pad.push_event(gst.event_new_flush_start())
       
        pad = gst.Pad('src', gst.PAD_SRC)
        pad.set_active(True)
        pad.set_blocked_async(True, blocked_cb)

        for i in xrange(10):
            buf = gst.Buffer('ciao')
            pad.push(buf)
            pad.push_event(gst.event_new_flush_stop())

    def testCallbackRefcount(self):
        def blocked_cb(pad, blocked):
            pad.set_blocked_async(False, unblocked_cb)

        def unblocked_cb(pad, blocked):
            pass

        cb_refcount = sys.getrefcount(blocked_cb)
        # sys.getrefcount() returns refcount + 1
        self.assertEquals(cb_refcount, 2)
       
        pad = gst.Pad('src', gst.PAD_SRC)
        pad.set_active(True)
        pad.set_blocked_async(True, blocked_cb)
        # set_blocked_async refs the callback
        self.assertEquals(sys.getrefcount(blocked_cb), 3)
            
        buf = gst.Buffer('ciao')
        pad.push(buf)
        
        # in blocked_cb() we called set_blocked_async() with a different
        # callback, so blocked_cb() should have been unreffed
        cb_refcount_after = sys.getrefcount(blocked_cb)
        self.assertEquals(sys.getrefcount(blocked_cb), cb_refcount)

if __name__ == "__main__":
    unittest.main()