gstreamer/tests/check/python/common.py
Henry Wilkes 142456d8ba errors: added edit errors
Added more errors to GES_ERROR for when edits fail (other than
programming or usage errors). Also promoted some GST messages if they
related to a usage error.

Also added explanation of timeline overlap rules in user docs.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-editing-services/-/merge_requests/177>
2020-05-22 19:15:57 +01:00

735 lines
27 KiB
Python

# -*- coding: utf-8 -*-
#
# Copyright (c) 2016 Alexandru Băluț <alexandru.balut@gmail.com>
# Copyright (c) 2016, Thibault Saunier
#
# This program 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 program 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 program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301, USA.
import gi
gi.require_version("Gst", "1.0")
gi.require_version("GES", "1.0")
from gi.repository import Gst # noqa
from gi.repository import GES # noqa
from gi.repository import GLib # noqa
from gi.repository import GObject # noqa
import contextlib # noqa
import os #noqa
import unittest # noqa
import tempfile # noqa
Gst.init(None)
GES.init()
def create_main_loop():
"""Creates a MainLoop with a timeout."""
mainloop = GLib.MainLoop()
timed_out = False
def timeout_cb(unused):
nonlocal timed_out
timed_out = True
mainloop.quit()
def run(timeout_seconds=5, until_empty=False):
source = GLib.timeout_source_new_seconds(timeout_seconds)
source.set_callback(timeout_cb)
source.attach()
if until_empty:
GLib.idle_add(mainloop.quit)
GLib.MainLoop.run(mainloop)
source.destroy()
if timed_out:
raise Exception("Timed out after %s seconds" % timeout_seconds)
mainloop.run = run
return mainloop
def create_project(with_group=False, saved=False):
"""Creates a project with two clips in a group."""
timeline = GES.Timeline.new_audio_video()
layer = timeline.append_layer()
if with_group:
clip1 = GES.TitleClip()
clip1.set_start(0)
clip1.set_duration(10*Gst.SECOND)
layer.add_clip(clip1)
clip2 = GES.TitleClip()
clip2.set_start(100 * Gst.SECOND)
clip2.set_duration(10*Gst.SECOND)
layer.add_clip(clip2)
group = GES.Container.group([clip1, clip2])
if saved:
if isinstance(saved, str):
suffix = "-%s.xges" % saved
else:
suffix = ".xges"
uri = "file://%s" % tempfile.NamedTemporaryFile(suffix=suffix).name
timeline.get_asset().save(timeline, uri, None, overwrite=True)
return timeline
@contextlib.contextmanager
def created_project_file(xges):
_, xges_path = tempfile.mkstemp(suffix=".xges")
with open(xges_path, "w") as f:
f.write(xges)
yield Gst.filename_to_uri(os.path.abspath(xges_path))
os.remove(xges_path)
def get_asset_uri(name):
python_tests_dir = os.path.dirname(os.path.abspath(__file__))
assets_dir = os.path.join(python_tests_dir, "..", "assets")
return Gst.filename_to_uri(os.path.join(assets_dir, name))
class GESTest(unittest.TestCase):
def _log(self, func, format, *args):
string = format
if args:
string = string % args[0]
func(string)
def log(self, format, *args):
self._log(Gst.log, format, *args)
def debug(self, format, *args):
self._log(Gst.debug, format, *args)
def info(self, format, *args):
self._log(Gst.info, format, *args)
def fixme(self, format, *args):
self._log(Gst.fixme, format, *args)
def warning(self, format, *args):
self._log(Gst.warning, format, *args)
def error(self, format, *args):
self._log(Gst.error, format, *args)
def check_clip_values(self, clip, start, in_point, duration):
for elem in [clip] + clip.get_children(False):
self.check_element_values(elem, start, in_point, duration)
def check_element_values(self, element, start, in_point, duration):
self.assertEqual(element.props.start, start, element)
self.assertEqual(element.props.in_point, in_point, element)
self.assertEqual(element.props.duration, duration, element)
def assert_effects(self, clip, *effects):
# Make sure there are no other effects.
self.assertEqual(set(clip.get_top_effects()), set(effects))
# Make sure their order is correct.
indexes = [clip.get_top_effect_index(effect)
for effect in effects]
self.assertEqual(indexes, list(range(len(effects))))
def assertGESError(self, error, code, message=""):
if error is None:
raise AssertionError(
"{}{}Received no error".format(message, message and ": "))
if error.domain != "GES_ERROR":
raise AssertionError(
"{}{}Received error ({}) in domain {} rather than "
"GES_ERROR".format(
message, message and ": ", error.message, error.domain))
err_code = GES.Error(error.code)
if err_code != code:
raise AssertionError(
"{}{}Received {} error ({}) rather than {}".format(
message, message and ": ", err_code.value_name,
error.message, code.value_name))
class GESSimpleTimelineTest(GESTest):
def __init__(self, *args):
self.track_types = [GES.TrackType.AUDIO, GES.TrackType.VIDEO]
super(GESSimpleTimelineTest, self).__init__(*args)
def timeline_as_str(self):
res = "====== %s =======\n" % self.timeline
for layer in self.timeline.get_layers():
res += "Layer %04d: " % layer.get_priority()
for clip in layer.get_clips():
res += "{ %s }" % clip
res += '\n------------------------\n'
for group in self.timeline.get_groups():
res += "GROUP %s :" % group
for clip in group.get_children(False):
res += " { %s }" % clip.props.name
res += '\n'
res += "================================\n"
return res
def print_timeline(self):
print(self.timeline_as_str())
def setUp(self):
self.timeline = GES.Timeline.new()
for track_type in self.track_types:
self.assertIn(
track_type, [GES.TrackType.AUDIO, GES.TrackType.VIDEO])
if track_type == GES.TrackType.AUDIO:
self.assertTrue(self.timeline.add_track(GES.AudioTrack.new()))
else:
self.assertTrue(self.timeline.add_track(GES.VideoTrack.new()))
self.assertEqual(len(self.timeline.get_tracks()),
len(self.track_types))
self.layer = self.timeline.append_layer()
def add_clip(self, start, in_point, duration, asset_type=GES.TestClip):
clip = GES.Asset.request(asset_type, None).extract()
clip.props.start = start
clip.props.in_point = in_point
clip.props.duration = duration
self.assertTrue(self.layer.add_clip(clip))
return clip
def append_clip(self, layer=0, asset_type=GES.TestClip):
while len(self.timeline.get_layers()) < layer + 1:
self.timeline.append_layer()
layer = self.timeline.get_layers()[layer]
clip = GES.Asset.request(asset_type, None).extract()
clip.props.start = layer.get_duration()
clip.props.duration = 10
self.assertTrue(layer.add_clip(clip))
return clip
def assertElementAreEqual(self, ref, element):
self.assertTrue(isinstance(element, type(ref)), "%s and %s do not have the same type!" % (ref, element))
props = [p for p in ref.list_properties() if p.name not in ['name']
and not GObject.type_is_a(p.value_type, GObject.Object)]
for p in props:
pname = p.name
v0 = GObject.Value()
v0.init(p.value_type)
v0.set_value(ref.get_property(pname))
v1 = GObject.Value()
v1.init(p.value_type)
v1.set_value(element.get_property(pname))
self.assertTrue(Gst.value_compare(v0, v1) == Gst.VALUE_EQUAL,
"%s are not equal: %s != %s" % (pname, v0, v1))
if isinstance(ref, GES.TrackElement):
self.assertElementAreEqual(ref.get_nleobject(), element.get_nleobject())
return
if not isinstance(ref, GES.Clip):
return
ttypes = [track.type for track in self.timeline.get_tracks()]
for ttype in ttypes:
if ttypes.count(ttype) > 1:
self.warning("Can't deeply check %s and %s "
"(only one track per type supported %s %s found)" % (ref,
element, ttypes.count(ttype), ttype))
return
children = element.get_children(False)
for ref_child in ref.get_children(False):
ref_track = ref_child.get_track()
if not ref_track:
self.warning("Can't check %s as not in a track" % (ref_child))
continue
child = None
for tmpchild in children:
if not isinstance(tmpchild, type(ref_child)):
continue
if ref_track.type != tmpchild.get_track().type:
continue
if not isinstance(ref_child, GES.Effect):
child = tmpchild
break
elif ref_child.props.bin_description == child.props.bin_description:
child = tmpchild
break
self.assertIsNotNone(child, "Could not find equivalent child %s in %s(%s)" % (ref_child,
element, children))
self.assertElementAreEqual(ref_child, child)
def check_reload_timeline(self):
tmpf = tempfile.NamedTemporaryFile(suffix='.xges')
uri = Gst.filename_to_uri(tmpf.name)
self.assertTrue(self.timeline.save_to_uri(uri, None, True))
project = GES.Project.new(uri)
mainloop = create_main_loop()
def loaded_cb(unused_project, unused_timeline):
mainloop.quit()
project.connect("loaded", loaded_cb)
reloaded_timeline = project.extract()
mainloop.run()
self.assertIsNotNone(reloaded_timeline)
layers = self.timeline.get_layers()
reloaded_layers = reloaded_timeline.get_layers()
self.assertEqual(len(layers), len(reloaded_layers))
for layer, reloaded_layer in zip(layers, reloaded_layers):
clips = layer.get_clips()
reloaded_clips = reloaded_layer.get_clips()
self.assertEqual(len(clips), len(reloaded_clips))
for clip, reloaded_clip in zip(clips, reloaded_clips):
self.assertElementAreEqual(clip, reloaded_clip)
return reloaded_timeline
def assertTimelineTopology(self, topology, groups=[]):
res = []
for layer in self.timeline.get_layers():
layer_timings = []
for clip in layer.get_clips():
layer_timings.append(
(type(clip), clip.props.start, clip.props.duration))
for child in clip.get_children(True):
self.assertEqual(child.props.start, clip.props.start)
self.assertEqual(child.props.duration, clip.props.duration)
res.append(layer_timings)
if topology != res:
Gst.error(self.timeline_as_str())
self.assertEqual(topology, res)
timeline_groups = self.timeline.get_groups()
if groups and timeline_groups:
for i, group in enumerate(groups):
self.assertEqual(set(group), set(timeline_groups[i].get_children(False)))
self.assertEqual(len(timeline_groups), i + 1)
return res
class GESTimelineConfigTest(GESTest):
"""
Tests where all the configuration changes, snapping positions and
auto-transitions are accounted for.
"""
def setUp(self):
timeline = GES.Timeline.new()
self.timeline = timeline
timeline.set_auto_transition(True)
self.snap_occured = False
self.snap = None
def snap_started(tl, el1, el2, pos):
if self.snap_occured:
raise AssertionError(
"Previous snap {} not accounted for".format(self.snap))
self.snap_occured = True
if self.snap is not None:
raise AssertionError(
"Previous snap {} not ended".format(self.snap))
self.snap = (el1.get_parent(), el2.get_parent(), pos)
def snap_ended(tl, el1, el2, pos):
self.assertEqual(
self.snap, (el1.get_parent(), el2.get_parent(), pos))
self.snap = None
timeline.connect("snapping-started", snap_started)
timeline.connect("snapping-ended", snap_ended)
self.lost_clips = []
def unrecord_lost_clip(layer, clip):
if clip in self.lost_clips:
self.lost_clips.remove(clip)
def record_lost_clip(layer, clip):
self.lost_clips.append(clip)
def layer_added(tl, layer):
layer.connect("clip-added", unrecord_lost_clip)
layer.connect("clip-removed", record_lost_clip)
timeline.connect("layer-added", layer_added)
self.clips = []
self.auto_transitions = {}
self.config = {}
@staticmethod
def new_config(start, duration, inpoint, maxduration, layer):
return {"start": start, "duration": duration, "in-point": inpoint,
"max-duration": maxduration, "layer": layer}
def add_clip(self, name, layer, tracks, start, duration, inpoint=0,
maxduration=Gst.CLOCK_TIME_NONE, clip_type=GES.TestClip,
asset_id=None, effects=None):
"""
Create a clip with the given @name and properties and add it to the
layer of priority @layer to the tracks in @tracks. Also registers
its expected configuration.
"""
if effects is None:
effects = []
lay = self.timeline.get_layer(layer)
while lay is None:
self.timeline.append_layer()
lay = self.timeline.get_layer(layer)
asset = GES.Asset.request(clip_type, asset_id)
clip = asset.extract()
self.assertTrue(clip.set_name(name))
# FIXME: would be better to use select-tracks-for-object
# hack around the fact that we cannot use select-tracks-for-object
# in python by setting start to large number to ensure no conflict
# when adding a clip
self.assertTrue(clip.set_start(10000))
self.assertTrue(clip.set_duration(duration))
self.assertTrue(clip.set_inpoint(inpoint))
for effect in effects:
self.assertTrue(clip.add(effect))
if lay.add_clip(clip) != True:
raise AssertionError(
"Failed to add clip {} to layer {}".format(name, layer))
# then remove the children not in the selected tracks, which may
# now allow some clips to fully/triple overlap because they do
# not share a track
for child in clip.get_children(False):
if child.get_track() not in tracks:
clip.remove(child)
# then move to the desired start
prev_snap = self.timeline.get_snapping_distance()
self.timeline.set_snapping_distance(0)
self.assertTrue(clip.set_start(start))
self.timeline.set_snapping_distance(prev_snap)
self.assertTrue(clip.set_max_duration(maxduration))
self.config[clip] = self.new_config(
start, duration, inpoint, maxduration, layer)
self.clips.append(clip)
return clip
def add_group(self, name, to_group):
"""
Create a group with the given @name and the elements in @to_group.
Also registers its expected configuration.
"""
group = GES.Group.new()
self.assertTrue(group.set_name(name))
start = None
end = None
layer = None
for element in to_group:
if start is None:
start = element.start
end = element.start + element.duration
layer = element.get_layer_priority()
else:
start = min(start, element.start)
end = max(end, element.start + element.duration)
layer = min(layer, element.get_layer_priority())
self.assertTrue(group.add(element))
self.config[group] = self.new_config(
start, end - start, 0, Gst.CLOCK_TIME_NONE, layer)
return group
def register_auto_transition(self, clip1, clip2, track):
"""
Register that we expect an auto-transition to exist between
@clip1 and @clip2 in @track.
"""
transition = self._find_transition(clip1, clip2, track)
if transition is None:
raise AssertionError(
"{} and {} have no auto-transition in track {}".format(
clip1, clip2, track))
if transition in self.auto_transitions.values():
raise AssertionError(
"Auto-transition between {} and {} in track {} already "
"registered".format(clip1, clip2, track))
key = (clip1, clip2, track)
if key in self.auto_transitions:
raise AssertionError(
"Auto-transition already registered for {}".format(key))
self.auto_transitions[key] = transition
def add_video_track(self):
track = GES.VideoTrack.new()
self.assertTrue(self.timeline.add_track(track))
return track
def add_audio_track(self):
track = GES.AudioTrack.new()
self.assertTrue(self.timeline.add_track(track))
return track
def assertElementConfig(self, element, config):
for prop in config:
if prop == "layer":
val = element.get_layer_priority()
else:
val = element.get_property(prop)
if val != config[prop]:
raise AssertionError("{} property {}: {} != {}".format(
element, prop, val, config[prop]))
@staticmethod
def _source_in_track(clip, track):
if clip.find_track_element(track, GES.Source):
return True
return False
def _find_transition(self, clip1, clip2, track):
"""find transition from earlier clip1 to later clip2"""
if not self._source_in_track(clip1, track) or \
not self._source_in_track(clip2, track):
return None
layer_prio = clip1.get_layer_priority()
if layer_prio != clip2.get_layer_priority():
return None
if clip1.start >= clip2.start:
return None
start = clip2.start
end = clip1.start + clip1.duration
if start >= end:
return None
duration = end - start
layer = self.timeline.get_layer(layer_prio)
self.assertIsNotNone(layer)
for clip in layer.get_clips():
children = clip.get_children(False)
if len(children) == 1:
child = children[0]
else:
continue
if isinstance(clip, GES.TransitionClip) and clip.start == start \
and clip.duration == duration and child.get_track() == track:
return clip
raise AssertionError(
"No auto-transition between {} and {} in track {}".format(
clip1, clip2, track))
def _transition_between(self, new, existing, clip1, clip2, track):
if clip1.start < clip2.start:
entry = (clip1, clip2, track)
else:
entry = (clip2, clip1, track)
trans = self._find_transition(*entry)
if trans is None:
return
if entry in new:
new.remove(entry)
self.auto_transitions[entry] = trans
elif entry in existing:
existing.remove(entry)
expect = self.auto_transitions[entry]
if trans != expect:
raise AssertionError(
"Auto-transition between {} and {} in track {} changed "
"from {} to {}".format(
clip1, clip2, track, expect, trans))
else:
raise AssertionError(
"Unexpected transition found between {} and {} in track {}"
"".format(clip1, clip2, track))
def assertTimelineConfig(
self, new_props=None, snap_position=None, snap_froms=None,
snap_tos=None, new_transitions=None, lost_transitions=None):
"""
Check that the timeline configuration has only changed by the
differences present in @new_props.
Check that a snap occurred at @snap_position between one of the
clips in @snap_froms and one of the clips in @snap_tos.
Check that all new transitions in the timeline are present in
@new_transitions.
Checl that all the transitions that were lost are in
@lost_transitions.
"""
if new_props is None:
new_props = {}
if snap_froms is None:
snap_froms = []
if snap_tos is None:
snap_tos = []
if new_transitions is None:
new_transitions = []
if lost_transitions is None:
lost_transitions = []
for element, config in new_props.items():
if element not in self.config:
self.config[element] = {}
for prop in config:
self.config[element][prop] = new_props[element][prop]
for element, config in self.config.items():
self.assertElementConfig(element, config)
# check that snapping occurred
snaps = []
for snap_from in snap_froms:
for snap_to in snap_tos:
snaps.append((snap_from, snap_to, snap_position))
if self.snap is None:
if snaps:
raise AssertionError(
"No snap occurred, but expected a snap in {}".format(snaps))
elif not snaps:
if self.snap_occured:
raise AssertionError(
"Snap {} occurred, but expected no snap".format(self.snap))
elif self.snap not in snaps:
raise AssertionError(
"Snap {} occurred, but expected a snap in {}".format(
self.snap, snaps))
self.snap_occured = False
# check that lost transitions are not part of the layer
for clip1, clip2, track in lost_transitions:
key = (clip1, clip2, track)
if key not in self.auto_transitions:
raise AssertionError(
"No such auto-transition between {} and {} in track {} "
"is registered".format(clip1, clip2, track))
# make sure original transition was removed from the layer
trans = self.auto_transitions[key]
if trans not in self.lost_clips:
raise AssertionError(
"The auto-transition {} between {} and {} track {} was "
"not removed from the layers, but expect it to be lost"
"".format(trans, clip1, clip2, track))
self.lost_clips.remove(trans)
# make sure a new one wasn't created
trans = self._find_transition(clip1, clip2, track)
if trans is not None:
raise AssertionError(
"Found auto-transition between {} and {} in track {} "
"is present, but expected it to be lost".format(
clip1, clip2, track))
# since it was lost, remove it
del self.auto_transitions[key]
# check that all lost clips are accounted for
if self.lost_clips:
raise AssertionError(
"Clips were lost that are not accounted for: {}".format(
self.lost_clips))
# check that all other transitions are either existing ones or
# new ones
new = set(new_transitions)
existing = set(self.auto_transitions.keys())
for i, clip1 in enumerate(self.clips):
for clip2 in self.clips[i+1:]:
for track in self.timeline.get_tracks():
self._transition_between(
new, existing, clip1, clip2, track)
# make sure we are not missing any expected transitions
if new:
raise AssertionError(
"Did not find new transitions for {}".format(new))
if existing:
raise AssertionError(
"Did not find existing transitions for {}".format(existing))
# make sure there aren't any clips we are unaware of
transitions = self.auto_transitions.values()
for layer in self.timeline.get_layers():
for clip in layer.get_clips():
if clip not in self.clips and clip not in transitions:
raise AssertionError("Unknown clip {}".format(clip))
def assertEdit(self, element, layer, mode, edge, position, snap,
snap_froms, snap_tos, new_props, new_transitions,
lost_transitions):
if not element.edit_full(layer, mode, edge, position):
raise AssertionError(
"Edit of {} to layer {}, mode {}, edge {}, at position {} "
"failed when a success was expected".format(
element, layer, mode, edge, position))
self.assertTimelineConfig(
new_props=new_props, snap_position=snap, snap_froms=snap_froms,
snap_tos=snap_tos, new_transitions=new_transitions,
lost_transitions=lost_transitions)
def assertFailEdit(self, element, layer, mode, edge, position, err_code):
res = None
error = None
try:
res = element.edit_full(layer, mode, edge, position)
except GLib.Error as exception:
error = exception
if err_code is None:
if res is not False:
raise AssertionError(
"Edit of {} to layer {}, mode {}, edge {}, at "
"position {} succeeded when a failure was expected"
"".format(
element, layer, mode, edge, position))
if error is not None:
raise AssertionError(
"Edit of {} to layer {}, mode {}, edge {}, at "
"position {} did produced an error when none was "
"expected".format(
element, layer, mode, edge, position))
else:
self.assertGESError(
error, err_code,
"Edit of {} to layer {}, mode {}, edge {}, at "
"position {}".format(element, layer, mode, edge, position))
# should be no change or snapping if edit fails
self.assertTimelineConfig()