1
# PiTiVi , Non-linear video editor
3
# tests/test_pipeline_action.py
5
# Copyright (c) 2009, Edward Hervey <bilboed@bilboed.com>
7
# This program 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.
12
# This program 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.
17
# You should have received a copy of the GNU Lesser General Public
18
# License along with this program; if not, write to the
19
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
20
# Boston, MA 02110-1301, USA.
23
Tests for interaction between action and pipeline
27
from unittest import TestCase, main
28
from pitivi.pipeline import Pipeline, STATE_READY, STATE_PLAYING, STATE_NULL
29
from pitivi.action import Action, STATE_ACTIVE, STATE_NOT_ACTIVE, ActionError
30
from pitivi.stream import MultimediaStream, VideoStream
31
from pitivi.factories.test import VideoTestSourceFactory
32
from common import TestCase
37
class DynamicAction(Action):
38
def getDynamicLinks(self, producer, stream):
39
links = Action.getDynamicLinks(self, producer, stream)
40
consumer = common.FakeSinkFactory()
42
links.append((producer, consumer, stream, None))
43
gst.debug("Returning link")
47
class TestPipelineAction(TestCase):
50
gst.debug("Test starting")
53
def testPipelineAction(self):
54
"""Testing pipeline state interaction"""
57
src = VideoTestSourceFactory()
58
sink = common.FakeSinkFactory()
59
sink.addInputStream(MultimediaStream(gst.Caps("any"), pad_name="sink"))
61
# set the Action on the Pipeline
63
self.assertEquals(p.actions, [a])
65
# set the Producer and Consumer
74
self.failUnlessEqual(src.current_bins, 1)
75
self.failUnlessEqual(sink.current_bins, 1)
77
# call get*ForFactoryStream(..., automake=False). They will raise
78
# exceptions if the action didn't create the elements.
79
bin = p.getBinForFactoryStream(src, automake=False)
80
p.releaseBinForFactoryStream(src)
82
tee = p.getTeeForFactoryStream(src, automake=False)
83
p.releaseTeeForFactoryStream(src)
85
bin = p.getBinForFactoryStream(sink, automake=False)
87
queue = p.getQueueForFactoryStream(sink, automake=False)
89
self.failUnlessEqual(queue.get_pad('src').get_peer().get_parent(), bin)
91
p.releaseBinForFactoryStream(sink)
92
p.releaseQueueForFactoryStream(sink)
95
p.setState(STATE_PLAYING)
100
p.setState(STATE_READY)
105
# since we're the last Action to be release, the tees
106
# and queues should have gone
107
self.failUnlessEqual(src.current_bins, 0)
108
self.failUnlessEqual(sink.current_bins, 0)
110
# remove the action from the pipeline
113
# the gst.Pipeline should be empty !
114
self.assertEquals(list(p._pipeline.elements()), [])
118
def testPendingLink(self):
121
src = common.FakeGnlFactory()
122
src.addOutputStream(VideoStream(gst.Caps("video/x-raw-yuv"),
124
sink = common.FakeSinkFactory()
125
sink.addInputStream(MultimediaStream(gst.Caps("any"),
128
# set the link, it will be activated once the pad is added
130
# Let's see if the link is present
131
self.assertEquals(a._links, [(src, sink, None, None)])
135
gst.debug("about to activate action")
137
# only the producer and the consumer are created, the other elements are
138
# created dinamically
139
self.assertEquals(len(list(p._pipeline.elements())), 2)
141
p.setState(STATE_PLAYING)
143
# and make sure that all other elements were created (4)
144
# FIXME if it's failing here, run the test a few times trying to raise
145
# the time.sleep() above, it may just be racy...
146
self.assertEquals(len(list(p._pipeline.elements())), 4)
149
p.setState(STATE_NULL)
150
self.assertEquals(len(list(p._pipeline.elements())), 0)
153
def testDynamicLink(self):
156
src = common.FakeGnlFactory()
157
src.addOutputStream(VideoStream(gst.Caps("video/x-raw-yuv"),
160
# the link will be added dynamically
161
self.assertEquals(a._links, [])
166
self.assertEquals(len(list(p._pipeline.elements())), 0)
169
# theoretically... there shouldn't only be the source, since
170
# the pad for the source hasn't been created yet (and therefore not
171
# requiring a consumer
172
self.assertEquals(len(list(p._pipeline.elements())), 1)
174
p.setState(STATE_PLAYING)
178
# and make sure that all other elements were created (4)
179
self.assertEquals(len(list(p._pipeline.elements())), 4)
181
p.setState(STATE_READY)
185
self.assertEquals(len(list(p._pipeline.elements())), 0)
186
p.setState(STATE_NULL)
190
if __name__ == "__main__":