a180841fbc415df0c82628b33add488c5b59532f
[platform/upstream/gst-editing-services.git] / tests / check / python / common.py
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright (c) 2016 Alexandru Băluț <alexandru.balut@gmail.com>
4 # Copyright (c) 2016, Thibault Saunier
5 #
6 # This program is free software; you can redistribute it and/or
7 # modify it under the terms of the GNU Lesser General Public
8 # License as published by the Free Software Foundation; either
9 # version 2.1 of the License, or (at your option) any later version.
10 #
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 # Lesser General Public License for more details.
15 #
16 # You should have received a copy of the GNU Lesser General Public
17 # License along with this program; if not, write to the
18 # Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 # Boston, MA 02110-1301, USA.
20
21 import gi
22
23 gi.require_version("Gst", "1.0")
24 gi.require_version("GES", "1.0")
25
26 from gi.repository import Gst  # noqa
27 from gi.repository import GES  # noqa
28 from gi.repository import GLib  # noqa
29 import contextlib  # noqa
30 import os  #noqa
31 import unittest  # noqa
32 import tempfile  # noqa
33
34 Gst.init(None)
35 GES.init()
36
37
38 def create_main_loop():
39     """Creates a MainLoop with a timeout."""
40     mainloop = GLib.MainLoop()
41     timed_out = False
42
43     def timeout_cb(unused):
44         nonlocal timed_out
45         timed_out = True
46         mainloop.quit()
47
48     def run(timeout_seconds=5, until_empty=False):
49         source = GLib.timeout_source_new_seconds(timeout_seconds)
50         source.set_callback(timeout_cb)
51         source.attach()
52         if until_empty:
53             GLib.idle_add(mainloop.quit)
54         GLib.MainLoop.run(mainloop)
55         source.destroy()
56         if timed_out:
57             raise Exception("Timed out after %s seconds" % timeout_seconds)
58
59     mainloop.run = run
60     return mainloop
61
62
63 def create_project(with_group=False, saved=False):
64     """Creates a project with two clips in a group."""
65     project = GES.Project()
66     timeline = project.extract()
67     layer = timeline.append_layer()
68
69     if with_group:
70         clip1 = GES.TitleClip()
71         clip1.set_start(0)
72         clip1.set_duration(10)
73         layer.add_clip(clip1)
74         clip2 = GES.TitleClip()
75         clip2.set_start(100)
76         clip2.set_duration(10)
77         layer.add_clip(clip2)
78         group = GES.Container.group([clip1, clip2])
79
80     if saved:
81         uri = "file://%s" % tempfile.NamedTemporaryFile(suffix=".xges").name
82         project.save(timeline, uri, None, overwrite=True)
83
84     return timeline
85
86
87 @contextlib.contextmanager
88 def created_project_file(xges):
89     _, xges_path = tempfile.mkstemp(suffix=".xges")
90     with open(xges_path, "w") as f:
91         f.write(xges)
92
93     yield Gst.filename_to_uri(os.path.abspath(xges_path))
94
95     os.remove(xges_path)
96
97
98 def get_asset_uri(name):
99     python_tests_dir = os.path.dirname(os.path.abspath(__file__))
100     assets_dir = os.path.join(python_tests_dir, "..", "assets")
101     return Gst.filename_to_uri(os.path.join(assets_dir, name))
102
103
104 class GESTest(unittest.TestCase):
105
106     def _log(self, func, format, *args):
107         string = format
108         if args:
109             string = string % args[0]
110         func(string)
111
112     def log(self, format, *args):
113         self._log(Gst.log, format, *args)
114
115     def debug(self, format, *args):
116         self._log(Gst.debug, format, *args)
117
118     def info(self, format, *args):
119         self._log(Gst.info, format, *args)
120
121     def fixme(self, format, *args):
122         self._log(Gst.fixme, format, *args)
123
124     def warning(self, format, *args):
125         self._log(Gst.warning, format, *args)
126
127     def error(self, format, *args):
128         self._log(Gst.error, format, *args)
129
130     def check_clip_values(self, clip, start, in_point, duration):
131         for elem in [clip] + clip.get_children(False):
132             self.check_element_values(elem, start, in_point, duration)
133
134     def check_element_values(self, element, start, in_point, duration):
135         self.assertEqual(element.props.start, start, element)
136         self.assertEqual(element.props.in_point, in_point, element)
137         self.assertEqual(element.props.duration, duration, element)
138
139     def assert_effects(self, clip, *effects):
140         # Make sure there are no other effects.
141         self.assertEqual(set(clip.get_top_effects()), set(effects))
142
143         # Make sure their order is correct.
144         indexes = [clip.get_top_effect_index(effect)
145                    for effect in effects]
146         self.assertEqual(indexes, list(range(len(effects))))
147
148
149 class GESSimpleTimelineTest(GESTest):
150
151     def __init__(self, *args):
152         self.track_types = [GES.TrackType.AUDIO, GES.TrackType.VIDEO]
153         super(GESSimpleTimelineTest, self).__init__(*args)
154
155     def setUp(self):
156         self.timeline = GES.Timeline.new()
157         for track_type in self.track_types:
158             self.assertIn(
159                 track_type, [GES.TrackType.AUDIO, GES.TrackType.VIDEO])
160             if track_type == GES.TrackType.AUDIO:
161                 self.assertTrue(self.timeline.add_track(GES.AudioTrack.new()))
162             else:
163                 self.assertTrue(self.timeline.add_track(GES.VideoTrack.new()))
164
165         self.assertEqual(len(self.timeline.get_tracks()),
166                          len(self.track_types))
167         self.layer = self.timeline.append_layer()
168
169     def add_clip(self, start, in_point, duration):
170         clip = GES.TestClip()
171         clip.props.start = start
172         clip.props.in_point = in_point
173         clip.props.duration = duration
174         self.assertTrue(self.layer.add_clip(clip))
175
176         return clip
177