subprojects/gst-devtools/validate/launcher/apps/gstvalidate.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (c) 2013,Thibault Saunier <thibault.saunier@collabora.com>
   4 #
   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.
   9 #
  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.
  14 #
  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.
  19 import argparse
  20 from fractions import Fraction
  21 import os
  22 import copy
  23 import sys
  24 import time
  25 import urllib.parse
  26 import shlex
  27 import socket
  28 import subprocess
  29 import configparser
  30 import json
  31 import glob
  32 import math
  33 from launcher.loggable import Loggable, error
  34
  35 from launcher.baseclasses import GstValidateTest, Test, \
  36     ScenarioManager, NamedDic, GstValidateTestsGenerator, \
  37     GstValidateMediaDescriptor, GstValidateEncodingTestInterface, \
  38     GstValidateBaseTestManager, MediaDescriptor, MediaFormatCombination, VariableFramerateMode
  39
  40 from launcher.utils import path2url, url2path, DEFAULT_TIMEOUT, which, \
  41     GST_SECOND, Result, Protocols, mkdir, printc, Colors, get_data_file, \
  42     kill_subprocess, format_config_template, get_fakesink_for_media_type, \
  43     parse_gsttimeargs, GstCaps
  44
  45 #
  46 # Private global variables     #
  47 #
  48
  49 # definitions of commands to use
  50 parser = argparse.ArgumentParser(add_help=False)
  51 parser.add_argument("--validate-tools-path", dest="validate_tools_path",
  52                     default="",
  53                     help="defines the paths to look for GstValidate tools.")
  54 options, args = parser.parse_known_args()
  55
  56 GstValidateBaseTestManager.update_commands(options.validate_tools_path)
  57 AUDIO_ONLY_FILE_TRANSCODING_RATIO = 5
  58
  59 #
  60 # API to be used to create testsuites     #
  61 #
  62
  63 """
  64 Some info about protocols and how to handle them
  65 """
  66 GST_VALIDATE_CAPS_TO_PROTOCOL = [("application/x-hls", Protocols.HLS),
  67                                  ("application/dash+xml", Protocols.DASH)]
  68
  69
  70 def expand_vars_in_list_recurse(lines, data):
  71     for i, v in enumerate(lines):
  72         if isinstance(v, dict):
  73             lines[i] = expand_vars_in_dict_recurse(v, data)
  74         elif isinstance(v, str):
  75             lines[i] = v % data
  76         elif isinstance(v, list):
  77             lines[i] = expand_vars_in_list_recurse(v, data)
  78
  79     return lines
  80
  81
  82 def expand_vars_in_dict_recurse(dico, data):
  83     for key, value in dico.items():
  84         if isinstance(value, dict):
  85             dico[key] = expand_vars_in_dict_recurse(value, data)
  86         elif isinstance(value, str):
  87             dico[key] = value % data
  88         elif isinstance(value, list):
  89             dico[key] = expand_vars_in_list_recurse(value, data)
  90
  91     return dico
  92
  93
  94 class GstValidateMediaCheckTestsGenerator(GstValidateTestsGenerator):
  95
  96     def __init__(self, test_manager):
  97         GstValidateTestsGenerator.__init__(self, "media_check", test_manager)
  98
  99     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 100         for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
 101             protocol = mediainfo.media_descriptor.get_protocol()
 102             timeout = DEFAULT_TIMEOUT
 103
 104             classname = "%s.media_check.%s" % (protocol,
 105                                                os.path.basename(url2path(uri)).replace(".", "_"))
 106             self.add_test(GstValidateMediaCheckTest(classname,
 107                                                     self.test_manager.options,
 108                                                     self.test_manager.reporter,
 109                                                     mediainfo.media_descriptor,
 110                                                     uri,
 111                                                     mediainfo.path,
 112                                                     timeout=timeout))
 113
 114
 115 class GstValidateTranscodingTestsGenerator(GstValidateTestsGenerator):
 116     HARD_TIMEOUT_FACTOR = 10
 117
 118     def __init__(self, test_manager):
 119         GstValidateTestsGenerator.__init__(self, "transcode", test_manager)
 120
 121     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 122         for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
 123             if mediainfo.media_descriptor.is_image():
 124                 continue
 125
 126             protocol = mediainfo.media_descriptor.get_protocol()
 127             if protocol == Protocols.RTSP:
 128                 continue
 129
 130             options = self.test_manager.options
 131             for comb in self.test_manager.get_encoding_formats():
 132                 classname = "%s.transcode.to_%s.%s" % (mediainfo.media_descriptor.get_protocol(),
 133                                                        str(comb).replace(
 134                                                        ' ', '_'),
 135                                                        mediainfo.media_descriptor.get_clean_name())
 136
 137                 self.add_test(GstValidateTranscodingTest(classname,
 138                                                          options,
 139                                                          self.test_manager.reporter,
 140                                                          comb,
 141                                                          uri,
 142                                                          mediainfo.media_descriptor))
 143
 144
 145 class FakeMediaDescriptor(MediaDescriptor):
 146
 147     def __init__(self, infos, pipeline_desc):
 148         MediaDescriptor.__init__(self)
 149         self._infos = infos
 150         self._pipeline_desc = pipeline_desc
 151
 152     def get_path(self):
 153         return self._infos.get('path', None)
 154
 155     def get_tracks_caps(self):
 156         return self._info.get('tracks-caps', [])
 157
 158     def get_media_filepath(self):
 159         return self._infos.get('media-filepath', None)
 160
 161     def get_caps(self):
 162         return self._infos.get('caps', None)
 163
 164     def get_uri(self):
 165         return self._infos.get('uri', None)
 166
 167     def get_duration(self):
 168         return int(self._infos.get('duration', 0)) * GST_SECOND
 169
 170     def get_protocol(self):
 171         return self._infos.get('protocol', "launch_pipeline")
 172
 173     def is_seekable(self):
 174         return self._infos.get('is-seekable', True)
 175
 176     def is_image(self):
 177         return self._infos.get('is-image', False)
 178
 179     def is_live(self):
 180         return self._infos.get('is-live', False)
 181
 182     def get_num_tracks(self, track_type):
 183         return self._infos.get('num-%s-tracks' % track_type,
 184                                self._pipeline_desc.count(track_type + "sink"))
 185
 186     def can_play_reverse(self):
 187         return self._infos.get('plays-reverse', False)
 188
 189
 190 class GstValidateSimpleTestsGenerator(GstValidateTestsGenerator):
 191     def __init__(self, name, test_manager, tests_dir):
 192         self.tests_dir = tests_dir
 193         super().__init__(name, test_manager)
 194
 195     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 196         validatetests = []
 197         for root, _, files in os.walk(self.tests_dir):
 198             for f in files:
 199                 name, ext = os.path.splitext(f)
 200                 if ext != ".validatetest":
 201                     continue
 202
 203                 validatetests.append(os.path.abspath(os.path.join(root, f)))
 204
 205         for validatetest in self.test_manager.scenarios_manager.discover_scenarios(validatetests):
 206             pathname, _ext = os.path.splitext(validatetest.path)
 207             name = pathname.replace(os.path.commonpath([self.tests_dir, root]), '').replace('/', '.')
 208
 209             self.add_test(GstValidateSimpleTest(validatetest.path,
 210                                                 'test' + name,
 211                                                 self.test_manager.options,
 212                                                 self.test_manager.reporter,
 213                                                 test_info=validatetest))
 214
 215
 216 class GstValidatePipelineTestsGenerator(GstValidateTestsGenerator):
 217
 218     def __init__(self, name, test_manager, pipeline_template=None,
 219                  pipelines_descriptions=None, valid_scenarios=None):
 220         """
 221         @name: The name of the generator
 222         @pipeline_template: A template pipeline to be used to generate actual pipelines
 223         @pipelines_descriptions: A list of tuple of the form:
 224                                  (test_name, pipeline_description, extra_data)
 225                                  extra_data being a dictionary with the following keys:
 226                                     'scenarios': ["the", "valide", "scenarios", "names"]
 227                                     'duration': the_duration # in seconds
 228                                     'timeout': a_timeout # in seconds
 229                                     'hard_timeout': a_hard_timeout # in seconds
 230
 231         @valid_scenarios: A list of scenario name that can be used with that generator
 232         """
 233         valid_scenarios = valid_scenarios or []
 234         GstValidateTestsGenerator.__init__(self, name, test_manager)
 235         self._pipeline_template = pipeline_template
 236         self._pipelines_descriptions = []
 237         for description in pipelines_descriptions or []:
 238             if not isinstance(description, dict):
 239                 desc_dict = {"name": description[0],
 240                      "pipeline": description[1]}
 241                 if len(description) >= 3:
 242                     desc_dict["extra_data"] = description[2]
 243                 self._pipelines_descriptions.append(desc_dict)
 244             else:
 245                 self._pipelines_descriptions.append(description)
 246         self._valid_scenarios = valid_scenarios
 247
 248     @staticmethod
 249     def get_config_file(config, private_dir, test_name, extra_data):
 250         if isinstance(config, str):
 251             return config
 252
 253         os.makedirs(private_dir, exist_ok=True)
 254         config_file = os.path.join(private_dir, test_name + '.config')
 255         with open(config_file, 'w') as f:
 256             f.write(format_config_template(extra_data,
 257                     '\n'.join(config) + '\n', test_name))
 258
 259         return config_file
 260
 261     @classmethod
 262     def from_dict(cls, test_manager, name, descriptions, extra_data=None):
 263         """
 264         :param json_file: Path to a JSON file containing pipeline tests.
 265         :param extra_data: Variables available for interpolation in validate
 266         configs and scenario actions.
 267         """
 268         if extra_data is None:
 269             extra_data = {}
 270
 271         pipelines_descriptions = []
 272         for test_name, defs in descriptions.items():
 273             tests_definition = {'name': test_name, 'pipeline': defs.pop('pipeline')}
 274             test_private_dir = os.path.join(test_manager.options.privatedir,
 275                                             name, test_name)
 276
 277             config_files = {}
 278             config = defs.pop('config', None)
 279             timeout = defs.pop('timeout', DEFAULT_TIMEOUT)
 280             scenario_defs = defs.pop('scenarios', [])
 281             if not scenario_defs and config:
 282                 config_files[None] = cls.get_config_file(config, test_private_dir, test_name, extra_data)
 283
 284             scenarios = []
 285             for scenario in scenario_defs:
 286                 if isinstance(scenario, str):
 287                     # Path to a scenario file
 288                     scenarios.append(scenario)
 289                     scenario_name = os.path.basename(scenario).replace('.scenario', '')
 290                     test_private_dir = os.path.join(test_manager.options.privatedir,
 291                                                     name, test_name, scenario_name)
 292                 else:
 293                     # Dictionary defining a new scenario in-line
 294                     scenario_name = scenario_file = scenario['name']
 295                     test_private_dir = os.path.join(test_manager.options.privatedir,
 296                                                     name, test_name, scenario_name)
 297                     actions = scenario.get('actions')
 298                     if actions:
 299                         os.makedirs(test_private_dir, exist_ok=True)
 300                         scenario_file = os.path.join(
 301                             test_private_dir, scenario_name + '.scenario')
 302                         with open(scenario_file, 'w') as f:
 303                             f.write('\n'.join(action % extra_data for action in actions) + '\n')
 304                     scenarios.append(scenario_file)
 305
 306                 if config:
 307                     config_files[scenario_name] = cls.get_config_file(config, test_private_dir, test_name + '.' + scenario_name, extra_data)
 308
 309             local_extra_data = extra_data.copy()
 310             local_extra_data.update(defs)
 311             envvars = defs.pop('extra_env_vars', {})
 312             local_extra_data.update({
 313                 'scenarios': scenarios,
 314                 'config_files': config_files,
 315                 'plays-reverse': True,
 316                 'extra_env_vars': envvars,
 317                 'timeout': timeout,
 318             })
 319
 320             expand_vars_in_dict_recurse(local_extra_data, extra_data)
 321             tests_definition['extra_data'] = local_extra_data
 322             tests_definition['pipeline_data'] = {}
 323             tests_definition['pipeline_data'].update(local_extra_data)
 324             pipelines_descriptions.append(tests_definition)
 325
 326         return GstValidatePipelineTestsGenerator(name, test_manager, pipelines_descriptions=pipelines_descriptions)
 327
 328     def get_fname(self, scenario, protocol=None, name=None):
 329         if name is None:
 330             name = self.name
 331
 332         if protocol is not None:
 333             protocol_str = "%s." % protocol
 334         else:
 335             protocol_str = ""
 336
 337         if scenario is not None and scenario.name.lower() != "none":
 338             return "%s%s.%s" % (protocol_str, name, scenario.name)
 339
 340         return ("%s.%s.%s" % (protocol_str, self.name, name)).replace("..", ".")
 341
 342     def generate_tests(self, uri_minfo_special_scenarios, scenarios):
 343         if self._valid_scenarios is None:
 344             scenarios = [None]
 345         elif self._valid_scenarios:
 346             scenarios = [scenario for scenario in scenarios if
 347                          scenario is not None and scenario.name in self._valid_scenarios]
 348
 349         return super(GstValidatePipelineTestsGenerator, self).generate_tests(
 350             uri_minfo_special_scenarios, scenarios)
 351
 352     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 353
 354         special_scenarios = []
 355         for description in self._pipelines_descriptions:
 356             for s in description.get('extra_data', {}).get('scenarios', []):
 357                 if os.path.isabs(s):
 358                     special_scenarios.append(s)
 359
 360         self.test_manager.scenarios_manager.discover_scenarios(special_scenarios)
 361         for description in self._pipelines_descriptions:
 362             pipeline = description['pipeline']
 363             extra_data = description.get('extra_data', {})
 364             pipeline_data = description.get('pipeline_data', {})
 365
 366             if 'scenarios' in extra_data:
 367                 # A pipeline description can override the default scenario set.
 368                 # The pipeline description may specify an empty list of
 369                 # scenarios, in which case one test will be generated with no
 370                 # scenario.
 371                 scenarios_to_iterate = extra_data['scenarios'] or [None]
 372             else:
 373                 scenarios_to_iterate = scenarios
 374
 375             config_files = extra_data.get('config_files')
 376             timeout = extra_data.get('timeout', DEFAULT_TIMEOUT)
 377             mediainfo = extra_data.get(
 378                 'media_info', FakeMediaDescriptor(extra_data, pipeline))
 379             for scenario in scenarios_to_iterate:
 380                 if isinstance(scenario, str):
 381                     tmpscenario = self.test_manager.scenarios_manager.get_scenario(
 382                         scenario)
 383                     if tmpscenario is None:
 384                         raise RuntimeError("Could not find scenario file: %s" % scenario)
 385                     scenario = tmpscenario
 386
 387                 if not mediainfo.is_compatible(scenario):
 388                     continue
 389
 390                 if self.test_manager.options.mute:
 391                     needs_clock = scenario.needs_clock_sync() \
 392                         if scenario else False
 393                     audiosink = get_fakesink_for_media_type("audio", needs_clock)
 394                     videosink = get_fakesink_for_media_type("video", needs_clock)
 395                 else:
 396                     audiosink = 'autoaudiosink'
 397                     videosink = 'autovideosink'
 398
 399                 pipeline_data.update({'videosink': videosink, 'audiosink': audiosink})
 400                 pipeline_desc = pipeline % pipeline_data
 401
 402                 fname = self.get_fname(
 403                     scenario, protocol=mediainfo.get_protocol(), name=description["name"])
 404
 405                 expected_issues = extra_data.get("expected-issues")
 406                 extra_env_vars = extra_data.get("extra_env_vars")
 407                 test = GstValidateLaunchTest(fname,
 408                                              self.test_manager.options,
 409                                              self.test_manager.reporter,
 410                                              pipeline_desc,
 411                                              scenario=scenario,
 412                                              timeout=timeout,
 413                                              media_descriptor=mediainfo,
 414                                              expected_issues=expected_issues,
 415                                              extra_env_variables=extra_env_vars)
 416                 if config_files:
 417                     test.add_validate_config(config_files[scenario.name if scenario is not None else None])
 418                 self.add_test(test)
 419
 420
 421 class GstValidatePlaybinTestsGenerator(GstValidatePipelineTestsGenerator):
 422
 423     def __init__(self, test_manager):
 424         if os.getenv("USE_PLAYBIN3") is None:
 425             GstValidatePipelineTestsGenerator.__init__(
 426                 self, "playback", test_manager, "playbin")
 427         else:
 428             GstValidatePipelineTestsGenerator.__init__(
 429                 self, "playback", test_manager, "playbin3")
 430
 431     def _set_sinks(self, minfo, pipe_str, scenario):
 432         if self.test_manager.options.mute:
 433             needs_clock = scenario.needs_clock_sync() or minfo.media_descriptor.need_clock_sync()
 434
 435             afakesink = get_fakesink_for_media_type("audio", needs_clock)
 436             vfakesink = get_fakesink_for_media_type("video", needs_clock)
 437             pipe_str += " audio-sink='%s' video-sink='%s'" % (
 438                 afakesink, vfakesink)
 439
 440         return pipe_str
 441
 442     def _get_name(self, scenario, protocol, minfo):
 443         return "%s.%s" % (self.get_fname(scenario,
 444                                          protocol),
 445                           os.path.basename(minfo.media_descriptor.get_clean_name()))
 446
 447     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 448         test_rtsp = GstValidateBaseTestManager.RTSP_SERVER_COMMAND
 449         if not test_rtsp:
 450             printc("\n\nRTSP server not available, you should make sure"
 451                    " that %s is available in your $PATH." % GstValidateBaseTestManager.RTSP_SERVER_COMMAND,
 452                    Colors.FAIL)
 453         elif self.test_manager.options.disable_rtsp:
 454             printc("\n\nRTSP tests are disabled")
 455             test_rtsp = False
 456
 457         for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
 458             pipe = self._pipeline_template
 459             protocol = minfo.media_descriptor.get_protocol()
 460
 461             if protocol == Protocols.RTSP:
 462                 self.debug("SKIPPING %s as it is a RTSP stream" % uri)
 463                 continue
 464
 465             pipe += " uri=%s" % uri
 466
 467             for scenario in special_scenarios + scenarios:
 468                 cpipe = pipe
 469                 if not minfo.media_descriptor.is_compatible(scenario):
 470                     continue
 471
 472                 cpipe = self._set_sinks(minfo, cpipe, scenario)
 473                 fname = self._get_name(scenario, protocol, minfo)
 474
 475                 self.debug("Adding: %s", fname)
 476
 477                 if scenario.does_reverse_playback() and protocol == Protocols.HTTP:
 478                     # 10MB so we can reverse playback
 479                     cpipe += " ring-buffer-max-size=10485760"
 480
 481                 self.add_test(GstValidateLaunchTest(fname,
 482                                                     self.test_manager.options,
 483                                                     self.test_manager.reporter,
 484                                                     cpipe,
 485                                                     scenario=scenario,
 486                                                     media_descriptor=minfo.media_descriptor)
 487                               )
 488
 489                 if test_rtsp and protocol == Protocols.FILE and not minfo.media_descriptor.is_image():
 490                     rtspminfo = NamedDic({"path": minfo.media_descriptor.get_path(),
 491                                           "media_descriptor": GstValidateRTSPMediaDescriptor(minfo.media_descriptor.get_path())})
 492                     if not rtspminfo.media_descriptor.is_compatible(scenario):
 493                         continue
 494
 495                     cpipe = self._set_sinks(rtspminfo, "%s uri=rtsp://127.0.0.1:<RTSPPORTNUMBER>/test"
 496                                             % self._pipeline_template, scenario)
 497                     fname = self._get_name(scenario, Protocols.RTSP, rtspminfo)
 498
 499                     self.add_test(GstValidateRTSPTest(
 500                         fname, self.test_manager.options, self.test_manager.reporter,
 501                         cpipe, uri, scenario=scenario,
 502                         media_descriptor=rtspminfo.media_descriptor))
 503
 504                     fname = self._get_name(scenario, Protocols.RTSP + '2', rtspminfo)
 505                     self.add_test(GstValidateRTSPTest(
 506                         fname, self.test_manager.options, self.test_manager.reporter,
 507                         cpipe, uri, scenario=scenario,
 508                         media_descriptor=rtspminfo.media_descriptor,
 509                         rtsp2=True))
 510
 511
 512 class GstValidateCheckAccurateSeekingTestGenerator(GstValidatePipelineTestsGenerator):
 513     def __new__(cls, name, test_manager, media_infos, extra_data=None):
 514         pipelines = {}
 515
 516         for path, reference_frame_dir in media_infos:
 517             media_info = GstValidateMediaDescriptor(path)
 518             media_info.set_protocol("file")
 519             if not media_info:
 520                 error("GstValidateCheckAccurateSeekingTestGenerator",
 521                       "Could not create a media info file from %s" % path)
 522                 continue
 523
 524             if media_info.is_image():
 525                 error("GstValidateCheckAccurateSeekingTestGenerator",
 526                       "%s is an image, can't run accurate seeking tests" % path)
 527                 continue
 528
 529             if media_info.get_num_tracks("video") < 1:
 530                 error("GstValidateCheckAccurateSeekingTestGenerator",
 531                       "%s is not a video, can't run accurate seeking tests" % path)
 532                 continue
 533
 534             if media_info.get_num_tracks("video") < 1:
 535                 error("GstValidateCheckAccurateSeekingTestGenerator",
 536                       "No video track, can't run accurate seeking tests" % path)
 537                 continue
 538
 539             if test_manager.options.validate_generate_ssim_reference_files:
 540                 scenario = None
 541                 test_name = media_info.get_clean_name() + '.generate_reference_files'
 542                 config = [
 543                     'validatessim, element-name="videoconvert", output-dir="%s"' % reference_frame_dir]
 544             else:
 545                 test_name = media_info.get_clean_name()
 546                 framerate, scenario = cls.generate_scenario(test_manager.options, reference_frame_dir, media_info)
 547                 if scenario is None:
 548                     error("GstValidateCheckAccurateSeekingTestGenerator",
 549                         "Could not generate test for media info: %s" % path)
 550                     continue
 551
 552                 config = [
 553                     '%(ssim)s, element-name="videoconvert", reference-images-dir="'
 554                     + reference_frame_dir + '", framerate=%d/%d' % (framerate.numerator, framerate.denominator)
 555                 ]
 556
 557             pipelines[test_name] = {
 558                 "pipeline": "uridecodebin uri=" + media_info.get_uri() + " ! deinterlace ! videoconvert ! video/x-raw,interlace-mode=progressive,format=I420 ! videoconvert name=videoconvert ! %(videosink)s",
 559                 "media_info": media_info,
 560                 "config": config,
 561             }
 562
 563             if scenario:
 564                 pipelines[test_name]["scenarios"] = [scenario]
 565
 566         return GstValidatePipelineTestsGenerator.from_dict(test_manager, name, pipelines, extra_data=extra_data)
 567
 568     @classmethod
 569     def generate_scenario(cls, options, reference_frame_dir, media_info):
 570         actions = [
 571             "description, seek=true, handles-states=true, needs_preroll=true",
 572             "pause",
 573         ]
 574
 575         framerate = None
 576         for track_type, caps in media_info.get_tracks_caps():
 577             if track_type == 'video':
 578                 for struct, _ in GstCaps.new_from_str(caps):
 579                     framerate = struct["framerate"]
 580             if framerate:
 581                 break
 582         assert framerate
 583
 584         n_frames = int((media_info.get_duration() * framerate.numerator) / (GST_SECOND * framerate.denominator))
 585         frames_timestamps = [math.ceil(i * framerate.denominator * GST_SECOND / framerate.numerator) for i in range(n_frames)]
 586         # Ensure tests are not longer than long_limit, empirically considering we take 0.2 secs per frames.
 587         acceptable_n_frames = options.long_limit * 5
 588         if n_frames > acceptable_n_frames:
 589             n_frames_per_groups = int(acceptable_n_frames / 3)
 590             frames_timestamps = frames_timestamps[0:n_frames_per_groups] \
 591                 + frames_timestamps[int(n_frames / 2 - int(n_frames_per_groups / 2)):int(n_frames / 2 + int(n_frames_per_groups / 2))] \
 592                 + frames_timestamps[-n_frames_per_groups:n_frames]
 593
 594         actions += ['seek, flags=flush+accurate, start=(guint64)%s' % ts for ts in frames_timestamps]
 595         actions += ['stop']
 596
 597         return framerate, {
 598             "name": "check_accurate_seek",
 599             "actions": actions,
 600         }
 601
 602
 603 class GstValidateMixerTestsGenerator(GstValidatePipelineTestsGenerator):
 604
 605     def __init__(self, name, test_manager, mixer, media_type, converter="",
 606                  num_sources=3, mixed_srcs=None, valid_scenarios=None):
 607         mixed_srcs = mixed_srcs or {}
 608         valid_scenarios = valid_scenarios or []
 609
 610         pipe_template = "%(mixer)s name=_mixer !  " + \
 611             converter + " ! %(sink)s "
 612         self.converter = converter
 613         self.mixer = mixer
 614         self.media_type = media_type
 615         self.num_sources = num_sources
 616         self.mixed_srcs = mixed_srcs
 617         super(
 618             GstValidateMixerTestsGenerator, self).__init__(name, test_manager, pipe_template,
 619                                                            valid_scenarios=valid_scenarios)
 620
 621     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 622         if self.test_manager.options.validate_uris:
 623             return
 624
 625         wanted_ressources = []
 626         for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
 627             protocol = minfo.media_descriptor.get_protocol()
 628             if protocol == Protocols.FILE and \
 629                     minfo.media_descriptor.get_num_tracks(self.media_type) > 0:
 630                 wanted_ressources.append((uri, minfo))
 631
 632         if not self.mixed_srcs:
 633             if not wanted_ressources:
 634                 return
 635
 636             for i in range(len(uri_minfo_special_scenarios) / self.num_sources):
 637                 srcs = []
 638                 name = ""
 639                 for nsource in range(self.num_sources):
 640                     uri, minfo = wanted_ressources[i + nsource]
 641                     if os.getenv("USE_PLAYBIN3") is None:
 642                         srcs.append(
 643                             "uridecodebin uri=%s ! %s" % (uri, self.converter))
 644                     else:
 645                         srcs.append(
 646                             "uridecodebin3 uri=%s ! %s" % (uri, self.converter))
 647                     fname = os.path.basename(uri).replace(".", "_")
 648                     if not name:
 649                         name = fname
 650                     else:
 651                         name += "+%s" % fname
 652
 653                 self.mixed_srcs[name] = tuple(srcs)
 654
 655         for name, srcs in self.mixed_srcs.items():
 656             if isinstance(srcs, dict):
 657                 pipe_arguments = {
 658                     "mixer": self.mixer + " %s" % srcs["mixer_props"]}
 659                 srcs = srcs["sources"]
 660             else:
 661                 pipe_arguments = {"mixer": self.mixer}
 662
 663             for scenario in scenarios:
 664                 fname = self.get_fname(scenario, Protocols.FILE) + "."
 665                 fname += name
 666
 667                 self.debug("Adding: %s", fname)
 668
 669                 if self.test_manager.options.mute:
 670                     pipe_arguments["sink"] = get_fakesink_for_media_type(self.media_type,
 671                                                                          scenario.needs_clock_sync())
 672                 else:
 673                     pipe_arguments["sink"] = "auto%ssink" % self.media_type
 674
 675                 pipe = self._pipeline_template % pipe_arguments
 676
 677                 for src in srcs:
 678                     pipe += "%s ! _mixer. " % src
 679
 680                 self.add_test(GstValidateLaunchTest(fname,
 681                                                     self.test_manager.options,
 682                                                     self.test_manager.reporter,
 683                                                     pipe,
 684                                                     scenario=scenario)
 685                               )
 686
 687
 688 class GstValidateSimpleTest(GstValidateTest):
 689     def __init__(self, test_file, *args, test_info=None, **kwargs):
 690         self.test_file = test_file
 691         self.test_info = test_info
 692
 693         super().__init__(GstValidateBaseTestManager.COMMAND, *args, **kwargs)
 694
 695     def build_arguments(self):
 696         self.add_arguments('--set-test-file', self.test_file)
 697         if self.options.mute:
 698             self.add_arguments('--use-fakesinks')
 699
 700     def needs_http_server(self):
 701         try:
 702             return bool(self.test_info.needs_http_server)
 703         except AttributeError:
 704             return False
 705
 706
 707 class GstValidateLaunchTest(GstValidateTest):
 708
 709     def __init__(self, classname, options, reporter, pipeline_desc,
 710                  timeout=DEFAULT_TIMEOUT, scenario=None,
 711                  media_descriptor=None, duration=0, hard_timeout=None,
 712                  extra_env_variables=None, expected_issues=None):
 713
 714         self.extra_env_variables = extra_env_variables or {}
 715
 716         if scenario:
 717             duration = scenario.get_duration()
 718         elif media_descriptor:
 719             duration = media_descriptor.get_duration() / GST_SECOND
 720
 721         super(
 722             GstValidateLaunchTest, self).__init__(GstValidateBaseTestManager.COMMAND,
 723                                                   classname,
 724                                                   options, reporter,
 725                                                   duration=duration,
 726                                                   scenario=scenario,
 727                                                   timeout=timeout,
 728                                                   hard_timeout=hard_timeout,
 729                                                   media_descriptor=media_descriptor,
 730                                                   extra_env_variables=extra_env_variables,
 731                                                   expected_issues=expected_issues)
 732
 733         self.pipeline_desc = pipeline_desc
 734         self.media_descriptor = media_descriptor
 735
 736     def build_arguments(self):
 737         GstValidateTest.build_arguments(self)
 738         self.add_arguments(*shlex.split(self.pipeline_desc))
 739         if self.media_descriptor is not None and self.media_descriptor.get_path():
 740             self.add_arguments(
 741                 "--set-media-info", self.media_descriptor.get_path())
 742
 743
 744 class GstValidateMediaCheckTest(GstValidateTest):
 745
 746     def __init__(self, classname, options, reporter, media_descriptor,
 747                  uri, minfo_path, timeout=DEFAULT_TIMEOUT,
 748                  extra_env_variables=None,
 749                  expected_issues=None):
 750         self.extra_env_variables = extra_env_variables or {}
 751
 752         super(
 753             GstValidateMediaCheckTest, self).__init__(GstValidateBaseTestManager.MEDIA_CHECK_COMMAND, classname,
 754                                                       options, reporter,
 755                                                       timeout=timeout,
 756                                                       media_descriptor=media_descriptor,
 757                                                       extra_env_variables=extra_env_variables,
 758                                                       expected_issues=expected_issues)
 759         self._uri = uri
 760         self._media_info_path = minfo_path
 761
 762     def build_arguments(self):
 763         Test.build_arguments(self)
 764         self.add_arguments(self._uri, "--expected-results",
 765                            self._media_info_path)
 766
 767         if self.media_descriptor.skip_parsers():
 768             self.add_arguments("--skip-parsers")
 769
 770
 771 class GstValidateTranscodingTest(GstValidateTest, GstValidateEncodingTestInterface):
 772     scenarios_manager = ScenarioManager()
 773
 774     def __init__(self, classname, options, reporter,
 775                  combination, uri, media_descriptor,
 776                  timeout=DEFAULT_TIMEOUT,
 777                  scenario=None,
 778                  extra_env_variables=None,
 779                  expected_issues=None):
 780         Loggable.__init__(self)
 781
 782         self.extra_env_variables = extra_env_variables or {}
 783
 784         file_dur = int(media_descriptor.get_duration()) / GST_SECOND
 785         if not media_descriptor.get_num_tracks("video"):
 786             self.debug("%s audio only file applying transcoding ratio."
 787                        "File 'duration' : %s" % (classname, file_dur))
 788             duration = file_dur / AUDIO_ONLY_FILE_TRANSCODING_RATIO
 789         else:
 790             duration = file_dur
 791
 792         super(
 793             GstValidateTranscodingTest, self).__init__(GstValidateBaseTestManager.TRANSCODING_COMMAND,
 794                                                        classname,
 795                                                        options,
 796                                                        reporter,
 797                                                        duration=duration,
 798                                                        timeout=timeout,
 799                                                        scenario=scenario,
 800                                                        media_descriptor=media_descriptor,
 801                                                        extra_env_variables=None,
 802                                                        expected_issues=expected_issues)
 803         extra_env_variables = extra_env_variables or {}
 804
 805         GstValidateEncodingTestInterface.__init__(
 806             self, combination, media_descriptor)
 807
 808         self.uri = uri
 809
 810     def run_external_checks(self):
 811         if self.media_descriptor.get_num_tracks("video") == 1 and \
 812                 self.options.validate_enable_iqa_tests:
 813             self.run_iqa_test(self.uri)
 814
 815     def set_rendering_info(self):
 816         self.dest_file = os.path.join(self.options.dest,
 817                                       self.classname.replace(".transcode.", os.sep).
 818                                       replace(".", os.sep))
 819         mkdir(os.path.dirname(urllib.parse.urlsplit(self.dest_file).path))
 820         if urllib.parse.urlparse(self.dest_file).scheme == "":
 821             self.dest_file = path2url(self.dest_file)
 822
 823         variable_framerate = VariableFramerateMode.DISABLED
 824         if self.media_descriptor.get_num_tracks("video") == 1:
 825             caps, = [c for (t, c) in self.media_descriptor.get_tracks_caps() if t == 'video']
 826             framerate = None
 827             for struct, _ in GstCaps.new_from_str(caps):
 828                 framerate = struct.get("framerate", None)
 829                 if framerate is not None and \
 830                         framerate.numerator == 0 and framerate.denominator == 1:
 831                     variable_framerate = VariableFramerateMode.AUTO
 832                     break
 833
 834         profile = self.get_profile(variable_framerate=variable_framerate)
 835         self.add_arguments("-o", profile)
 836
 837     def build_arguments(self):
 838         GstValidateTest.build_arguments(self)
 839         self.set_rendering_info()
 840         self.add_arguments(self.uri, self.dest_file)
 841
 842     def get_current_value(self):
 843         if self.scenario:
 844             sent_eos = self.sent_eos_position()
 845             if sent_eos is not None:
 846                 t = time.time()
 847                 if ((t - sent_eos)) > 30:
 848                     if self.media_descriptor.get_protocol() == Protocols.HLS:
 849                         self.set_result(Result.PASSED,
 850                                         """Got no EOS 30 seconds after sending EOS,
 851                                         in HLS known and tolerated issue:
 852                                         https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/issues/132""")
 853                         return Result.KNOWN_ERROR
 854
 855                     self.set_result(
 856                         Result.FAILED, "Pipeline did not stop 30 Seconds after sending EOS")
 857
 858                     return Result.FAILED
 859
 860         size = self.get_current_size()
 861         if size is None:
 862             return self.get_current_position()
 863
 864         return size
 865
 866     def check_results(self):
 867         self.check_encoded_file()
 868         GstValidateTest.check_results(self)
 869
 870
 871 class GstValidateBaseRTSPTest:
 872     """ Interface for RTSP tests, requires implementing Test"""
 873     __used_ports = set()
 874
 875     def __init__(self, local_uri):
 876         self._local_uri = local_uri
 877         self.rtsp_server = None
 878         self._unsetport_pipeline_desc = None
 879         self.optional = True
 880
 881     @classmethod
 882     def __get_open_port(cls):
 883         while True:
 884             # hackish trick from
 885             # http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python?answertab=votes#tab-top
 886             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 887             s.bind(("", 0))
 888             port = s.getsockname()[1]
 889             if port not in cls.__used_ports:
 890                 cls.__used_ports.add(port)
 891                 s.close()
 892                 return port
 893
 894             s.close()
 895
 896     def launch_server(self):
 897         if self.options.redirect_logs == 'stdout':
 898             self.rtspserver_logs = sys.stdout
 899         elif self.options.redirect_logs == 'stderr':
 900             self.rtspserver_logs = sys.stderr
 901
 902         self.server_port = self.__get_open_port()
 903         command = [GstValidateBaseTestManager.RTSP_SERVER_COMMAND, self._local_uri, '--port', str(self.server_port)]
 904
 905         if self.options.validate_gdb_server:
 906             command = self.use_gdb(command)
 907             self.rtspserver_logs = sys.stdout
 908         elif self.options.redirect_logs:
 909             self.rtspserver_logs = sys.stdout
 910         else:
 911             self.rtspserver_logs = open(self.logfile + '_rtspserver.log', 'w+')
 912             self.extra_logfiles.add(self.rtspserver_logs.name)
 913
 914         server_env = os.environ.copy()
 915
 916         self.rtsp_server = subprocess.Popen(command,
 917                                             stderr=self.rtspserver_logs,
 918                                             stdout=self.rtspserver_logs,
 919                                             env=server_env)
 920         while True:
 921             s = socket.socket()
 922             try:
 923                 s.connect((("127.0.0.1", self.server_port)))
 924                 break
 925             except ConnectionRefusedError:
 926                 time.sleep(0.1)
 927                 continue
 928             finally:
 929                 s.close()
 930
 931         if not self._unsetport_pipeline_desc:
 932             self._unsetport_pipeline_desc = self.pipeline_desc
 933
 934         self.pipeline_desc = self._unsetport_pipeline_desc.replace(
 935             "<RTSPPORTNUMBER>", str(self.server_port))
 936
 937         return ' '.join(command)
 938
 939     def close_logfile(self):
 940         super().close_logfile()
 941         if not self.options.redirect_logs:
 942             self.rtspserver_logs.close()
 943
 944     def process_update(self):
 945         res = super().process_update()
 946         if res:
 947             kill_subprocess(self, self.rtsp_server, DEFAULT_TIMEOUT)
 948             self.__used_ports.remove(self.server_port)
 949
 950         return res
 951
 952
 953 class GstValidateRTSPTest(GstValidateBaseRTSPTest, GstValidateLaunchTest):
 954
 955     def __init__(self, classname, options, reporter, pipeline_desc,
 956                  local_uri, timeout=DEFAULT_TIMEOUT, scenario=None,
 957                  media_descriptor=None, rtsp2=False):
 958         GstValidateLaunchTest.__init__(self, classname, options, reporter,
 959                                        pipeline_desc, timeout, scenario,
 960                                        media_descriptor)
 961         GstValidateBaseRTSPTest.__init__(self, local_uri)
 962         self.rtsp2 = rtsp2
 963
 964     def get_subproc_env(self):
 965         env = super().get_subproc_env()
 966         path = env.get('GST_VALIDATE_SCENARIOS_PATH', '')
 967         override_dir = get_data_file(os.path.join('data', 'scenarios'), 'rtsp_overrides')
 968         env['GST_VALIDATE_SCENARIOS_PATH'] = '%s:%s' % (override_dir, path)
 969         if self.rtsp2:
 970             env['GST_VALIDATE_SCENARIO'] = env.get('GST_VALIDATE_SCENARIO', '') + ':' + 'force_rtsp2'
 971
 972         return env
 973
 974
 975 class GstValidateRTSPMediaDescriptor(GstValidateMediaDescriptor):
 976
 977     def __init__(self, xml_path):
 978         GstValidateMediaDescriptor.__init__(self, xml_path)
 979
 980     def get_uri(self):
 981         return "rtsp://127.0.0.1:8554/test"
 982
 983     def get_protocol(self):
 984         return Protocols.RTSP
 985
 986     def prerrols(self):
 987         return False
 988
 989
 990 class GstValidateTestManager(GstValidateBaseTestManager):
 991
 992     name = "validate"
 993
 994     # List of all classes to create testsuites
 995     GstValidateMediaCheckTestsGenerator = GstValidateMediaCheckTestsGenerator
 996     GstValidateTranscodingTestsGenerator = GstValidateTranscodingTestsGenerator
 997     GstValidatePipelineTestsGenerator = GstValidatePipelineTestsGenerator
 998     GstValidatePlaybinTestsGenerator = GstValidatePlaybinTestsGenerator
 999     GstValidateMixerTestsGenerator = GstValidateMixerTestsGenerator
1000     GstValidateCheckAccurateSeekingTestGenerator = GstValidateCheckAccurateSeekingTestGenerator
1001     GstValidateLaunchTest = GstValidateLaunchTest
1002     GstValidateMediaCheckTest = GstValidateMediaCheckTest
1003     GstValidateTranscodingTest = GstValidateTranscodingTest
1004
1005     def __init__(self):
1006         super(GstValidateTestManager, self).__init__()
1007         self._uris = []
1008         self._run_defaults = True
1009         self._is_populated = False
1010         self._default_generators_registered = False
1011
1012     def init(self):
1013         for command, name in [
1014                 (GstValidateBaseTestManager.TRANSCODING_COMMAND, "gst-validate-transcoding-1.0"),
1015                 (GstValidateBaseTestManager.COMMAND, "gst-validate-1.0"),
1016                 (GstValidateBaseTestManager.MEDIA_CHECK_COMMAND, "gst-validate-media-check-1.0")]:
1017             if not command:
1018                 self.error("command not found: %s" % name)
1019                 return False
1020
1021         return True
1022
1023     def add_options(self, parser):
1024         group = parser.add_argument_group("GstValidate tools specific options"
1025                                           " and behaviours",
1026                                           description="""When using --wanted-tests, all the scenarios can be used, even those which have
1027 not been tested and explicitly activated if you set use --wanted-tests ALL""")
1028         group.add_argument("--validate-check-uri", dest="validate_uris",
1029                            action="append", help="defines the uris to run default tests on")
1030         group.add_argument("--validate-tools-path", dest="validate_tools_path",
1031                            action="append", help="defines the paths to look for GstValidate tools.")
1032         group.add_argument("--validate-gdb-server", dest="validate_gdb_server",
1033                            help="Run the server in GDB.")
1034         group.add_argument("--validate-disable-rtsp", dest="disable_rtsp",
1035                            help="Disable RTSP tests.", default=False, action='store_true')
1036         group.add_argument("--validate-enable-iqa-tests", dest="validate_enable_iqa_tests",
1037                            help="Enable Image Quality Assessment validation tests.",
1038                            default=False, action='store_true')
1039         group.add_argument("--validate-generate-expectations", dest="validate_generate_expectations",
1040                            choices=['auto', 'enabled', 'disabled'],
1041                            help="Force generating expectations (when set to `enabed`)"
1042                                 " force failure on missing expactations when set to `disabled`"
1043                                 " and create if needed when set to `auto`.",
1044                            default='auto')
1045         group.add_argument("--validate-generate-ssim-reference-files",
1046                            help="(re)generate ssim reference image files.",
1047                            default=False, action='store_true')
1048
1049     def print_valgrind_bugs(self):
1050         # Look for all the 'pending' bugs in our supp file
1051         bugs = []
1052         p = get_data_file('data', 'gstvalidate.supp')
1053         with open(p) as f:
1054             for line in f.readlines():
1055                 line = line.strip()
1056                 if line.startswith('# PENDING:'):
1057                     tmp = line.split(' ')
1058                     bugs.append(tmp[2])
1059
1060         if bugs:
1061             msg = "Ignored valgrind bugs:\n"
1062             for b in bugs:
1063                 msg += "  + %s\n" % b
1064             printc(msg, Colors.FAIL, True)
1065
1066     def populate_testsuite(self):
1067
1068         if self._is_populated is True:
1069             return
1070
1071         if not self.options.config and not self.options.testsuites:
1072             if self._run_defaults:
1073                 self.register_defaults()
1074             else:
1075                 self.register_all()
1076
1077         self._is_populated = True
1078
1079     def list_tests(self):
1080         if self.tests:
1081             return self.tests
1082
1083         if self._run_defaults:
1084             scenarios = [self.scenarios_manager.get_scenario(scenario_name)
1085                          for scenario_name in self.get_scenarios()]
1086         else:
1087             scenarios = self.scenarios_manager.get_scenario(None)
1088         uris = self._list_uris()
1089
1090         for generator in self.get_generators():
1091             for test in generator.generate_tests(uris, scenarios):
1092                 self.add_test(test)
1093
1094         if not self.tests and not uris and not self.options.wanted_tests:
1095             self.info("No valid uris present in the path. Check if media files and info files exist")
1096
1097         return self.tests
1098
1099     def _add_media(self, media_info, uri=None):
1100         self.debug("Checking %s", media_info)
1101         if isinstance(media_info, GstValidateMediaDescriptor):
1102             media_descriptor = media_info
1103             media_info = media_descriptor.get_path()
1104         else:
1105             media_descriptor = GstValidateMediaDescriptor(media_info)
1106
1107         try:
1108             # Just testing that the various mandatory infos are present
1109             caps = media_descriptor.get_caps()
1110             if uri is None or media_descriptor.get_protocol() == Protocols.IMAGESEQUENCE:
1111                 uri = media_descriptor.get_uri()
1112
1113             # Adjust local http uri
1114             if self.options.http_server_port != 8079 and \
1115                uri.startswith("http://127.0.0.1:8079/"):
1116                 uri = uri.replace("http://127.0.0.1:8079/",
1117                                   "http://127.0.0.1:%r/" % self.options.http_server_port, 1)
1118             media_descriptor.set_protocol(urllib.parse.urlparse(uri).scheme)
1119             for caps2, prot in GST_VALIDATE_CAPS_TO_PROTOCOL:
1120                 if caps2 == caps:
1121                     media_descriptor.set_protocol(prot)
1122                     break
1123
1124             scenario_bname = media_descriptor.get_media_filepath()
1125             special_scenarios = self.scenarios_manager.find_special_scenarios(
1126                 scenario_bname)
1127             self._uris.append((uri,
1128                                NamedDic({"path": media_info,
1129                                          "media_descriptor": media_descriptor}),
1130                                special_scenarios))
1131         except configparser.NoOptionError as e:
1132             self.debug("Exception: %s for %s", e, media_info)
1133
1134     def _discover_file(self, uri, fpath):
1135         for ext in (GstValidateMediaDescriptor.MEDIA_INFO_EXT,
1136                 GstValidateMediaDescriptor.PUSH_MEDIA_INFO_EXT,
1137                 GstValidateMediaDescriptor.SKIPPED_MEDIA_INFO_EXT):
1138             try:
1139                 is_push = ext == GstValidateMediaDescriptor.PUSH_MEDIA_INFO_EXT
1140                 is_skipped = ext == GstValidateMediaDescriptor.SKIPPED_MEDIA_INFO_EXT
1141                 media_info = "%s.%s" % (fpath, ext)
1142                 if is_push or is_skipped:
1143                     if not os.path.exists(media_info):
1144                         continue
1145                 if is_push:
1146                     uri = "push" + uri
1147                 args = GstValidateBaseTestManager.MEDIA_CHECK_COMMAND.split(" ")
1148                 args.append(uri)
1149                 if os.path.isfile(media_info) and not self.options.update_media_info and not is_skipped:
1150                     self._add_media(media_info, uri)
1151                     continue
1152                 elif fpath.endswith(GstValidateMediaDescriptor.STREAM_INFO_EXT) and not is_skipped:
1153                     self._add_media(fpath)
1154                     continue
1155                 elif not self.options.generate_info and not self.options.update_media_info and not self.options.validate_uris:
1156                     continue
1157                 elif self.options.update_media_info and not os.path.isfile(media_info):
1158                     self.info(
1159                         "%s not present. Use --generate-media-info", media_info)
1160                     continue
1161                 elif os.path.islink(media_info):
1162                     self.info(
1163                         "%s is a symlink, not updating and hopefully the actual file gets updated!", media_info)
1164                     continue
1165
1166                 include_frames = 0
1167                 if self.options.update_media_info:
1168                     include_frames = 2
1169                 elif self.options.generate_info_full:
1170                     include_frames = 1
1171
1172                 media_descriptor = GstValidateMediaDescriptor.new_from_uri(
1173                     uri, True, include_frames, is_push, is_skipped)
1174                 if media_descriptor:
1175                     self._add_media(media_descriptor, uri)
1176                 else:
1177                     self.warning("Could not get any descriptor for %s" % uri)
1178
1179             except subprocess.CalledProcessError as e:
1180                 if self.options.generate_info:
1181                     printc("Result: Failed", Colors.FAIL)
1182                 else:
1183                     self.error("Exception: %s", e)
1184                 return False
1185         return True
1186
1187     def _list_uris(self):
1188         if self._uris:
1189             return self._uris
1190
1191         if self.options.validate_uris:
1192             for uri in self.options.validate_uris:
1193                 self._discover_file(uri, uri)
1194             return self._uris
1195
1196         if not self.args:
1197             if isinstance(self.options.paths, str):
1198                 self.options.paths = [os.path.join(self.options.paths)]
1199
1200             for path in self.options.paths:
1201                 if os.path.isfile(path):
1202                     path = os.path.abspath(path)
1203                     self._discover_file(path2url(path), path)
1204                 else:
1205                     for root, dirs, files in os.walk(path):
1206                         for f in files:
1207                             fpath = os.path.abspath(os.path.join(root, f))
1208                             if os.path.isdir(fpath) or \
1209                                     fpath.endswith(GstValidateMediaDescriptor.MEDIA_INFO_EXT) or\
1210                                     fpath.endswith(ScenarioManager.FILE_EXTENSION):
1211                                 continue
1212                             else:
1213                                 self._discover_file(path2url(fpath), fpath)
1214
1215         self.debug("Uris found: %s", self._uris)
1216
1217         return self._uris
1218
1219     def needs_http_server(self):
1220         for test in self.list_tests():
1221             if self._is_test_wanted(test):
1222                 if test.needs_http_server():
1223                     return True
1224
1225         return False
1226
1227     def set_settings(self, options, args, reporter):
1228         if options.wanted_tests:
1229             for i in range(len(options.wanted_tests)):
1230                 if "ALL" in options.wanted_tests[i]:
1231                     self._run_defaults = False
1232                     options.wanted_tests[
1233                         i] = options.wanted_tests[i].replace("ALL", "")
1234
1235         options.validate_default_config = None
1236         if options.validate_generate_expectations != 'auto':
1237             options.validate_default_config = os.path.join(options.logsdir, "__validate_default.config")
1238             with open(options.validate_default_config, 'w') as f:
1239                 val = "true" if options.validate_generate_expectations == "enabled" else "false"
1240                 print("validateflow,generate-expectations=%s" % val, file=f)
1241         try:
1242             options.wanted_tests.remove("")
1243         except ValueError:
1244             pass
1245
1246         if options.validate_uris or options.validate_generate_ssim_reference_files:
1247             self.check_testslist = False
1248
1249         super(GstValidateTestManager, self).set_settings(
1250             options, args, reporter)
1251
1252     def register_defaults(self):
1253         """
1254         Registers the defaults:
1255             * Scenarios to be used
1256             * Encoding formats to be used
1257             * Blacklisted tests
1258             * Test generators
1259         """
1260         printc("-> Registering default 'validate' tests... ", end='')
1261         self.register_default_scenarios()
1262         self.register_default_encoding_formats()
1263         self.register_default_blacklist()
1264         self.register_default_test_generators()
1265         printc("OK", Colors.OKGREEN)
1266
1267     def register_default_scenarios(self):
1268         """
1269         Registers default test scenarios
1270         """
1271         if self.options.long_limit != 0:
1272             self.add_scenarios([
1273                 "play_15s",
1274                 "reverse_playback",
1275                 "fast_forward",
1276                 "seek_forward",
1277                 "seek_backward",
1278                 "seek_with_stop",
1279                 "switch_audio_track",
1280                 "switch_audio_track_while_paused",
1281                 "switch_subtitle_track",
1282                 "switch_subtitle_track_while_paused",
1283                 "disable_subtitle_track_while_paused",
1284                 "change_state_intensive",
1285                 "scrub_forward_seeking"])
1286         else:
1287             self.add_scenarios([
1288                 "play_15s",
1289                 "reverse_playback",
1290                 "fast_forward",
1291                 "seek_forward",
1292                 "seek_backward",
1293                 "seek_with_stop",
1294                 "switch_audio_track",
1295                 "switch_audio_track_while_paused",
1296                 "switch_subtitle_track",
1297                 "switch_subtitle_track_while_paused",
1298                 "disable_subtitle_track_while_paused",
1299                 "change_state_intensive",
1300                 "scrub_forward_seeking"])
1301
1302     def register_default_encoding_formats(self):
1303         """
1304         Registers default encoding formats
1305         """
1306         self.add_encoding_formats([
1307             MediaFormatCombination("ogg", "vorbis", "theora"),
1308             MediaFormatCombination("webm", "vorbis", "vp8"),
1309             MediaFormatCombination("webm", "vorbis", "vp9",
1310                 video_restriction="video/x-raw,width=160,height=120"),
1311             MediaFormatCombination("mp4", "mp3", "h264"),
1312             MediaFormatCombination("mkv", "vorbis", "h264"),
1313         ])
1314
1315     def register_default_blacklist(self):
1316         self.set_default_blacklist([
1317             # testbin known issues
1318             ("testbin.media_check.*",
1319              "Not supported by GstDiscoverer."),
1320
1321             # dash known issues
1322             ("dash.media_check.*",
1323              "Caps are different depending on selected bitrates, etc"),
1324
1325             # Matroska/WEBM known issues:
1326             ("*.reverse_playback.*webm$",
1327              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/65"),
1328             ("*.reverse_playback.*mkv$",
1329              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/65"),
1330             ("http.playback.seek_with_stop.*webm",
1331              "matroskademux.gst_matroska_demux_handle_seek_push: Seek end-time not supported in streaming mode"),
1332             ("http.playback.seek_with_stop.*mkv",
1333              "matroskademux.gst_matroska_demux_handle_seek_push: Seek end-time not supported in streaming mode"),
1334
1335             # MPEG TS known issues:
1336             ('(?i)*playback.reverse_playback.*(?:_|.)(?:|m)ts$',
1337              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/issues/97"),
1338
1339             # Fragmented MP4 disabled tests:
1340             ('*.playback..*seek.*.fragmented_nonseekable_sink_mp4',
1341              "Seeking on fragmented files without indexes isn't implemented"),
1342             ('*.playback.reverse_playback.fragmented_nonseekable_sink_mp4',
1343              "Seeking on fragmented files without indexes isn't implemented"),
1344
1345             # HTTP known issues:
1346             ("http.*scrub_forward_seeking.*",
1347              "This is not stable enough for now."),
1348             ("http.playback.change_state_intensive.raw_video_mov",
1349              "This is not stable enough for now. (flow return from pad push doesn't match expected value)"),
1350
1351             # MXF known issues"
1352             ("*reverse_playback.*mxf",
1353              "Reverse playback is not handled in MXF"),
1354             ("file\.transcode.*mxf",
1355              "FIXME: Transcoding and mixing tests need to be tested"),
1356
1357             # WMV known issues"
1358             ("*reverse_playback.*wmv",
1359              "Reverse playback is not handled in wmv"),
1360             (".*reverse_playback.*asf",
1361              "Reverse playback is not handled in asf"),
1362
1363             # ogg known issues
1364             ("http.playback.seek.*vorbis_theora_1_ogg",
1365              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/issues/281"),
1366             # RTSP known issues
1367             ('rtsp.*playback.reverse.*',
1368              'https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/32'),
1369             ('rtsp.*playback.seek_with_stop.*',
1370              'https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/issues/386'),
1371             ('rtsp.*playback.fast_*',
1372              'rtpbasedepayload does not handle rate != 1.0 correctly'),
1373         ])
1374
1375     def register_default_test_generators(self):
1376         """
1377         Registers default test generators
1378         """
1379         if self._default_generators_registered:
1380             return
1381
1382         self.add_generators([GstValidatePlaybinTestsGenerator(self),
1383                              GstValidateMediaCheckTestsGenerator(self),
1384                              GstValidateTranscodingTestsGenerator(self)])
1385         self._default_generators_registered = True