1
# PiTiVi , Non-linear video editor
3
# tests/test_integration.py
5
# Copyright (c) 2008, Alessandro Decina <alessandro.decina@collabora.co.uk>
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., 59 Temple Place - Suite 330,
20
# Boston, MA 02111-1307, USA.
22
"""Test pitivi core objects at the API level, simulating the UI input for
26
TestCase = unittest.TestCase
27
from pitivi.application import InteractivePitivi
28
from pitivi.timeline.timeline import MoveContext, TrimStartContext,\
30
from pitivi.signalinterface import Signallable
31
from pitivi.stream import AudioStream, VideoStream
32
import pitivi.instance
38
base_uri = "file:///" + os.getcwd() + "/media/"
39
test1 = base_uri + "test1.ogg"
40
test2 = base_uri + "test2.ogg"
41
test3 = base_uri + "test3.ogg"
43
class WatchDog(object):
45
"""A simple watchdog timer to aid developing integration tests. If
46
keepAlive() is not called every <timeout> ms, then the watchdog timer will
47
quit the specified mainloop."""
49
def __init__(self, mainloop, timeout=10000):
50
self.timeout = timeout
51
self.mainloop = mainloop
52
self.will_quit = False
53
self.keep_going = True
54
self.activated = False
57
self.will_quit = False
58
self.keep_going = True
59
gobject.timeout_add(self.timeout, self._timeoutcb)
63
self.keep_going = False
69
self.keep_going = False
71
return self.keep_going
74
self.will_quit = False
76
class TestWatchdog(TestCase):
78
def testWatchdog(self):
79
self.ml = gobject.MainLoop()
80
wd = WatchDog(self.ml, 100)
81
self.timeout_called = False
83
gobject.timeout_add(2000, self._timeoutCb)
85
self.assertFalse(self.timeout_called)
86
self.assertTrue(wd.activated)
88
def testKeepAlive(self):
89
self.ml = gobject.MainLoop()
90
wd = WatchDog(self.ml, 2000)
91
self.timeout_called = False
93
gobject.timeout_add(500, wd.keepAlive)
94
gobject.timeout_add(2500, self._timeoutCb)
96
self.assertTrue(self.timeout_called)
97
self.assertFalse(wd.activated)
99
def testSuspend(self):
100
self.ml = gobject.MainLoop()
101
wd = WatchDog(self.ml, 500)
102
self.timeout_called = False
105
gobject.timeout_add(2000, self._timeoutCb)
107
self.assertTrue(self.timeout_called)
108
self.assertFalse(wd.activated)
110
def _timeoutCb(self):
112
self.timeout_called = True
115
class Configuration(object):
117
def __init__(self, *sources):
120
for source in sources:
121
self.addSource(*source)
124
ret = Configuration()
125
for source in self.sources:
127
name, uri, props = source
128
ret.addSource(name, uri, dict(props))
130
ret.addBadSource(*source)
133
def addSource(self, name, uri, props=None, error=False):
134
if name in self.source_map:
135
raise Exception("Duplicate source: '%d' already defined" % name)
136
self.sources.append((name, uri, props))
137
self.source_map[name] = uri, props
139
def updateSource(self, name, uri=None, props=None):
140
def findSource(name):
141
for i, source in enumerate(self.sources):
142
if source[0] == name:
144
raise Exception("Source %s not in configuration" %
148
name, orig_uri, orig_props = self.sources[i]
152
orig_props.update(props)
154
self.sources[i] = (name, uri, orig_props)
155
self.source_map[name] = (uri, orig_props)
157
def addBadSource(self, name, uri):
158
if name in self.source_map:
159
raise Exception("Duplicate source: '%d' already defined" % name)
160
self.sources.append((name, uri))
161
self.source_map[name] = uri, None
164
return set((source[1] for source in self.sources))
166
def getGoodUris(self):
167
return set((source[1] for source in self.sources if
170
def getGoodSources(self):
171
return (source for source in self.sources if len(source) > 2)
173
def matches(self, instance_runner):
174
for name, uri, props in self.getGoodSources():
175
if not hasattr(instance_runner, name):
176
raise Exception("Project missing source %s" % name)
177
timelineObject = getattr(instance_runner, name)
178
if timelineObject.factory.uri != uri:
179
raise Exception("%s has wrong factory type!" % name)
181
for prop, value in props.iteritems():
182
actual = getattr(timelineObject, prop)
183
if not actual == value:
184
raise Exception("%s.%s: %r != %r" % (name, prop,
187
names = set((source[0] for source in self.getGoodSources()))
188
timelineObjects = set(instance_runner.timelineObjects.iterkeys())
189
if names != timelineObjects:
190
raise Exception("Project has extra sources: %r" % (timelineObjects -
194
return (source for source in self.sources if len(source) > 2)
196
class InstanceRunner(Signallable):
198
no_ui = not(os.getenv("ENABLE_UI"))
200
class container(object):
206
"sources-loaded" : [],
207
"timeline-configured" : [],
210
def __init__(self, instance):
211
self.instance = instance
212
self.watchdog = WatchDog(instance.mainloop, 10000)
213
self.factories = set()
218
self.timelineObjects = {}
219
self.pending_configuration = None
222
instance.connect("new-project-loaded", self._newProjectLoadedCb)
224
def loadConfiguration(self, configuration):
225
self.pending_configuration = configuration
227
def _newProjectLoadedCb(self, instance, project):
228
self.project = instance.current
229
self.timeline = self.project.timeline
230
for track in self.timeline.tracks:
231
self._trackAddedCb(self.timeline, track)
232
self.project.sources.connect("source-added", self._sourceAdded)
233
self.project.sources.connect("discovery-error", self._discoveryError)
234
self.project.sources.connect("ready", self._readyCb)
235
self.timeline.connect("track-added", self._trackAddedCb)
237
if self.pending_configuration:
238
self._loadSources(self.pending_configuration)
240
def _sourceAdded(self, sourcelist, factory):
241
self.factories.add(factory.uri)
243
def _discoveryError(self, sourcelist, uri, reason, unused):
246
def _readyCb(self, soucelist):
247
assert self.factories == self.pending_configuration.getGoodUris()
248
if self.pending_configuration:
249
self._setupTimeline(self.pending_configuration)
250
self.emit("sources-loaded")
252
def _loadSources(self, configuration):
253
for uri in configuration.getUris():
254
self.project.sources.addUri(uri)
256
def _trackAddedCb(self, timeline, track):
257
if type(track.stream) is AudioStream:
258
self.audioTracks += 1
259
attrname = "audio%d" % self.audioTracks
260
elif type(track.stream) is VideoStream:
261
self.videoTracks += 1
262
attrname = "video%d" % self.videoTracks
263
container = self.container()
264
setattr(self, attrname, container)
265
self.tracks[track] = container
266
container.transitions = {}
267
track.connect("transition-added", self._transitionAddedCb, container)
268
track.connect("transition-removed", self._transitionRemovedCb,
271
def _transitionAddedCb(self, track, transition, container):
272
container.transitions[(transition.a, transition.b)] = transition
274
def _transitionRemovedCb(self, track, transition, container):
275
del container.transitions[(transition.a, transition.b)]
277
def _setupTimeline(self, configuration):
278
for name, uri, props in configuration:
279
factory = self.project.sources.getUri(uri)
281
raise Exception("Could not find '%s' in sourcelist" %
287
timelineObject = self.timeline.addSourceFactory(factory)
288
setattr(self, name, timelineObject)
289
self.timelineObjects[name] = timelineObject
290
for trackObject in timelineObject.track_objects:
291
track = self.tracks[trackObject.track]
292
setattr(track, name, trackObject)
294
if not timelineObject:
295
raise Exception("Could not add source '%s' to timeline" %
297
for prop, value in props.iteritems():
298
setattr(timelineObject, prop, value)
299
self.emit("timeline-configured")
302
self.watchdog.start()
304
self.instance.run(["--no-ui"])
306
from pitivi.ui.zoominterface import Zoomable
307
# set a common zoom ratio so that things like edge snapping values
309
Zoomable.setZoomLevel((3 * Zoomable.zoom_steps) / 4)
310
self.instance.run([])
313
gobject.idle_add(self.instance.shutdown)
314
self.project._dirty = False
316
class Brush(Signallable):
317
"""Scrubs your timelines until they're squeaky clean."""
320
"scrub-step" : ["time", "priority"],
324
def __init__(self, runner, delay=100, maxtime=7200, maxpriority=10):
328
self.maxPriority = maxpriority
329
self.maxTime = maxtime
334
self.watchdog = runner.watchdog
336
def scrub(self, context, finalTime, finalPriority, steps=10):
337
self.context = context
338
self.time = finalTime
339
self.priority = finalPriority
342
gobject.timeout_add(self.delay, self._scrubTimeoutCb)
344
def _scrubTimeoutCb(self):
345
self.watchdog.keepAlive()
347
if self.count < self.steps:
348
time_ = random.randint(0, self.maxTime)
349
priority = random.randint(0, self.maxPriority)
350
self.context.editTo(time_, priority)
351
self.emit("scrub-step", time_, priority)
354
self.context.editTo(self.time, self.priority)
355
self.emit("scrub-step", self.time, self.priority)
356
self.context.finish()
357
self.emit("scrub-done")
360
class Base(TestCase):
362
Creates and runs an InteractivePitivi object, then starts the mainloop.
363
Uses a WatchDog to ensure that test cases will eventually terminate with an
364
assertion failure if runtime errors occur inside the mainloop."""
366
def run(self, result):
367
self._result = result
368
self._num_failures = len(result.failures)
369
self._num_errors = len(result.errors)
370
TestCase.run(self, result)
374
ptv = InteractivePitivi()
375
# was the pitivi object created
378
# were the contents of pitivi properly created
379
self.assertEqual(ptv.current, None)
381
# was the unique instance object properly set
382
self.assertEquals(pitivi.instance.PiTiVi, ptv)
385
# create an instance runner
386
self.runner = InstanceRunner(ptv)
389
# make sure we aren't exiting because our watchdog activated
390
self.assertFalse(self.runner.watchdog.activated)
391
# make sure the instance has been unset
393
if ((self._num_errors == self._result.errors) and
394
(self._num_failures == self._result.failures)):
395
will_fail = not (pitivi.instance.PiTiVi is None)
397
pitivi.instance.PiTiVi = None
402
raise Exception("Instance was not unset")
403
TestCase.tearDown(self)
405
class TestBasic(Base):
407
def testWatchdog(self):
409
self.assertTrue(self.runner.watchdog.activated)
410
self.runner.watchdog.activated = False
414
def newProjectLoaded(pitivi, project):
415
self.runner.shutDown()
417
self.ptv.connect("new-project-loaded", newProjectLoaded)
420
def testImport(self):
422
def sourcesLoaded(runner):
423
self.runner.shutDown()
425
config = Configuration()
426
config.addSource("test1", test1)
427
config.addSource("test2", test2)
428
config.addBadSource("test3", test3)
430
self.runner.connect("sources-loaded", sourcesLoaded)
431
self.runner.loadConfiguration(config)
434
self.assertFalse(hasattr(self.runner,test1))
435
self.assertFalse(hasattr(self.runner,test2))
436
self.failUnlessEqual(self.runner.factories, set((test1, test2)))
437
self.failUnlessEqual(self.runner.errors, set((test3,)))
439
def testConfigureTimeline(self):
441
config = Configuration()
447
"duration" : gst.SECOND,
448
"media-start" : gst.SECOND,
454
"start" : gst.SECOND,
455
"duration" : gst.SECOND,
458
def timelineConfigured(runner):
459
config.matches(self.runner)
460
self.runner.shutDown()
462
self.runner.loadConfiguration(config)
463
self.runner.connect("timeline-configured", timelineConfigured)
466
self.assertTrue(self.runner.object1)
467
self.assertTrue(self.runner.object2)
468
self.assertTrue(self.runner.video1.object1)
469
self.assertTrue(self.runner.audio1.object2)
471
def testMoveSources(self):
472
initial = Configuration()
478
"duration" : gst.SECOND,
479
"media-start" : gst.SECOND,
486
"start" : gst.SECOND,
487
"duration" : gst.SECOND,
490
final = Configuration()
495
"start" : 10 * gst.SECOND,
501
"start" : 11 * gst.SECOND,
505
def timelineConfigured(runner):
506
context = MoveContext(self.runner.timeline,
507
self.runner.video1.object1,
508
set((self.runner.audio1.object2,)))
509
brush.scrub(context, 10 * gst.SECOND, 1, steps=10)
511
def scrubStep(brush, time, priority):
514
def scrubDone(brush):
515
final.matches(self.runner)
516
self.runner.shutDown()
518
self.runner.loadConfiguration(initial)
519
self.runner.connect("timeline-configured", timelineConfigured)
521
brush = Brush(self.runner)
522
brush.connect("scrub-step", scrubStep)
523
brush.connect("scrub-done", scrubDone)
527
def testRippleMoveSimple(self):
529
initial = Configuration()
530
initial.addSource('clip1', test1, {
531
"duration" : gst.SECOND,
532
"start" : gst.SECOND,
534
initial.addSource('clip2', test1, {
535
"duration" : gst.SECOND,
536
"start" : 2 * gst.SECOND,
538
final = Configuration()
539
final.addSource('clip1', test1, {
540
"duration" : gst.SECOND,
541
"start" : 11 * gst.SECOND,
543
final.addSource('clip2', test1, {
544
"duration" : gst.SECOND,
545
"start" : 12 * gst.SECOND,
548
def timelineConfigured(runner):
549
initial.matches(self.runner)
550
context = MoveContext(self.runner.timeline,
551
self.runner.video1.clip1, set())
552
context.setMode(context.RIPPLE)
553
brush.scrub(context, 11 * gst.SECOND, 0, steps=0)
555
def scrubDone(brush):
556
final.matches(self.runner)
557
self.runner.shutDown()
559
self.runner.connect("timeline-configured", timelineConfigured)
560
brush = Brush(self.runner)
561
brush.connect("scrub-done", scrubDone)
563
self.runner.loadConfiguration(initial)
566
def testRippleTrimStartSimple(self):
567
initial = Configuration()
568
initial.addSource('clip1', test1,
570
"start" : gst.SECOND,
571
"duration" : gst.SECOND,
573
initial.addSource('clip2', test1,
575
"start" : 2 * gst.SECOND,
576
"duration" : gst.SECOND,
578
initial.addSource('clip3', test1,
580
"start" : 5 * gst.SECOND,
581
"duration" : 10 * gst.SECOND,
584
final = Configuration()
585
final.addSource('clip1', test1,
587
"start" : 6 * gst.SECOND,
588
"duration": gst.SECOND,
590
final.addSource('clip2', test1,
592
"start" : 7 * gst.SECOND,
593
"duration" : gst.SECOND,
595
final.addSource('clip3', test1,
597
"start" : 10 * gst.SECOND,
598
"duration" : 5 * gst.SECOND,
601
self.runner.loadConfiguration(initial)
602
def timelineConfigured(runner):
603
context = TrimStartContext(self.runner.timeline,
604
self.runner.video1.clip3, set())
605
context.setMode(context.RIPPLE)
606
brush.scrub(context, 10 * gst.SECOND, 0)
607
self.runner.connect("timeline-configured", timelineConfigured)
609
def scrubDone(brush):
610
final.matches(self.runner)
611
self.runner.shutDown()
613
brush = Brush(self.runner)
614
brush.connect("scrub-done", scrubDone)
617
from pitivi.pipeline import PipelineError
619
class TestSeeking(Base):
625
config = Configuration()
626
for i in xrange(0, 10):
627
config.addSource("clip%d" % i, test1, {
628
"start" : i * gst.SECOND,
629
"duration" : gst.SECOND,
634
def _startSeeking(self, interval, steps=10):
638
self.runner.project.pipeline.connect("position", self._positionCb)
639
gobject.timeout_add(interval, self._seekTimeoutCb)
641
def _seekTimeoutCb(self):
642
if self.count < self.steps:
643
self.runner.watchdog.keepAlive()
645
self.cur_pos = random.randint(0,
646
self.runner.timeline.duration)
647
self.runner.project.pipeline.seek(self.cur_pos)
649
self.failUnlessEqual(self.positions, self.count)
650
self.runner.shutDown()
653
def _positionCb(self, pipeline, position):
655
self.failUnlessEqual(position,
658
def testSeeking(self):
660
self.runner.loadConfiguration(self.config)
662
def timelineConfigured(runner):
663
self._startSeeking(100, 10)
665
def timelineConfiguredNoUI(runner):
666
self.runner.shutDown()
668
if self.runner.no_ui:
669
print "UI Disabled: Skipping Seeking Test. " \
670
"Use ENABLE_UI to test" \
672
self.runner.connect("timeline-configured", timelineConfiguredNoUI)
674
self.runner.connect("timeline-configured", timelineConfigured)
678
class TestRippleExtensive(Base):
680
"""Test suite for ripple editing minutia and corner-cases"""
682
def __init__(self, unknown):
683
# The following set of tests share common configuration, harness, and
684
# business logic. We create the configurations in the constructor to
685
# avoid having to re-create them for every test.
687
# create a seqence of adjacent clips in staggered formation, each one
689
self.initial = Configuration()
691
for i in xrange(0, 10):
692
self.initial.addSource('clip%d' % i, test1,
693
{ 'start' : gst.SECOND * i, 'duration' : gst.SECOND,
694
'priority' : i % 2 })
695
# we're going to repeat the same operation using each clip as the
696
# focus of the editing context. We create one final
697
# configuration for the expected result of each scenario.
698
final = Configuration()
699
for j in xrange(0, 10):
701
final.addSource('clip%d' % j, test1,
702
{ 'start' : gst.SECOND * j,
703
'duration' : gst.SECOND,
706
final.addSource('clip%d' % j, test1,
707
{ 'start' : gst.SECOND * (j + 10),
708
'duration' : gst.SECOND,
709
'priority' : (j % 2) + 1})
710
self.finals.append(final)
711
Base.__init__(self, unknown)
717
self.brush = Brush(self.runner)
718
self.runner.loadConfiguration(self.initial)
719
self.runner.connect("timeline-configured", self.timelineConfigured)
720
self.brush.connect("scrub-done", self.scenarioDone)
722
# when the timeline is configured, kick off the test by starting the
724
def timelineConfigured(self, runner):
727
# for each scenario, create the context using the specified clip as
728
# focus, and not specifying any other clips.
729
def nextScenario(self):
731
clipname = "clip%d" % cur
732
context = MoveContext(self.runner.timeline,
733
getattr(self.runner.video1, clipname), set())
735
context.setMode(context.RIPPLE)
736
self.context = context
737
# this isn't a method, but an attribute that will be set by specific
739
self.scrub_func(context, (cur + 10) * gst.SECOND, (cur % 2) + 1)
741
# when each scrub has finished, verify the current configuration is
742
# correct, reset the timeline, and kick off the next scenario. Shut down
743
# pitivi when we have finished the last scenario.
744
def scenarioDone(self, brush):
746
config = self.finals[cur]
747
context = self.context
749
config.matches(self.runner)
750
restore = MoveContext(self.runner.timeline, context.focus, set())
751
restore.setMode(restore.RIPPLE)
752
restore.editTo(cur * gst.SECOND, (cur % 2))
754
self.initial.matches(self.runner)
759
self.runner.shutDown()
761
def testRippleMoveComplex(self):
762
# in this test we move directly to the given position (steps=0)
763
def rippleMoveComplexScrubFunc(context, position, priority):
764
self.brush.scrub(context, position, priority, steps=0)
765
self.scrub_func = rippleMoveComplexScrubFunc
768
def testRippleMoveComplexRandom(self):
769
# same as above test, but scrub randomly (steps=100)
770
# FIXME: this test fails for unknown reasons
771
def rippleMoveComplexRandomScrubFunc(context, position, priority):
772
self.brush.scrub(context, position, priority, steps=100)
773
self.scrub_func = rippleMoveComplexRandomScrubFunc
776
class TestTransitions(Base):
778
def testSimple(self):
779
initial = Configuration()
785
"duration" : 5 * gst.SECOND,
792
"start" : 5 * gst.SECOND,
793
"duration" : 5 * gst.SECOND,
800
"start" : 10 * gst.SECOND,
801
"duration" : 5 * gst.SECOND,
811
("object2", "object3", 10 * gst.SECOND, 4 * gst.SECOND, 0),
812
("object1", "object2", 1 * gst.SECOND, 4 * gst.SECOND, 0),
815
def timelineConfigured(runner):
820
self._cur_move = moves.pop(0)
821
context = MoveContext(self.runner.timeline,
822
self.runner.video1.object2,
823
set([self.runner.video1.object2]))
824
brush.scrub(context, self._cur_move[0], self._cur_move[1], steps=10)
826
self.runner.shutDown()
828
def scrubDone(brush):
829
a, b, start, duration, priority = expected.pop(0)
830
a = getattr(self.runner.video1, a)
831
b = getattr(self.runner.video1, b)
833
tr = self.runner.video1.transitions[(a, b)]
835
self.failUnlessEqual(b.start, start)
836
self.failUnlessEqual(a.start + a.duration - start,
838
self.failUnlessEqual(tr.start, start)
839
self.failUnlessEqual(tr.duration, duration)
840
self.failUnlessEqual(tr.priority, 0)
841
self.failUnlessEqual(a.priority, 0)
842
self.failUnlessEqual(b.priority, 0)
845
self.runner.loadConfiguration(initial)
846
self.runner.connect("timeline-configured", timelineConfigured)
848
brush = Brush(self.runner)
849
brush.connect("scrub-done", scrubDone)
853
def testNoTransitionWhenMovingMultipleClips(self):
854
initial = Configuration()
860
"duration" : 5 * gst.SECOND,
867
"start" : 5 * gst.SECOND,
868
"duration" : 5 * gst.SECOND,
875
"start" : 10 * gst.SECOND,
876
"duration" : 5 * gst.SECOND,
881
("object1", 9 * gst.SECOND, 0),
882
("object3", 1* gst.SECOND, 0),
885
def timelineConfigured(runner):
890
self._cur_move = moves.pop(0)
891
other, start, priority = self._cur_move
892
context = MoveContext(self.runner.timeline,
893
self.runner.video1.object2,
894
set([getattr(self.runner.video1, other)]))
895
brush.scrub(context, start, priority, steps=10)
897
self.runner.shutDown()
899
def scrubDone(brush):
900
self.failUnlessEqual(self.runner.video1.transitions,
902
initial.matches(self.runner)
905
self.runner.loadConfiguration(initial)
906
self.runner.connect("timeline-configured", timelineConfigured)
908
brush = Brush(self.runner)
909
brush.connect("scrub-done", scrubDone)
913
def testOverlapOnlyWithValidTransitions(self):
914
initial = Configuration()
920
"duration" : 5 * gst.SECOND,
927
"start" : 5 * gst.SECOND,
928
"duration" : 3 * gst.SECOND,
935
"start" : 8 * gst.SECOND,
936
"duration" : 5 * gst.SECOND,
940
phase2 = initial.clone()
944
"start" : 4 * gst.SECOND,
947
phase3 = phase2.clone()
951
"duration" : 1 * gst.SECOND
954
phase4 = initial.clone()
958
"start" : 3 * gst.SECOND,
963
"start" : 5 * gst.SECOND,
964
"duration" : 5 * gst.SECOND,
968
# [1------] [3--[2==]]
969
(MoveContext, "object2", 9 * gst.SECOND, 0, initial, []),
971
# [1--[2=]] [3-------]
972
(MoveContext, "object2", 1 * gst.SECOND, 0, initial, []),
974
# [1------] [3-------]
976
(MoveContext, "object2", 4 * gst.SECOND, 0, phase2,
977
[("object1", "object2")]),
979
# Activates overlap prevention
984
(MoveContext, "object3", 3 * gst.SECOND, 0, phase2,
985
[("object1", "object2")]),
989
(TrimEndContext, "object3", 9 * gst.SECOND, 0, phase3,
990
[("object1", "object2")]),
992
# Activates overlap prevention
996
(MoveContext, "object3", 4 * gst.SECOND, 0, phase3,
997
[("object1", "object2")]),
999
# Activates overlap prevention
1003
(MoveContext, "object3", long(3.5 * gst.SECOND), 0, phase3,
1004
[("object1", "object2")]),
1006
# Activates overlap prevention
1010
(MoveContext, "object3", long(4.5 * gst.SECOND), 0,
1011
phase3, [("object1", "object2")]),
1013
# Next few commands build this arrangement
1018
(MoveContext, "object2", 3 * gst.SECOND, 0,
1020
(MoveContext, "object3", 5 * gst.SECOND, 0,
1022
(TrimEndContext, "object3", 10 * gst.SECOND, 0,
1023
phase4, [("object1", "object2"), ("object2",
1026
# Activates Overlap Prevention
1031
(MoveContext, "object3", 4 * gst.SECOND, 0,
1032
phase4, [("object1", "object2"),
1033
("object2", "object3")]),
1039
def timelineConfigured(runner):
1044
print "cur_move: %d/%d" % (nmoves - len(moves) + 1, nmoves)
1045
self._cur_move = moves.pop(0)
1046
context, focus, start, priority, config, trans = self._cur_move
1047
obj = getattr(self.runner.video1, focus)
1048
context = context(self.runner.timeline,
1050
brush.scrub(context, start, priority, steps=10)
1052
self.runner.shutDown()
1054
def scrubDone(brush):
1055
connect, focus, stream, priority, config, trans = self._cur_move
1058
config.matches(self.runner)
1061
expected = set([(getattr(self.runner.video1, a),
1062
getattr(self.runner.video1, b)) for a, b in
1065
self.failUnlessEqual(set(self.runner.video1.transitions.keys()),
1069
self.runner.loadConfiguration(initial)
1070
self.runner.connect("timeline-configured", timelineConfigured)
1072
brush = Brush(self.runner)
1073
brush.connect("scrub-done", scrubDone)
1077
def testSaveAndLoadWithTransitions(self):
added
removed
tests/test_integration.py
