testsuite/test_message.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) 2005 Thomas Vander Stichele
   6 #
   7 # This library is free software; you can redistribute it and/or
   8 # modify it under the terms of the GNU Lesser General Public
   9 # License as published by the Free Software Foundation; either
  10 # version 2.1 of the License, or (at your option) any later version.
  11 #
  12 # This library is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15 # Lesser General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU Lesser General Public
  18 # License along with this library; if not, write to the Free Software
  19 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
  20
  21 import sys
  22 from common import gobject, gst, unittest, TestCase
  23 import gc
  24
  25 class NewTest(TestCase):
  26     def testEOS(self):
  27         gst.info("creating new bin")
  28         b = gst.Bin()
  29         gst.info("creating new EOS message from that bin")
  30         m = gst.message_new_eos(b)
  31         gst.info("got message : %s" % m)
  32
  33     def message_application_cb(self, bus, message):
  34         gst.info("got application message")
  35         self.got_message = True
  36         self.loop.quit()
  37
  38     def testApplication(self):
  39         self.loop = gobject.MainLoop()
  40         gst.info("creating new pipeline")
  41         bin = gst.Pipeline()
  42         bus = bin.get_bus()
  43         bus.add_signal_watch()
  44         self.got_message = False
  45         bus.connect('message::application', self.message_application_cb)
  46
  47         struc = gst.Structure("foo")
  48         msg = gst.message_new_application(bin, struc)
  49         # the bus is flushing in NULL, so we need to set the pipeline to READY
  50         bin.set_state(gst.STATE_READY)
  51         bus.post(msg)
  52         self.loop.run()
  53         bus.remove_signal_watch()
  54         bin.set_state(gst.STATE_NULL)
  55         self.failUnless(self.got_message == True)
  56         self.gccollect()
  57
  58 class TestCreateMessages(TestCase):
  59
  60     def setUp(self):
  61         TestCase.setUp(self)
  62         self.element = gst.Bin()
  63
  64     def tearDown(self):
  65         del self.element
  66
  67     def testCustomMessage(self):
  68         # create two custom messages using the same structure
  69         s = gst.Structure("something")
  70         assert s != None
  71         e1 = gst.message_new_custom(gst.MESSAGE_APPLICATION, self.element, s)
  72         assert e1
  73         e2 = gst.message_new_custom(gst.MESSAGE_APPLICATION, self.element, s)
  74         assert e2
  75
  76         # make sure the two structures are equal
  77         self.assertEquals(e1.structure.to_string(),
  78                           e2.structure.to_string())
  79
  80     def testTagMessage(self):
  81         # Create a taglist
  82         t = gst.TagList()
  83         t['something'] = "else"
  84         t['another'] = 42
  85
  86         # Create two messages using that same taglist
  87         m1 = gst.message_new_tag(self.element, t)
  88         assert m1
  89         m2 = gst.message_new_tag(self.element, t)
  90         assert m2
  91
  92         # make sure the two messages have the same taglist
  93         t1 = m1.parse_tag()
  94         assert t1
  95         keys = t1.keys()
  96         keys.sort()
  97         self.assertEquals(keys, ['another', 'something'])
  98         self.assertEquals(t1['something'], "else")
  99         self.assertEquals(t1['another'], 42)
 100         t2 = m2.parse_tag()
 101         assert t2
 102         keys = t2.keys()
 103         keys.sort()
 104         self.assertEquals(keys, ['another', 'something'])
 105         self.assertEquals(t2['something'], "else")
 106         self.assertEquals(t2['another'], 42)
 107
 108     def testTagFullMessage(self):
 109         if hasattr(gst.Message, 'parse_tag_full'):
 110             p = gst.Pad("blahblah", gst.PAD_SRC)
 111             # Create a taglist
 112             t = gst.TagList()
 113             t['something'] = "else"
 114             t['another'] = 42
 115
 116             # Create two messages using that same taglist
 117             m1 = gst.message_new_tag_full(self.element, p, t)
 118             assert m1
 119             m2 = gst.message_new_tag_full(self.element, p, t)
 120             assert m2
 121
 122             # make sure the two messages have the same taglist
 123             p1, t1 = m1.parse_tag_full()
 124             assert t1
 125             keys = t1.keys()
 126             keys.sort()
 127             self.assertEquals(p1, p)
 128             self.assertEquals(keys, ['another', 'something'])
 129             self.assertEquals(t1['something'], "else")
 130             self.assertEquals(t1['another'], 42)
 131             p2, t2 = m2.parse_tag_full()
 132             assert t2
 133             keys = t2.keys()
 134             keys.sort()
 135             self.assertEquals(p2, p)
 136             self.assertEquals(keys, ['another', 'something'])
 137             self.assertEquals(t2['something'], "else")
 138             self.assertEquals(t2['another'], 42)
 139
 140     def testStepStartMessage(self):
 141         if hasattr(gst, 'message_new_step_start'):
 142             m = gst.message_new_step_start(self.element, True,
 143                                            gst.FORMAT_TIME, 42, 1.0,
 144                                            True, True)
 145             self.assertEquals(m.type, gst.MESSAGE_STEP_START)
 146             active, format, amount, rate, flush, intermediate = m.parse_step_start()
 147             self.assertEquals(active, True)
 148             self.assertEquals(format, gst.FORMAT_TIME)
 149             self.assertEquals(amount, 42)
 150             self.assertEquals(rate, 1.0)
 151             self.assertEquals(flush, True)
 152             self.assertEquals(intermediate, True)
 153
 154     def testStepDoneMessage(self):
 155         if hasattr(gst, 'message_new_step_done'):
 156             m = gst.message_new_step_done(self.element, gst.FORMAT_TIME, 42,
 157                                           1.0, True, True, 54, True)
 158             self.assertEquals(m.type, gst.MESSAGE_STEP_DONE)
 159
 160             fmt, am, rat, flu, inter, dur, eos = m.parse_step_done()
 161             self.assertEquals(fmt, gst.FORMAT_TIME)
 162             self.assertEquals(am, 42)
 163             self.assertEquals(rat, 1.0)
 164             self.assertEquals(flu, True)
 165             self.assertEquals(inter, True)
 166             self.assertEquals(dur, 54)
 167             self.assertEquals(eos, True)
 168
 169     def testStructureChangeMessage(self):
 170         if hasattr(gst, 'message_new_structure_change'):
 171             p = gst.Pad("blah", gst.PAD_SINK)
 172             m = gst.message_new_structure_change(p,
 173                                                  gst.STRUCTURE_CHANGE_TYPE_PAD_LINK,
 174                                                  self.element, True)
 175
 176             self.assertEquals(m.type, gst.MESSAGE_STRUCTURE_CHANGE)
 177             sct, owner, busy = m.parse_structure_change()
 178             self.assertEquals(sct, gst.STRUCTURE_CHANGE_TYPE_PAD_LINK)
 179             self.assertEquals(owner, self.element)
 180             self.assertEquals(busy, True)
 181
 182     def testRequestStateMessage(self):
 183         if hasattr(gst, 'message_new_request_state'):
 184             m = gst.message_new_request_state(self.element, gst.STATE_NULL)
 185             self.assertEquals(m.type, gst.MESSAGE_REQUEST_STATE)
 186             self.assertEquals(m.parse_request_state(), gst.STATE_NULL)
 187
 188     def testBufferingStatsMessage(self):
 189         if hasattr(gst.Message, 'set_buffering_stats'):
 190             gst.debug("Creating buffering message")
 191             m = gst.message_new_buffering(self.element, 50)
 192             gst.debug("Setting stats")
 193             m.set_buffering_stats(gst.BUFFERING_LIVE, 30, 1024, 123456)
 194             self.assertEquals(m.type, gst.MESSAGE_BUFFERING)
 195             mode, ain, aout, left = m.parse_buffering_stats()
 196             self.assertEquals(mode, gst.BUFFERING_LIVE)
 197             self.assertEquals(ain, 30)
 198             self.assertEquals(aout, 1024)
 199             self.assertEquals(left, 123456)
 200
 201 if __name__ == "__main__":
 202     unittest.main()