3 # Copyright (c) 2013,Thibault Saunier <thibault.saunier@collabora.com>
5 # This program is free software; you can redistribute it and/or
6 # modify it under the terms of the GNU Lesser General Public
7 # License as published by the Free Software Foundation; either
8 # version 2.1 of the License, or (at your option) any later version.
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 # Lesser General Public License for more details.
15 # You should have received a copy of the GNU Lesser General Public
16 # License along with this program; if not, write to the
17 # Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
18 # Boston, MA 02110-1301, USA.
32 from launcher.loggable import Loggable, error
34 from launcher.baseclasses import GstValidateTest, Test, \
35 ScenarioManager, NamedDic, GstValidateTestsGenerator, \
36 GstValidateMediaDescriptor, GstValidateEncodingTestInterface, \
37 GstValidateBaseTestManager, MediaDescriptor, MediaFormatCombination
39 from launcher.utils import path2url, url2path, DEFAULT_TIMEOUT, which, \
40 GST_SECOND, Result, Protocols, mkdir, printc, Colors, get_data_file, \
41 kill_subprocess, format_config_template, get_fakesink_for_media_type, \
42 parse_gsttimeargs, GstCaps
45 # Private global variables #
48 # definitions of commands to use
49 parser = argparse.ArgumentParser(add_help=False)
50 parser.add_argument("--validate-tools-path", dest="validate_tools_path",
52 help="defines the paths to look for GstValidate tools.")
53 options, args = parser.parse_known_args()
55 GstValidateBaseTestManager.update_commands(options.validate_tools_path)
56 AUDIO_ONLY_FILE_TRANSCODING_RATIO = 5
59 # API to be used to create testsuites #
63 Some info about protocols and how to handle them
65 GST_VALIDATE_CAPS_TO_PROTOCOL = [("application/x-hls", Protocols.HLS),
66 ("application/dash+xml", Protocols.DASH)]
69 def expand_vars_in_list_recurse(l, data):
70 for i, v in enumerate(l):
71 if isinstance(v, dict):
72 l[i] = expand_vars_in_dict_recurse(v, data)
73 elif isinstance(v, str):
75 elif isinstance(v, list):
76 l[i] = expand_vars_in_list_recurse(v, data)
81 def expand_vars_in_dict_recurse(dico, data):
82 for key, value in dico.items():
83 if isinstance(value, dict):
84 dico[key] = expand_vars_in_dict_recurse(value, data)
85 elif isinstance(value, str):
86 dico[key] = value % data
87 elif isinstance(value, list):
88 dico[key] = expand_vars_in_list_recurse(value, data)
93 class GstValidateMediaCheckTestsGenerator(GstValidateTestsGenerator):
95 def __init__(self, test_manager):
96 GstValidateTestsGenerator.__init__(self, "media_check", test_manager)
98 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
99 for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
100 protocol = mediainfo.media_descriptor.get_protocol()
101 timeout = DEFAULT_TIMEOUT
103 classname = "%s.media_check.%s" % (protocol,
104 os.path.basename(url2path(uri)).replace(".", "_"))
105 self.add_test(GstValidateMediaCheckTest(classname,
106 self.test_manager.options,
107 self.test_manager.reporter,
108 mediainfo.media_descriptor,
114 class GstValidateTranscodingTestsGenerator(GstValidateTestsGenerator):
115 HARD_TIMEOUT_FACTOR = 10
117 def __init__(self, test_manager):
118 GstValidateTestsGenerator.__init__(self, "transcode", test_manager)
120 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
121 for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
122 if mediainfo.media_descriptor.is_image():
125 protocol = mediainfo.media_descriptor.get_protocol()
126 if protocol == Protocols.RTSP:
129 options = self.test_manager.options
130 for comb in self.test_manager.get_encoding_formats():
131 classname = "%s.transcode.to_%s.%s" % (mediainfo.media_descriptor.get_protocol(),
134 mediainfo.media_descriptor.get_clean_name())
136 self.add_test(GstValidateTranscodingTest(classname,
138 self.test_manager.reporter,
141 mediainfo.media_descriptor))
144 class FakeMediaDescriptor(MediaDescriptor):
146 def __init__(self, infos, pipeline_desc):
147 MediaDescriptor.__init__(self)
149 self._pipeline_desc = pipeline_desc
152 return self._infos.get('path', None)
154 def get_tracks_caps(self):
155 return self._info.get('tracks-caps', [])
157 def get_media_filepath(self):
158 return self._infos.get('media-filepath', None)
161 return self._infos.get('caps', None)
164 return self._infos.get('uri', None)
166 def get_duration(self):
167 return int(self._infos.get('duration', 0)) * GST_SECOND
169 def get_protocol(self):
170 return self._infos.get('protocol', "launch_pipeline")
172 def is_seekable(self):
173 return self._infos.get('is-seekable', True)
176 return self._infos.get('is-image', False)
179 return self._infos.get('is-live', False)
181 def get_num_tracks(self, track_type):
182 return self._infos.get('num-%s-tracks' % track_type,
183 self._pipeline_desc.count(track_type + "sink"))
185 def can_play_reverse(self):
186 return self._infos.get('plays-reverse', False)
189 class GstValidateSimpleTestsGenerator(GstValidateTestsGenerator):
190 def __init__(self, name, test_manager, tests_dir):
191 self.tests_dir = tests_dir
192 super().__init__(name, test_manager)
194 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
195 for root, _, files in os.walk(self.tests_dir):
197 name, ext = os.path.splitext(f)
198 if ext != ".validatetest":
201 fpath = os.path.abspath(os.path.join(root, f))
202 pathname = os.path.abspath(os.path.join(root, name))
203 name = pathname.replace(os.path.commonpath([self.tests_dir, root]), '').replace('/', '.')
204 self.add_test(GstValidateSimpleTest(fpath, 'test' + name,
205 self.test_manager.options,
206 self.test_manager.reporter))
209 class GstValidatePipelineTestsGenerator(GstValidateTestsGenerator):
211 def __init__(self, name, test_manager, pipeline_template=None,
212 pipelines_descriptions=None, valid_scenarios=None):
214 @name: The name of the generator
215 @pipeline_template: A template pipeline to be used to generate actual pipelines
216 @pipelines_descriptions: A list of tuple of the form:
217 (test_name, pipeline_description, extra_data)
218 extra_data being a dictionary with the following keys:
219 'scenarios': ["the", "valide", "scenarios", "names"]
220 'duration': the_duration # in seconds
221 'timeout': a_timeout # in seconds
222 'hard_timeout': a_hard_timeout # in seconds
224 @valid_scenarios: A list of scenario name that can be used with that generator
226 valid_scenarios = valid_scenarios or []
227 GstValidateTestsGenerator.__init__(self, name, test_manager)
228 self._pipeline_template = pipeline_template
229 self._pipelines_descriptions = []
230 for description in pipelines_descriptions or []:
231 if not isinstance(description, dict):
232 desc_dict = {"name": description[0],
233 "pipeline": description[1]}
234 if len(description) >= 3:
235 desc_dict["extra_data"] = description[2]
236 self._pipelines_descriptions.append(desc_dict)
238 self._pipelines_descriptions.append(description)
239 self._valid_scenarios = valid_scenarios
242 def get_config_file(config, private_dir, test_name, extra_data):
243 if isinstance(config, str):
246 os.makedirs(private_dir, exist_ok=True)
247 config_file = os.path.join(private_dir, test_name + '.config')
248 with open(config_file, 'w') as f:
249 f.write(format_config_template(extra_data,
250 '\n'.join(config) + '\n', test_name))
255 def from_dict(cls, test_manager, name, descriptions, extra_data=None):
257 :param json_file: Path to a JSON file containing pipeline tests.
258 :param extra_data: Variables available for interpolation in validate
259 configs and scenario actions.
261 if extra_data is None:
264 pipelines_descriptions = []
265 for test_name, defs in descriptions.items():
266 tests_definition = {'name': test_name, 'pipeline': defs.pop('pipeline')}
267 test_private_dir = os.path.join(test_manager.options.privatedir,
271 config = defs.pop('config', None)
272 timeout = defs.pop('timeout', DEFAULT_TIMEOUT)
273 scenario_defs = defs.pop('scenarios', [])
274 if not scenario_defs and config:
275 config_files[None] = cls.get_config_file(config, test_private_dir, test_name, extra_data)
278 for scenario in scenario_defs:
279 if isinstance(scenario, str):
280 # Path to a scenario file
281 scenarios.append(scenario)
282 scenario_name = os.path.basename(scenario).replace('.scenario', '')
283 test_private_dir = os.path.join(test_manager.options.privatedir,
284 name, test_name, scenario_name)
286 # Dictionary defining a new scenario in-line
287 scenario_name = scenario_file = scenario['name']
288 test_private_dir = os.path.join(test_manager.options.privatedir,
289 name, test_name, scenario_name)
290 actions = scenario.get('actions')
292 os.makedirs(test_private_dir, exist_ok=True)
293 scenario_file = os.path.join(
294 test_private_dir, scenario_name + '.scenario')
295 with open(scenario_file, 'w') as f:
296 f.write('\n'.join(action % extra_data for action in actions) + '\n')
297 scenarios.append(scenario_file)
300 config_files[scenario_name] = cls.get_config_file(config, test_private_dir, test_name + '.' + scenario_name, extra_data)
302 local_extra_data = extra_data.copy()
303 local_extra_data.update(defs)
304 envvars = defs.pop('extra_env_vars', {})
305 local_extra_data.update({
306 'scenarios': scenarios,
307 'config_files': config_files,
308 'plays-reverse': True,
309 'extra_env_vars': envvars,
313 expand_vars_in_dict_recurse(local_extra_data, extra_data)
314 tests_definition['extra_data'] = local_extra_data
315 tests_definition['pipeline_data'] = {}
316 tests_definition['pipeline_data'].update(local_extra_data)
317 pipelines_descriptions.append(tests_definition)
319 return GstValidatePipelineTestsGenerator(name, test_manager, pipelines_descriptions=pipelines_descriptions)
321 def get_fname(self, scenario, protocol=None, name=None):
325 if protocol is not None:
326 protocol_str = "%s." % protocol
330 if scenario is not None and scenario.name.lower() != "none":
331 return "%s%s.%s" % (protocol_str, name, scenario.name)
333 return ("%s.%s.%s" % (protocol_str, self.name, name)).replace("..", ".")
335 def generate_tests(self, uri_minfo_special_scenarios, scenarios):
336 if self._valid_scenarios is None:
338 elif self._valid_scenarios:
339 scenarios = [scenario for scenario in scenarios if
340 scenario is not None and scenario.name in self._valid_scenarios]
342 return super(GstValidatePipelineTestsGenerator, self).generate_tests(
343 uri_minfo_special_scenarios, scenarios)
345 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
347 special_scenarios = []
348 for description in self._pipelines_descriptions:
349 for s in description.get('extra_data', {}).get('scenarios', []):
351 special_scenarios.append(s)
353 self.test_manager.scenarios_manager.discover_scenarios(special_scenarios)
354 for description in self._pipelines_descriptions:
355 pipeline = description['pipeline']
356 extra_data = description.get('extra_data', {})
357 pipeline_data = description.get('pipeline_data', {})
359 if 'scenarios' in extra_data:
360 # A pipeline description can override the default scenario set.
361 # The pipeline description may specify an empty list of
362 # scenarios, in which case one test will be generated with no
364 scenarios_to_iterate = extra_data['scenarios'] or [None]
366 scenarios_to_iterate = scenarios
368 config_files = extra_data.get('config_files')
369 timeout = extra_data.get('timeout', DEFAULT_TIMEOUT)
370 mediainfo = extra_data.get(
371 'media_info', FakeMediaDescriptor(extra_data, pipeline))
372 for scenario in scenarios_to_iterate:
373 if isinstance(scenario, str):
374 tmpscenario = self.test_manager.scenarios_manager.get_scenario(
376 if tmpscenario is None:
377 raise RuntimeError("Could not find scenario file: %s" % scenario)
378 scenario = tmpscenario
380 if not mediainfo.is_compatible(scenario):
383 if self.test_manager.options.mute:
384 needs_clock = scenario.needs_clock_sync() \
385 if scenario else False
386 audiosink = get_fakesink_for_media_type("audio", needs_clock)
387 videosink = get_fakesink_for_media_type("video", needs_clock)
389 audiosink = 'autoaudiosink'
390 videosink = 'autovideosink'
392 pipeline_data.update({'videosink': videosink, 'audiosink': audiosink})
393 pipeline_desc = pipeline % pipeline_data
395 fname = self.get_fname(
396 scenario, protocol=mediainfo.get_protocol(), name=description["name"])
398 expected_issues = extra_data.get("expected-issues")
399 extra_env_vars = extra_data.get("extra_env_vars")
400 test = GstValidateLaunchTest(fname,
401 self.test_manager.options,
402 self.test_manager.reporter,
406 media_descriptor=mediainfo,
407 expected_issues=expected_issues,
408 extra_env_variables=extra_env_vars)
410 test.add_validate_config(config_files[scenario.name if scenario is not None else None])
414 class GstValidatePlaybinTestsGenerator(GstValidatePipelineTestsGenerator):
416 def __init__(self, test_manager):
417 if os.getenv("USE_PLAYBIN3") is None:
418 GstValidatePipelineTestsGenerator.__init__(
419 self, "playback", test_manager, "playbin")
421 GstValidatePipelineTestsGenerator.__init__(
422 self, "playback", test_manager, "playbin3")
424 def _set_sinks(self, minfo, pipe_str, scenario):
425 if self.test_manager.options.mute:
426 needs_clock = scenario.needs_clock_sync() or minfo.media_descriptor.need_clock_sync()
428 afakesink = get_fakesink_for_media_type("audio", needs_clock)
429 vfakesink = get_fakesink_for_media_type("video", needs_clock)
430 pipe_str += " audio-sink='%s' video-sink='%s'" % (
431 afakesink, vfakesink)
435 def _get_name(self, scenario, protocol, minfo):
436 return "%s.%s" % (self.get_fname(scenario,
438 os.path.basename(minfo.media_descriptor.get_clean_name()))
440 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
441 test_rtsp = GstValidateBaseTestManager.RTSP_SERVER_COMMAND
443 printc("\n\nRTSP server not available, you should make sure"
444 " that %s is available in your $PATH." % GstValidateBaseTestManager.RTSP_SERVER_COMMAND,
446 elif self.test_manager.options.disable_rtsp:
447 printc("\n\nRTSP tests are disabled")
450 for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
451 pipe = self._pipeline_template
452 protocol = minfo.media_descriptor.get_protocol()
454 if protocol == Protocols.RTSP:
455 self.debug("SKIPPING %s as it is a RTSP stream" % uri)
458 pipe += " uri=%s" % uri
460 for scenario in special_scenarios + scenarios:
462 if not minfo.media_descriptor.is_compatible(scenario):
465 cpipe = self._set_sinks(minfo, cpipe, scenario)
466 fname = self._get_name(scenario, protocol, minfo)
468 self.debug("Adding: %s", fname)
470 if scenario.does_reverse_playback() and protocol == Protocols.HTTP:
471 # 10MB so we can reverse playback
472 cpipe += " ring-buffer-max-size=10485760"
474 self.add_test(GstValidateLaunchTest(fname,
475 self.test_manager.options,
476 self.test_manager.reporter,
479 media_descriptor=minfo.media_descriptor)
482 if test_rtsp and protocol == Protocols.FILE and not minfo.media_descriptor.is_image():
483 rtspminfo = NamedDic({"path": minfo.media_descriptor.get_path(),
484 "media_descriptor": GstValidateRTSPMediaDescriptor(minfo.media_descriptor.get_path())})
485 if not rtspminfo.media_descriptor.is_compatible(scenario):
488 cpipe = self._set_sinks(rtspminfo, "%s uri=rtsp://127.0.0.1:<RTSPPORTNUMBER>/test"
489 % self._pipeline_template, scenario)
490 fname = self._get_name(scenario, Protocols.RTSP, rtspminfo)
492 self.add_test(GstValidateRTSPTest(
493 fname, self.test_manager.options, self.test_manager.reporter,
494 cpipe, uri, scenario=scenario,
495 media_descriptor=rtspminfo.media_descriptor))
497 fname = self._get_name(scenario, Protocols.RTSP + '2', rtspminfo)
498 self.add_test(GstValidateRTSPTest(
499 fname, self.test_manager.options, self.test_manager.reporter,
500 cpipe, uri, scenario=scenario,
501 media_descriptor=rtspminfo.media_descriptor,
505 class GstValidateCheckAccurateSeekingTestGenerator(GstValidatePipelineTestsGenerator):
506 def __new__(cls, name, test_manager, media_infos, extra_data=None):
509 for path, reference_frame_dir in media_infos:
510 media_info = GstValidateMediaDescriptor(path)
511 media_info.set_protocol("file")
513 error("GstValidateCheckAccurateSeekingTestGenerator",
514 "Could not create a media info file from %s" % path)
517 if media_info.is_image():
518 error("GstValidateCheckAccurateSeekingTestGenerator",
519 "%s is an image, can't run accurate seeking tests" % path)
522 if media_info.get_num_tracks("video") < 1:
523 error("GstValidateCheckAccurateSeekingTestGenerator",
524 "%s is not a video, can't run accurate seeking tests" % path)
527 if media_info.get_num_tracks("video") < 1:
528 error("GstValidateCheckAccurateSeekingTestGenerator",
529 "No video track, can't run accurate seeking tests" % path)
532 if test_manager.options.validate_generate_ssim_reference_files:
534 test_name = media_info.get_clean_name() + '.generate_reference_files'
536 'validatessim, element-name="videoconvert", output-dir="%s"' % reference_frame_dir]
538 test_name = media_info.get_clean_name()
539 framerate, scenario = cls.generate_scenario(test_manager.options, reference_frame_dir, media_info)
541 error("GstValidateCheckAccurateSeekingTestGenerator",
542 "Could not generate test for media info: %s" % path)
546 '%(ssim)s, element-name="videoconvert", reference-images-dir="'
547 + reference_frame_dir + '", framerate=%d/%d' % (framerate.numerator, framerate.denominator)
550 pipelines[test_name] = {
551 "pipeline": "uridecodebin uri=" + media_info.get_uri() + " ! deinterlace ! videoconvert ! video/x-raw,interlace-mode=progressive,format=I420 ! videoconvert name=videoconvert ! %(videosink)s",
552 "media_info": media_info,
557 pipelines[test_name]["scenarios"] = [scenario]
559 return GstValidatePipelineTestsGenerator.from_dict(test_manager, name, pipelines, extra_data=extra_data)
562 def generate_scenario(cls, options, reference_frame_dir, media_info):
564 "description, seek=true, handles-states=true, needs_preroll=true",
569 for track_type, caps in media_info.get_tracks_caps():
570 if track_type == 'video':
571 for struct, _ in GstCaps.new_from_str(caps):
572 framerate = struct["framerate"]
577 n_frames = int((media_info.get_duration() * framerate.numerator) / (GST_SECOND * framerate.denominator))
578 frames_timestamps = [math.ceil(i * framerate.denominator * GST_SECOND / framerate.numerator) for i in range(n_frames)]
579 # Ensure tests are not longer than long_limit, empirically considering we take 0.2 secs per frames.
580 acceptable_n_frames = options.long_limit * 5
581 if n_frames > acceptable_n_frames:
582 n_frames_per_groups = int(acceptable_n_frames / 3)
583 frames_timestamps = frames_timestamps[0:n_frames_per_groups] \
584 + frames_timestamps[int(n_frames / 2 - int(n_frames_per_groups / 2)):int(n_frames / 2 + int(n_frames_per_groups / 2))] \
585 + frames_timestamps[-n_frames_per_groups:n_frames]
587 actions += ['seek, flags=flush+accurate, start=(guint64)%s' % ts for ts in frames_timestamps]
591 "name": "check_accurate_seek",
596 class GstValidateMixerTestsGenerator(GstValidatePipelineTestsGenerator):
598 def __init__(self, name, test_manager, mixer, media_type, converter="",
599 num_sources=3, mixed_srcs=None, valid_scenarios=None):
600 mixed_srcs = mixed_srcs or {}
601 valid_scenarios = valid_scenarios or []
603 pipe_template = "%(mixer)s name=_mixer ! " + \
604 converter + " ! %(sink)s "
605 self.converter = converter
607 self.media_type = media_type
608 self.num_sources = num_sources
609 self.mixed_srcs = mixed_srcs
611 GstValidateMixerTestsGenerator, self).__init__(name, test_manager, pipe_template,
612 valid_scenarios=valid_scenarios)
614 def populate_tests(self, uri_minfo_special_scenarios, scenarios):
615 if self.test_manager.options.validate_uris:
618 wanted_ressources = []
619 for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
620 protocol = minfo.media_descriptor.get_protocol()
621 if protocol == Protocols.FILE and \
622 minfo.media_descriptor.get_num_tracks(self.media_type) > 0:
623 wanted_ressources.append((uri, minfo))
625 if not self.mixed_srcs:
626 if not wanted_ressources:
629 for i in range(len(uri_minfo_special_scenarios) / self.num_sources):
632 for nsource in range(self.num_sources):
633 uri, minfo = wanted_ressources[i + nsource]
634 if os.getenv("USE_PLAYBIN3") is None:
636 "uridecodebin uri=%s ! %s" % (uri, self.converter))
639 "uridecodebin3 uri=%s ! %s" % (uri, self.converter))
640 fname = os.path.basename(uri).replace(".", "_")
644 name += "+%s" % fname
646 self.mixed_srcs[name] = tuple(srcs)
648 for name, srcs in self.mixed_srcs.items():
649 if isinstance(srcs, dict):
651 "mixer": self.mixer + " %s" % srcs["mixer_props"]}
652 srcs = srcs["sources"]
654 pipe_arguments = {"mixer": self.mixer}
656 for scenario in scenarios:
657 fname = self.get_fname(scenario, Protocols.FILE) + "."
660 self.debug("Adding: %s", fname)
662 if self.test_manager.options.mute:
663 pipe_arguments["sink"] = get_fakesink_for_media_type(self.media_type,
664 scenario.needs_clock_sync())
666 pipe_arguments["sink"] = "auto%ssink" % self.media_type
668 pipe = self._pipeline_template % pipe_arguments
671 pipe += "%s ! _mixer. " % src
673 self.add_test(GstValidateLaunchTest(fname,
674 self.test_manager.options,
675 self.test_manager.reporter,
681 class GstValidateSimpleTest(GstValidateTest):
682 def __init__(self, test_file, *args, **kwargs):
683 self.test_file = test_file
684 super().__init__(GstValidateBaseTestManager.COMMAND, *args, **kwargs)
686 def build_arguments(self):
687 self.add_arguments('--set-test-file', self.test_file)
688 if self.options.mute:
689 self.add_arguments('--use-fakesinks')
692 class GstValidateLaunchTest(GstValidateTest):
694 def __init__(self, classname, options, reporter, pipeline_desc,
695 timeout=DEFAULT_TIMEOUT, scenario=None,
696 media_descriptor=None, duration=0, hard_timeout=None,
697 extra_env_variables=None, expected_issues=None):
699 self.extra_env_variables = extra_env_variables or {}
702 duration = scenario.get_duration()
703 elif media_descriptor:
704 duration = media_descriptor.get_duration() / GST_SECOND
707 GstValidateLaunchTest, self).__init__(GstValidateBaseTestManager.COMMAND,
713 hard_timeout=hard_timeout,
714 media_descriptor=media_descriptor,
715 extra_env_variables=extra_env_variables,
716 expected_issues=expected_issues)
718 self.pipeline_desc = pipeline_desc
719 self.media_descriptor = media_descriptor
721 def build_arguments(self):
722 GstValidateTest.build_arguments(self)
723 self.add_arguments(*shlex.split(self.pipeline_desc))
724 if self.media_descriptor is not None and self.media_descriptor.get_path():
726 "--set-media-info", self.media_descriptor.get_path())
729 class GstValidateMediaCheckTest(GstValidateTest):
731 def __init__(self, classname, options, reporter, media_descriptor,
732 uri, minfo_path, timeout=DEFAULT_TIMEOUT,
733 extra_env_variables=None,
734 expected_issues=None):
735 self.extra_env_variables = extra_env_variables or {}
738 GstValidateMediaCheckTest, self).__init__(GstValidateBaseTestManager.MEDIA_CHECK_COMMAND, classname,
741 media_descriptor=media_descriptor,
742 extra_env_variables=extra_env_variables,
743 expected_issues=expected_issues)
745 self._media_info_path = minfo_path
747 def build_arguments(self):
748 Test.build_arguments(self)
749 self.add_arguments(self._uri, "--expected-results",
750 self._media_info_path)
752 if self.media_descriptor.skip_parsers():
753 self.add_arguments("--skip-parsers")
756 class GstValidateTranscodingTest(GstValidateTest, GstValidateEncodingTestInterface):
757 scenarios_manager = ScenarioManager()
759 def __init__(self, classname, options, reporter,
760 combination, uri, media_descriptor,
761 timeout=DEFAULT_TIMEOUT,
763 extra_env_variables=None,
764 expected_issues=None):
765 Loggable.__init__(self)
767 self.extra_env_variables = extra_env_variables or {}
769 file_dur = int(media_descriptor.get_duration()) / GST_SECOND
770 if not media_descriptor.get_num_tracks("video"):
771 self.debug("%s audio only file applying transcoding ratio."
772 "File 'duration' : %s" % (classname, file_dur))
773 duration = file_dur / AUDIO_ONLY_FILE_TRANSCODING_RATIO
778 GstValidateTranscodingTest, self).__init__(GstValidateBaseTestManager.TRANSCODING_COMMAND,
785 media_descriptor=media_descriptor,
786 extra_env_variables=None,
787 expected_issues=expected_issues)
788 extra_env_variables = extra_env_variables or {}
790 GstValidateEncodingTestInterface.__init__(
791 self, combination, media_descriptor)
795 def run_external_checks(self):
796 if self.media_descriptor.get_num_tracks("video") == 1 and \
797 self.options.validate_enable_iqa_tests:
798 self.run_iqa_test(self.uri)
800 def set_rendering_info(self):
801 self.dest_file = os.path.join(self.options.dest,
802 self.classname.replace(".transcode.", os.sep).
803 replace(".", os.sep))
804 mkdir(os.path.dirname(urllib.parse.urlsplit(self.dest_file).path))
805 if urllib.parse.urlparse(self.dest_file).scheme == "":
806 self.dest_file = path2url(self.dest_file)
808 profile = self.get_profile()
809 self.add_arguments("-o", profile)
811 def build_arguments(self):
812 GstValidateTest.build_arguments(self)
813 self.set_rendering_info()
814 self.add_arguments(self.uri, self.dest_file)
816 def get_current_value(self):
818 sent_eos = self.sent_eos_position()
819 if sent_eos is not None:
821 if ((t - sent_eos)) > 30:
822 if self.media_descriptor.get_protocol() == Protocols.HLS:
823 self.set_result(Result.PASSED,
824 """Got no EOS 30 seconds after sending EOS,
825 in HLS known and tolerated issue:
826 https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/issues/132""")
827 return Result.KNOWN_ERROR
830 Result.FAILED, "Pipeline did not stop 30 Seconds after sending EOS")
834 size = self.get_current_size()
836 return self.get_current_position()
840 def check_results(self):
841 self.check_encoded_file()
842 GstValidateTest.check_results(self)
845 class GstValidateBaseRTSPTest:
846 """ Interface for RTSP tests, requires implementing Test"""
849 def __init__(self, local_uri):
850 self._local_uri = local_uri
851 self.rtsp_server = None
852 self._unsetport_pipeline_desc = None
856 def __get_open_port(cls):
859 # http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python?answertab=votes#tab-top
860 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
862 port = s.getsockname()[1]
863 if port not in cls.__used_ports:
864 cls.__used_ports.add(port)
870 def launch_server(self):
871 if self.options.redirect_logs == 'stdout':
872 self.rtspserver_logs = sys.stdout
873 elif self.options.redirect_logs == 'stderr':
874 self.rtspserver_logs = sys.stderr
876 self.server_port = self.__get_open_port()
877 command = [GstValidateBaseTestManager.RTSP_SERVER_COMMAND, self._local_uri, '--port', str(self.server_port)]
879 if self.options.validate_gdb_server:
880 command = self.use_gdb(command)
881 self.rtspserver_logs = sys.stdout
882 elif self.options.redirect_logs:
883 self.rtspserver_logs = sys.stdout
885 self.rtspserver_logs = open(self.logfile + '_rtspserver.log', 'w+')
886 self.extra_logfiles.add(self.rtspserver_logs.name)
888 server_env = os.environ.copy()
890 self.rtsp_server = subprocess.Popen(command,
891 stderr=self.rtspserver_logs,
892 stdout=self.rtspserver_logs,
897 s.connect((("127.0.0.1", self.server_port)))
899 except ConnectionRefusedError:
905 if not self._unsetport_pipeline_desc:
906 self._unsetport_pipeline_desc = self.pipeline_desc
908 self.pipeline_desc = self._unsetport_pipeline_desc.replace(
909 "<RTSPPORTNUMBER>", str(self.server_port))
911 return ' '.join(command)
913 def close_logfile(self):
914 super().close_logfile()
915 if not self.options.redirect_logs:
916 self.rtspserver_logs.close()
918 def process_update(self):
919 res = super().process_update()
921 kill_subprocess(self, self.rtsp_server, DEFAULT_TIMEOUT)
922 self.__used_ports.remove(self.server_port)
927 class GstValidateRTSPTest(GstValidateBaseRTSPTest, GstValidateLaunchTest):
929 def __init__(self, classname, options, reporter, pipeline_desc,
930 local_uri, timeout=DEFAULT_TIMEOUT, scenario=None,
931 media_descriptor=None, rtsp2=False):
932 GstValidateLaunchTest.__init__(self, classname, options, reporter,
933 pipeline_desc, timeout, scenario,
935 GstValidateBaseRTSPTest.__init__(self, local_uri)
938 def get_subproc_env(self):
939 env = super().get_subproc_env()
940 path = env.get('GST_VALIDATE_SCENARIOS_PATH', '')
941 override_dir = get_data_file(os.path.join('data', 'scenarios'), 'rtsp_overrides')
942 env['GST_VALIDATE_SCENARIOS_PATH'] = '%s:%s' % (override_dir, path)
944 env['GST_VALIDATE_SCENARIO'] = env.get('GST_VALIDATE_SCENARIO', '') + ':' + 'force_rtsp2'
949 class GstValidateRTSPMediaDescriptor(GstValidateMediaDescriptor):
951 def __init__(self, xml_path):
952 GstValidateMediaDescriptor.__init__(self, xml_path)
955 return "rtsp://127.0.0.1:8554/test"
957 def get_protocol(self):
958 return Protocols.RTSP
964 class GstValidateTestManager(GstValidateBaseTestManager):
968 # List of all classes to create testsuites
969 GstValidateMediaCheckTestsGenerator = GstValidateMediaCheckTestsGenerator
970 GstValidateTranscodingTestsGenerator = GstValidateTranscodingTestsGenerator
971 GstValidatePipelineTestsGenerator = GstValidatePipelineTestsGenerator
972 GstValidatePlaybinTestsGenerator = GstValidatePlaybinTestsGenerator
973 GstValidateMixerTestsGenerator = GstValidateMixerTestsGenerator
974 GstValidateCheckAccurateSeekingTestGenerator = GstValidateCheckAccurateSeekingTestGenerator
975 GstValidateLaunchTest = GstValidateLaunchTest
976 GstValidateMediaCheckTest = GstValidateMediaCheckTest
977 GstValidateTranscodingTest = GstValidateTranscodingTest
980 super(GstValidateTestManager, self).__init__()
982 self._run_defaults = True
983 self._is_populated = False
984 self._default_generators_registered = False
987 for command, name in [
988 (GstValidateBaseTestManager.TRANSCODING_COMMAND, "gst-validate-transcoding-1.0"),
989 (GstValidateBaseTestManager.COMMAND, "gst-validate-1.0"),
990 (GstValidateBaseTestManager.MEDIA_CHECK_COMMAND, "gst-validate-media-check-1.0")]:
992 self.error("command not found: %s" % name)
997 def add_options(self, parser):
998 group = parser.add_argument_group("GstValidate tools specific options"
1000 description="""When using --wanted-tests, all the scenarios can be used, even those which have
1001 not been tested and explicitly activated if you set use --wanted-tests ALL""")
1002 group.add_argument("--validate-check-uri", dest="validate_uris",
1003 action="append", help="defines the uris to run default tests on")
1004 group.add_argument("--validate-tools-path", dest="validate_tools_path",
1005 action="append", help="defines the paths to look for GstValidate tools.")
1006 group.add_argument("--validate-gdb-server", dest="validate_gdb_server",
1007 help="Run the server in GDB.")
1008 group.add_argument("--validate-disable-rtsp", dest="disable_rtsp",
1009 help="Disable RTSP tests.", default=False, action='store_true')
1010 group.add_argument("--validate-enable-iqa-tests", dest="validate_enable_iqa_tests",
1011 help="Enable Image Quality Assessment validation tests.",
1012 default=False, action='store_true')
1013 group.add_argument("--validate-generate-expectations", dest="validate_generate_expectations",
1014 choices=['auto', 'enabled', 'disabled'],
1015 help="Force generating expectations (when set to `enabed`)"
1016 " force failure on missing expactations when set to `disabled`"
1017 " and create if needed when set to `auto`.",
1019 group.add_argument("--validate-generate-ssim-reference-files",
1020 help="(re)generate ssim reference image files.",
1021 default=False, action='store_true')
1023 def print_valgrind_bugs(self):
1024 # Look for all the 'pending' bugs in our supp file
1026 p = get_data_file('data', 'gstvalidate.supp')
1028 for line in f.readlines():
1030 if line.startswith('# PENDING:'):
1031 tmp = line.split(' ')
1035 msg = "Ignored valgrind bugs:\n"
1037 msg += " + %s\n" % b
1038 printc(msg, Colors.FAIL, True)
1040 def populate_testsuite(self):
1042 if self._is_populated is True:
1045 if not self.options.config and not self.options.testsuites:
1046 if self._run_defaults:
1047 self.register_defaults()
1051 self._is_populated = True
1053 def list_tests(self):
1057 if self._run_defaults:
1058 scenarios = [self.scenarios_manager.get_scenario(scenario_name)
1059 for scenario_name in self.get_scenarios()]
1061 scenarios = self.scenarios_manager.get_scenario(None)
1062 uris = self._list_uris()
1064 for generator in self.get_generators():
1065 for test in generator.generate_tests(uris, scenarios):
1068 if not self.tests and not uris and not self.options.wanted_tests:
1069 self.info("No valid uris present in the path. Check if media files and info files exist")
1073 def _add_media(self, media_info, uri=None):
1074 self.debug("Checking %s", media_info)
1075 if isinstance(media_info, GstValidateMediaDescriptor):
1076 media_descriptor = media_info
1077 media_info = media_descriptor.get_path()
1079 media_descriptor = GstValidateMediaDescriptor(media_info)
1082 # Just testing that the various mandatory infos are present
1083 caps = media_descriptor.get_caps()
1084 if uri is None or media_descriptor.get_protocol() == Protocols.IMAGESEQUENCE:
1085 uri = media_descriptor.get_uri()
1087 # Adjust local http uri
1088 if self.options.http_server_port != 8079 and \
1089 uri.startswith("http://127.0.0.1:8079/"):
1090 uri = uri.replace("http://127.0.0.1:8079/",
1091 "http://127.0.0.1:%r/" % self.options.http_server_port, 1)
1092 media_descriptor.set_protocol(urllib.parse.urlparse(uri).scheme)
1093 for caps2, prot in GST_VALIDATE_CAPS_TO_PROTOCOL:
1095 media_descriptor.set_protocol(prot)
1098 scenario_bname = media_descriptor.get_media_filepath()
1099 special_scenarios = self.scenarios_manager.find_special_scenarios(
1101 self._uris.append((uri,
1102 NamedDic({"path": media_info,
1103 "media_descriptor": media_descriptor}),
1105 except configparser.NoOptionError as e:
1106 self.debug("Exception: %s for %s", e, media_info)
1108 def _discover_file(self, uri, fpath):
1109 for ext in (GstValidateMediaDescriptor.MEDIA_INFO_EXT,
1110 GstValidateMediaDescriptor.PUSH_MEDIA_INFO_EXT,
1111 GstValidateMediaDescriptor.SKIPPED_MEDIA_INFO_EXT):
1113 is_push = ext == GstValidateMediaDescriptor.PUSH_MEDIA_INFO_EXT
1114 is_skipped = ext == GstValidateMediaDescriptor.SKIPPED_MEDIA_INFO_EXT
1115 media_info = "%s.%s" % (fpath, ext)
1116 if is_push or is_skipped:
1117 if not os.path.exists(media_info):
1121 args = GstValidateBaseTestManager.MEDIA_CHECK_COMMAND.split(" ")
1123 if os.path.isfile(media_info) and not self.options.update_media_info and not is_skipped:
1124 self._add_media(media_info, uri)
1126 elif fpath.endswith(GstValidateMediaDescriptor.STREAM_INFO_EXT) and not is_skipped:
1127 self._add_media(fpath)
1129 elif not self.options.generate_info and not self.options.update_media_info and not self.options.validate_uris:
1131 elif self.options.update_media_info and not os.path.isfile(media_info):
1133 "%s not present. Use --generate-media-info", media_info)
1135 elif os.path.islink(media_info):
1137 "%s is a symlink, not updating and hopefully the actual file gets updated!", media_info)
1141 if self.options.update_media_info:
1143 elif self.options.generate_info_full:
1146 media_descriptor = GstValidateMediaDescriptor.new_from_uri(
1147 uri, True, include_frames, is_push, is_skipped)
1148 if media_descriptor:
1149 self._add_media(media_descriptor, uri)
1151 self.warning("Could not get any descriptor for %s" % uri)
1153 except subprocess.CalledProcessError as e:
1154 if self.options.generate_info:
1155 printc("Result: Failed", Colors.FAIL)
1157 self.error("Exception: %s", e)
1161 def _list_uris(self):
1165 if self.options.validate_uris:
1166 for uri in self.options.validate_uris:
1167 self._discover_file(uri, uri)
1171 if isinstance(self.options.paths, str):
1172 self.options.paths = [os.path.join(self.options.paths)]
1174 for path in self.options.paths:
1175 if os.path.isfile(path):
1176 path = os.path.abspath(path)
1177 self._discover_file(path2url(path), path)
1179 for root, dirs, files in os.walk(path):
1181 fpath = os.path.abspath(os.path.join(root, f))
1182 if os.path.isdir(fpath) or \
1183 fpath.endswith(GstValidateMediaDescriptor.MEDIA_INFO_EXT) or\
1184 fpath.endswith(ScenarioManager.FILE_EXTENSION):
1187 self._discover_file(path2url(fpath), fpath)
1189 self.debug("Uris found: %s", self._uris)
1193 def needs_http_server(self):
1194 for test in self.list_tests():
1195 if self._is_test_wanted(test) and test.media_descriptor is not None:
1196 protocol = test.media_descriptor.get_protocol()
1197 uri = test.media_descriptor.get_uri()
1198 uri_requires_http_server = False
1200 if 'http-server-port' in uri:
1201 expanded_uri = uri % {
1202 'http-server-port': self.options.http_server_port}
1203 uri_requires_http_server = expanded_uri.find(
1204 "127.0.0.1:%s" % self.options.http_server_port) != -1
1205 if protocol in [Protocols.HTTP, Protocols.HLS, Protocols.DASH] or uri_requires_http_server:
1209 def set_settings(self, options, args, reporter):
1210 if options.wanted_tests:
1211 for i in range(len(options.wanted_tests)):
1212 if "ALL" in options.wanted_tests[i]:
1213 self._run_defaults = False
1214 options.wanted_tests[
1215 i] = options.wanted_tests[i].replace("ALL", "")
1217 options.validate_default_config = None
1218 if options.validate_generate_expectations != 'auto':
1219 options.validate_default_config = os.path.join(options.logsdir, "__validate_default.config")
1220 with open(options.validate_default_config, 'w') as f:
1221 val = "true" if options.validate_generate_expectations == "enabled" else "false"
1222 print("validateflow,generate-expectations=%s" % val, file=f)
1224 options.wanted_tests.remove("")
1228 if options.validate_uris or options.validate_generate_ssim_reference_files:
1229 self.check_testslist = False
1231 super(GstValidateTestManager, self).set_settings(
1232 options, args, reporter)
1234 def register_defaults(self):
1236 Registers the defaults:
1237 * Scenarios to be used
1238 * Encoding formats to be used
1242 printc("-> Registering default 'validate' tests... ", end='')
1243 self.register_default_scenarios()
1244 self.register_default_encoding_formats()
1245 self.register_default_blacklist()
1246 self.register_default_test_generators()
1247 printc("OK", Colors.OKGREEN)
1249 def register_default_scenarios(self):
1251 Registers default test scenarios
1253 if self.options.long_limit != 0:
1254 self.add_scenarios([
1261 "switch_audio_track",
1262 "switch_audio_track_while_paused",
1263 "switch_subtitle_track",
1264 "switch_subtitle_track_while_paused",
1265 "disable_subtitle_track_while_paused",
1266 "change_state_intensive",
1267 "scrub_forward_seeking"])
1269 self.add_scenarios([
1276 "switch_audio_track",
1277 "switch_audio_track_while_paused",
1278 "switch_subtitle_track",
1279 "switch_subtitle_track_while_paused",
1280 "disable_subtitle_track_while_paused",
1281 "change_state_intensive",
1282 "scrub_forward_seeking"])
1284 def register_default_encoding_formats(self):
1286 Registers default encoding formats
1288 self.add_encoding_formats([
1289 MediaFormatCombination("ogg", "vorbis", "theora"),
1290 MediaFormatCombination("webm", "vorbis", "vp8"),
1291 MediaFormatCombination("webm", "vorbis", "vp9",
1292 video_restriction="video/x-raw,width=160,height=120"),
1293 MediaFormatCombination("mp4", "mp3", "h264"),
1294 MediaFormatCombination("mkv", "vorbis", "h264"),
1297 def register_default_blacklist(self):
1298 self.set_default_blacklist([
1299 # testbin known issues
1300 ("testbin.media_check.*",
1301 "Not supported by GstDiscoverer."),
1304 ("dash.media_check.*",
1305 "Caps are different depending on selected bitrates, etc"),
1307 # Matroska/WEBM known issues:
1308 ("*.reverse_playback.*webm$",
1309 "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/65"),
1310 ("*.reverse_playback.*mkv$",
1311 "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/65"),
1312 ("http.playback.seek_with_stop.*webm",
1313 "matroskademux.gst_matroska_demux_handle_seek_push: Seek end-time not supported in streaming mode"),
1314 ("http.playback.seek_with_stop.*mkv",
1315 "matroskademux.gst_matroska_demux_handle_seek_push: Seek end-time not supported in streaming mode"),
1317 # MPEG TS known issues:
1318 ('(?i)*playback.reverse_playback.*(?:_|.)(?:|m)ts$',
1319 "https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/issues/97"),
1321 # Fragmented MP4 disabled tests:
1322 ('*.playback..*seek.*.fragmented_nonseekable_sink_mp4',
1323 "Seeking on fragmented files without indexes isn't implemented"),
1324 ('*.playback.reverse_playback.fragmented_nonseekable_sink_mp4',
1325 "Seeking on fragmented files without indexes isn't implemented"),
1327 # HTTP known issues:
1328 ("http.*scrub_forward_seeking.*",
1329 "This is not stable enough for now."),
1330 ("http.playback.change_state_intensive.raw_video_mov",
1331 "This is not stable enough for now. (flow return from pad push doesn't match expected value)"),
1334 ("*reverse_playback.*mxf",
1335 "Reverse playback is not handled in MXF"),
1336 ("file\.transcode.*mxf",
1337 "FIXME: Transcoding and mixing tests need to be tested"),
1340 ("*reverse_playback.*wmv",
1341 "Reverse playback is not handled in wmv"),
1342 (".*reverse_playback.*asf",
1343 "Reverse playback is not handled in asf"),
1346 ("http.playback.seek.*vorbis_theora_1_ogg",
1347 "https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/issues/281"),
1349 ('rtsp.*playback.reverse.*',
1350 'https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/32'),
1351 ('rtsp.*playback.seek_with_stop.*',
1352 'https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/386'),
1353 ('rtsp.*playback.fast_*',
1354 'rtpbasedepayload does not handle rate != 1.0 correctly'),
1357 def register_default_test_generators(self):
1359 Registers default test generators
1361 if self._default_generators_registered:
1364 self.add_generators([GstValidatePlaybinTestsGenerator(self),
1365 GstValidateMediaCheckTestsGenerator(self),
1366 GstValidateTranscodingTestsGenerator(self)])
1367 self._default_generators_registered = True