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 import os
  21 import copy
  22 import sys
  23 import time
  24 import urllib.parse
  25 import shlex
  26 import socket
  27 import subprocess
  28 import configparser
  29 import json
  30 import glob
  31 import math
  32 from launcher.loggable import Loggable, error
  33
  34 from launcher.baseclasses import GstValidateTest, Test, \
  35     ScenarioManager, NamedDic, GstValidateTestsGenerator, \
  36     GstValidateMediaDescriptor, GstValidateEncodingTestInterface, \
  37     GstValidateBaseTestManager, MediaDescriptor, MediaFormatCombination
  38
  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
  43
  44 #
  45 # Private global variables     #
  46 #
  47
  48 # definitions of commands to use
  49 parser = argparse.ArgumentParser(add_help=False)
  50 parser.add_argument("--validate-tools-path", dest="validate_tools_path",
  51                     default="",
  52                     help="defines the paths to look for GstValidate tools.")
  53 options, args = parser.parse_known_args()
  54
  55 GstValidateBaseTestManager.update_commands(options.validate_tools_path)
  56 AUDIO_ONLY_FILE_TRANSCODING_RATIO = 5
  57
  58 #
  59 # API to be used to create testsuites     #
  60 #
  61
  62 """
  63 Some info about protocols and how to handle them
  64 """
  65 GST_VALIDATE_CAPS_TO_PROTOCOL = [("application/x-hls", Protocols.HLS),
  66                                  ("application/dash+xml", Protocols.DASH)]
  67
  68
  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):
  74             l[i] = v % data
  75         elif isinstance(v, list):
  76             l[i] = expand_vars_in_list_recurse(v, data)
  77
  78     return l
  79
  80
  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)
  89
  90     return dico
  91
  92
  93 class GstValidateMediaCheckTestsGenerator(GstValidateTestsGenerator):
  94
  95     def __init__(self, test_manager):
  96         GstValidateTestsGenerator.__init__(self, "media_check", test_manager)
  97
  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
 102
 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,
 109                                                     uri,
 110                                                     mediainfo.path,
 111                                                     timeout=timeout))
 112
 113
 114 class GstValidateTranscodingTestsGenerator(GstValidateTestsGenerator):
 115     HARD_TIMEOUT_FACTOR = 10
 116
 117     def __init__(self, test_manager):
 118         GstValidateTestsGenerator.__init__(self, "transcode", test_manager)
 119
 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():
 123                 continue
 124
 125             protocol = mediainfo.media_descriptor.get_protocol()
 126             if protocol == Protocols.RTSP:
 127                 continue
 128
 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(),
 132                                                        str(comb).replace(
 133                                                        ' ', '_'),
 134                                                        mediainfo.media_descriptor.get_clean_name())
 135
 136                 self.add_test(GstValidateTranscodingTest(classname,
 137                                                          options,
 138                                                          self.test_manager.reporter,
 139                                                          comb,
 140                                                          uri,
 141                                                          mediainfo.media_descriptor))
 142
 143
 144 class FakeMediaDescriptor(MediaDescriptor):
 145
 146     def __init__(self, infos, pipeline_desc):
 147         MediaDescriptor.__init__(self)
 148         self._infos = infos
 149         self._pipeline_desc = pipeline_desc
 150
 151     def get_path(self):
 152         return self._infos.get('path', None)
 153
 154     def get_tracks_caps(self):
 155         return self._info.get('tracks-caps', [])
 156
 157     def get_media_filepath(self):
 158         return self._infos.get('media-filepath', None)
 159
 160     def get_caps(self):
 161         return self._infos.get('caps', None)
 162
 163     def get_uri(self):
 164         return self._infos.get('uri', None)
 165
 166     def get_duration(self):
 167         return int(self._infos.get('duration', 0)) * GST_SECOND
 168
 169     def get_protocol(self):
 170         return self._infos.get('protocol', "launch_pipeline")
 171
 172     def is_seekable(self):
 173         return self._infos.get('is-seekable', True)
 174
 175     def is_image(self):
 176         return self._infos.get('is-image', False)
 177
 178     def is_live(self):
 179         return self._infos.get('is-live', False)
 180
 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"))
 184
 185     def can_play_reverse(self):
 186         return self._infos.get('plays-reverse', False)
 187
 188
 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)
 193
 194     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 195         for root, _, files in os.walk(self.tests_dir):
 196             for f in files:
 197                 name, ext = os.path.splitext(f)
 198                 if ext != ".validatetest":
 199                     continue
 200
 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))
 207
 208
 209 class GstValidatePipelineTestsGenerator(GstValidateTestsGenerator):
 210
 211     def __init__(self, name, test_manager, pipeline_template=None,
 212                  pipelines_descriptions=None, valid_scenarios=None):
 213         """
 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
 223
 224         @valid_scenarios: A list of scenario name that can be used with that generator
 225         """
 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)
 237             else:
 238                 self._pipelines_descriptions.append(description)
 239         self._valid_scenarios = valid_scenarios
 240
 241     @staticmethod
 242     def get_config_file(config, private_dir, test_name, extra_data):
 243         if isinstance(config, str):
 244             return config
 245
 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))
 251
 252         return config_file
 253
 254     @classmethod
 255     def from_dict(cls, test_manager, name, descriptions, extra_data=None):
 256         """
 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.
 260         """
 261         if extra_data is None:
 262             extra_data = {}
 263
 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,
 268                                             name, test_name)
 269
 270             config_files = {}
 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)
 276
 277             scenarios = []
 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)
 285                 else:
 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')
 291                     if 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)
 298
 299                 if config:
 300                     config_files[scenario_name] = cls.get_config_file(config, test_private_dir, test_name + '.' + scenario_name, extra_data)
 301
 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,
 310                 'timeout': timeout,
 311             })
 312
 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)
 318
 319         return GstValidatePipelineTestsGenerator(name, test_manager, pipelines_descriptions=pipelines_descriptions)
 320
 321     def get_fname(self, scenario, protocol=None, name=None):
 322         if name is None:
 323             name = self.name
 324
 325         if protocol is not None:
 326             protocol_str = "%s." % protocol
 327         else:
 328             protocol_str = ""
 329
 330         if scenario is not None and scenario.name.lower() != "none":
 331             return "%s%s.%s" % (protocol_str, name, scenario.name)
 332
 333         return ("%s.%s.%s" % (protocol_str, self.name, name)).replace("..", ".")
 334
 335     def generate_tests(self, uri_minfo_special_scenarios, scenarios):
 336         if self._valid_scenarios is None:
 337             scenarios = [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]
 341
 342         return super(GstValidatePipelineTestsGenerator, self).generate_tests(
 343             uri_minfo_special_scenarios, scenarios)
 344
 345     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 346
 347         special_scenarios = []
 348         for description in self._pipelines_descriptions:
 349             for s in description.get('extra_data', {}).get('scenarios', []):
 350                 if os.path.isabs(s):
 351                     special_scenarios.append(s)
 352
 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', {})
 358
 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
 363                 # scenario.
 364                 scenarios_to_iterate = extra_data['scenarios'] or [None]
 365             else:
 366                 scenarios_to_iterate = scenarios
 367
 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(
 375                         scenario)
 376                     if tmpscenario is None:
 377                         raise RuntimeError("Could not find scenario file: %s" % scenario)
 378                     scenario = tmpscenario
 379
 380                 if not mediainfo.is_compatible(scenario):
 381                     continue
 382
 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)
 388                 else:
 389                     audiosink = 'autoaudiosink'
 390                     videosink = 'autovideosink'
 391
 392                 pipeline_data.update({'videosink': videosink, 'audiosink': audiosink})
 393                 pipeline_desc = pipeline % pipeline_data
 394
 395                 fname = self.get_fname(
 396                     scenario, protocol=mediainfo.get_protocol(), name=description["name"])
 397
 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,
 403                                              pipeline_desc,
 404                                              scenario=scenario,
 405                                              timeout=timeout,
 406                                              media_descriptor=mediainfo,
 407                                              expected_issues=expected_issues,
 408                                              extra_env_variables=extra_env_vars)
 409                 if config_files:
 410                     test.add_validate_config(config_files[scenario.name if scenario is not None else None])
 411                 self.add_test(test)
 412
 413
 414 class GstValidatePlaybinTestsGenerator(GstValidatePipelineTestsGenerator):
 415
 416     def __init__(self, test_manager):
 417         if os.getenv("USE_PLAYBIN3") is None:
 418             GstValidatePipelineTestsGenerator.__init__(
 419                 self, "playback", test_manager, "playbin")
 420         else:
 421             GstValidatePipelineTestsGenerator.__init__(
 422                 self, "playback", test_manager, "playbin3")
 423
 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()
 427
 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)
 432
 433         return pipe_str
 434
 435     def _get_name(self, scenario, protocol, minfo):
 436         return "%s.%s" % (self.get_fname(scenario,
 437                                          protocol),
 438                           os.path.basename(minfo.media_descriptor.get_clean_name()))
 439
 440     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 441         test_rtsp = GstValidateBaseTestManager.RTSP_SERVER_COMMAND
 442         if not test_rtsp:
 443             printc("\n\nRTSP server not available, you should make sure"
 444                    " that %s is available in your $PATH." % GstValidateBaseTestManager.RTSP_SERVER_COMMAND,
 445                    Colors.FAIL)
 446         elif self.test_manager.options.disable_rtsp:
 447             printc("\n\nRTSP tests are disabled")
 448             test_rtsp = False
 449
 450         for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
 451             pipe = self._pipeline_template
 452             protocol = minfo.media_descriptor.get_protocol()
 453
 454             if protocol == Protocols.RTSP:
 455                 self.debug("SKIPPING %s as it is a RTSP stream" % uri)
 456                 continue
 457
 458             pipe += " uri=%s" % uri
 459
 460             for scenario in special_scenarios + scenarios:
 461                 cpipe = pipe
 462                 if not minfo.media_descriptor.is_compatible(scenario):
 463                     continue
 464
 465                 cpipe = self._set_sinks(minfo, cpipe, scenario)
 466                 fname = self._get_name(scenario, protocol, minfo)
 467
 468                 self.debug("Adding: %s", fname)
 469
 470                 if scenario.does_reverse_playback() and protocol == Protocols.HTTP:
 471                     # 10MB so we can reverse playback
 472                     cpipe += " ring-buffer-max-size=10485760"
 473
 474                 self.add_test(GstValidateLaunchTest(fname,
 475                                                     self.test_manager.options,
 476                                                     self.test_manager.reporter,
 477                                                     cpipe,
 478                                                     scenario=scenario,
 479                                                     media_descriptor=minfo.media_descriptor)
 480                               )
 481
 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):
 486                         continue
 487
 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)
 491
 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))
 496
 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,
 502                         rtsp2=True))
 503
 504
 505 class GstValidateCheckAccurateSeekingTestGenerator(GstValidatePipelineTestsGenerator):
 506     def __new__(cls, name, test_manager, media_infos, extra_data=None):
 507         pipelines = {}
 508
 509         for path, reference_frame_dir in media_infos:
 510             media_info = GstValidateMediaDescriptor(path)
 511             media_info.set_protocol("file")
 512             if not media_info:
 513                 error("GstValidateCheckAccurateSeekingTestGenerator",
 514                       "Could not create a media info file from %s" % path)
 515                 continue
 516
 517             if media_info.is_image():
 518                 error("GstValidateCheckAccurateSeekingTestGenerator",
 519                       "%s is an image, can't run accurate seeking tests" % path)
 520                 continue
 521
 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)
 525                 continue
 526
 527             if media_info.get_num_tracks("video") < 1:
 528                 error("GstValidateCheckAccurateSeekingTestGenerator",
 529                       "No video track, can't run accurate seeking tests" % path)
 530                 continue
 531
 532             if test_manager.options.validate_generate_ssim_reference_files:
 533                 scenario = None
 534                 test_name = media_info.get_clean_name() + '.generate_reference_files'
 535                 config = [
 536                     'validatessim, element-name="videoconvert", output-dir="%s"' % reference_frame_dir]
 537             else:
 538                 test_name = media_info.get_clean_name()
 539                 framerate, scenario = cls.generate_scenario(test_manager.options, reference_frame_dir, media_info)
 540                 if scenario is None:
 541                     error("GstValidateCheckAccurateSeekingTestGenerator",
 542                         "Could not generate test for media info: %s" % path)
 543                     continue
 544
 545                 config = [
 546                     '%(ssim)s, element-name="videoconvert", reference-images-dir="'
 547                     + reference_frame_dir + '", framerate=%d/%d' % (framerate.numerator, framerate.denominator)
 548                 ]
 549
 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,
 553                 "config": config,
 554             }
 555
 556             if scenario:
 557                 pipelines[test_name]["scenarios"] = [scenario]
 558
 559         return GstValidatePipelineTestsGenerator.from_dict(test_manager, name, pipelines, extra_data=extra_data)
 560
 561     @classmethod
 562     def generate_scenario(cls, options, reference_frame_dir, media_info):
 563         actions = [
 564             "description, seek=true, handles-states=true, needs_preroll=true",
 565             "pause",
 566         ]
 567
 568         framerate = None
 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"]
 573             if framerate:
 574                 break
 575         assert framerate
 576
 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]
 586
 587         actions += ['seek, flags=flush+accurate, start=(guint64)%s' % ts for ts in frames_timestamps]
 588         actions += ['stop']
 589
 590         return framerate, {
 591             "name": "check_accurate_seek",
 592             "actions": actions,
 593         }
 594
 595
 596 class GstValidateMixerTestsGenerator(GstValidatePipelineTestsGenerator):
 597
 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 []
 602
 603         pipe_template = "%(mixer)s name=_mixer !  " + \
 604             converter + " ! %(sink)s "
 605         self.converter = converter
 606         self.mixer = mixer
 607         self.media_type = media_type
 608         self.num_sources = num_sources
 609         self.mixed_srcs = mixed_srcs
 610         super(
 611             GstValidateMixerTestsGenerator, self).__init__(name, test_manager, pipe_template,
 612                                                            valid_scenarios=valid_scenarios)
 613
 614     def populate_tests(self, uri_minfo_special_scenarios, scenarios):
 615         if self.test_manager.options.validate_uris:
 616             return
 617
 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))
 624
 625         if not self.mixed_srcs:
 626             if not wanted_ressources:
 627                 return
 628
 629             for i in range(len(uri_minfo_special_scenarios) / self.num_sources):
 630                 srcs = []
 631                 name = ""
 632                 for nsource in range(self.num_sources):
 633                     uri, minfo = wanted_ressources[i + nsource]
 634                     if os.getenv("USE_PLAYBIN3") is None:
 635                         srcs.append(
 636                             "uridecodebin uri=%s ! %s" % (uri, self.converter))
 637                     else:
 638                         srcs.append(
 639                             "uridecodebin3 uri=%s ! %s" % (uri, self.converter))
 640                     fname = os.path.basename(uri).replace(".", "_")
 641                     if not name:
 642                         name = fname
 643                     else:
 644                         name += "+%s" % fname
 645
 646                 self.mixed_srcs[name] = tuple(srcs)
 647
 648         for name, srcs in self.mixed_srcs.items():
 649             if isinstance(srcs, dict):
 650                 pipe_arguments = {
 651                     "mixer": self.mixer + " %s" % srcs["mixer_props"]}
 652                 srcs = srcs["sources"]
 653             else:
 654                 pipe_arguments = {"mixer": self.mixer}
 655
 656             for scenario in scenarios:
 657                 fname = self.get_fname(scenario, Protocols.FILE) + "."
 658                 fname += name
 659
 660                 self.debug("Adding: %s", fname)
 661
 662                 if self.test_manager.options.mute:
 663                     pipe_arguments["sink"] = get_fakesink_for_media_type(self.media_type,
 664                                                                          scenario.needs_clock_sync())
 665                 else:
 666                     pipe_arguments["sink"] = "auto%ssink" % self.media_type
 667
 668                 pipe = self._pipeline_template % pipe_arguments
 669
 670                 for src in srcs:
 671                     pipe += "%s ! _mixer. " % src
 672
 673                 self.add_test(GstValidateLaunchTest(fname,
 674                                                     self.test_manager.options,
 675                                                     self.test_manager.reporter,
 676                                                     pipe,
 677                                                     scenario=scenario)
 678                               )
 679
 680
 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)
 685
 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')
 690
 691
 692 class GstValidateLaunchTest(GstValidateTest):
 693
 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):
 698
 699         self.extra_env_variables = extra_env_variables or {}
 700
 701         if scenario:
 702             duration = scenario.get_duration()
 703         elif media_descriptor:
 704             duration = media_descriptor.get_duration() / GST_SECOND
 705
 706         super(
 707             GstValidateLaunchTest, self).__init__(GstValidateBaseTestManager.COMMAND,
 708                                                   classname,
 709                                                   options, reporter,
 710                                                   duration=duration,
 711                                                   scenario=scenario,
 712                                                   timeout=timeout,
 713                                                   hard_timeout=hard_timeout,
 714                                                   media_descriptor=media_descriptor,
 715                                                   extra_env_variables=extra_env_variables,
 716                                                   expected_issues=expected_issues)
 717
 718         self.pipeline_desc = pipeline_desc
 719         self.media_descriptor = media_descriptor
 720
 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():
 725             self.add_arguments(
 726                 "--set-media-info", self.media_descriptor.get_path())
 727
 728
 729 class GstValidateMediaCheckTest(GstValidateTest):
 730
 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 {}
 736
 737         super(
 738             GstValidateMediaCheckTest, self).__init__(GstValidateBaseTestManager.MEDIA_CHECK_COMMAND, classname,
 739                                                       options, reporter,
 740                                                       timeout=timeout,
 741                                                       media_descriptor=media_descriptor,
 742                                                       extra_env_variables=extra_env_variables,
 743                                                       expected_issues=expected_issues)
 744         self._uri = uri
 745         self._media_info_path = minfo_path
 746
 747     def build_arguments(self):
 748         Test.build_arguments(self)
 749         self.add_arguments(self._uri, "--expected-results",
 750                            self._media_info_path)
 751
 752         if self.media_descriptor.skip_parsers():
 753             self.add_arguments("--skip-parsers")
 754
 755
 756 class GstValidateTranscodingTest(GstValidateTest, GstValidateEncodingTestInterface):
 757     scenarios_manager = ScenarioManager()
 758
 759     def __init__(self, classname, options, reporter,
 760                  combination, uri, media_descriptor,
 761                  timeout=DEFAULT_TIMEOUT,
 762                  scenario=None,
 763                  extra_env_variables=None,
 764                  expected_issues=None):
 765         Loggable.__init__(self)
 766
 767         self.extra_env_variables = extra_env_variables or {}
 768
 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
 774         else:
 775             duration = file_dur
 776
 777         super(
 778             GstValidateTranscodingTest, self).__init__(GstValidateBaseTestManager.TRANSCODING_COMMAND,
 779                                                        classname,
 780                                                        options,
 781                                                        reporter,
 782                                                        duration=duration,
 783                                                        timeout=timeout,
 784                                                        scenario=scenario,
 785                                                        media_descriptor=media_descriptor,
 786                                                        extra_env_variables=None,
 787                                                        expected_issues=expected_issues)
 788         extra_env_variables = extra_env_variables or {}
 789
 790         GstValidateEncodingTestInterface.__init__(
 791             self, combination, media_descriptor)
 792
 793         self.uri = uri
 794
 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)
 799
 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)
 807
 808         profile = self.get_profile()
 809         self.add_arguments("-o", profile)
 810
 811     def build_arguments(self):
 812         GstValidateTest.build_arguments(self)
 813         self.set_rendering_info()
 814         self.add_arguments(self.uri, self.dest_file)
 815
 816     def get_current_value(self):
 817         if self.scenario:
 818             sent_eos = self.sent_eos_position()
 819             if sent_eos is not None:
 820                 t = time.time()
 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
 828
 829                     self.set_result(
 830                         Result.FAILED, "Pipeline did not stop 30 Seconds after sending EOS")
 831
 832                     return Result.FAILED
 833
 834         size = self.get_current_size()
 835         if size is None:
 836             return self.get_current_position()
 837
 838         return size
 839
 840     def check_results(self):
 841         self.check_encoded_file()
 842         GstValidateTest.check_results(self)
 843
 844
 845 class GstValidateBaseRTSPTest:
 846     """ Interface for RTSP tests, requires implementing Test"""
 847     __used_ports = set()
 848
 849     def __init__(self, local_uri):
 850         self._local_uri = local_uri
 851         self.rtsp_server = None
 852         self._unsetport_pipeline_desc = None
 853         self.optional = True
 854
 855     @classmethod
 856     def __get_open_port(cls):
 857         while True:
 858             # hackish trick from
 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)
 861             s.bind(("", 0))
 862             port = s.getsockname()[1]
 863             if port not in cls.__used_ports:
 864                 cls.__used_ports.add(port)
 865                 s.close()
 866                 return port
 867
 868             s.close()
 869
 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
 875
 876         self.server_port = self.__get_open_port()
 877         command = [GstValidateBaseTestManager.RTSP_SERVER_COMMAND, self._local_uri, '--port', str(self.server_port)]
 878
 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
 884         else:
 885             self.rtspserver_logs = open(self.logfile + '_rtspserver.log', 'w+')
 886             self.extra_logfiles.add(self.rtspserver_logs.name)
 887
 888         server_env = os.environ.copy()
 889
 890         self.rtsp_server = subprocess.Popen(command,
 891                                             stderr=self.rtspserver_logs,
 892                                             stdout=self.rtspserver_logs,
 893                                             env=server_env)
 894         while True:
 895             s = socket.socket()
 896             try:
 897                 s.connect((("127.0.0.1", self.server_port)))
 898                 break
 899             except ConnectionRefusedError:
 900                 time.sleep(0.1)
 901                 continue
 902             finally:
 903                 s.close()
 904
 905         if not self._unsetport_pipeline_desc:
 906             self._unsetport_pipeline_desc = self.pipeline_desc
 907
 908         self.pipeline_desc = self._unsetport_pipeline_desc.replace(
 909             "<RTSPPORTNUMBER>", str(self.server_port))
 910
 911         return ' '.join(command)
 912
 913     def close_logfile(self):
 914         super().close_logfile()
 915         if not self.options.redirect_logs:
 916             self.rtspserver_logs.close()
 917
 918     def process_update(self):
 919         res = super().process_update()
 920         if res:
 921             kill_subprocess(self, self.rtsp_server, DEFAULT_TIMEOUT)
 922             self.__used_ports.remove(self.server_port)
 923
 924         return res
 925
 926
 927 class GstValidateRTSPTest(GstValidateBaseRTSPTest, GstValidateLaunchTest):
 928
 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,
 934                                        media_descriptor)
 935         GstValidateBaseRTSPTest.__init__(self, local_uri)
 936         self.rtsp2 = rtsp2
 937
 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)
 943         if self.rtsp2:
 944             env['GST_VALIDATE_SCENARIO'] = env.get('GST_VALIDATE_SCENARIO', '') + ':' + 'force_rtsp2'
 945
 946         return env
 947
 948
 949 class GstValidateRTSPMediaDescriptor(GstValidateMediaDescriptor):
 950
 951     def __init__(self, xml_path):
 952         GstValidateMediaDescriptor.__init__(self, xml_path)
 953
 954     def get_uri(self):
 955         return "rtsp://127.0.0.1:8554/test"
 956
 957     def get_protocol(self):
 958         return Protocols.RTSP
 959
 960     def prerrols(self):
 961         return False
 962
 963
 964 class GstValidateTestManager(GstValidateBaseTestManager):
 965
 966     name = "validate"
 967
 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
 978
 979     def __init__(self):
 980         super(GstValidateTestManager, self).__init__()
 981         self._uris = []
 982         self._run_defaults = True
 983         self._is_populated = False
 984         self._default_generators_registered = False
 985
 986     def init(self):
 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")]:
 991             if not command:
 992                 self.error("command not found: %s" % name)
 993                 return False
 994
 995         return True
 996
 997     def add_options(self, parser):
 998         group = parser.add_argument_group("GstValidate tools specific options"
 999                                           " and behaviours",
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`.",
1018                            default='auto')
1019         group.add_argument("--validate-generate-ssim-reference-files",
1020                            help="(re)generate ssim reference image files.",
1021                            default=False, action='store_true')
1022
1023     def print_valgrind_bugs(self):
1024         # Look for all the 'pending' bugs in our supp file
1025         bugs = []
1026         p = get_data_file('data', 'gstvalidate.supp')
1027         with open(p) as f:
1028             for line in f.readlines():
1029                 line = line.strip()
1030                 if line.startswith('# PENDING:'):
1031                     tmp = line.split(' ')
1032                     bugs.append(tmp[2])
1033
1034         if bugs:
1035             msg = "Ignored valgrind bugs:\n"
1036             for b in bugs:
1037                 msg += "  + %s\n" % b
1038             printc(msg, Colors.FAIL, True)
1039
1040     def populate_testsuite(self):
1041
1042         if self._is_populated is True:
1043             return
1044
1045         if not self.options.config and not self.options.testsuites:
1046             if self._run_defaults:
1047                 self.register_defaults()
1048             else:
1049                 self.register_all()
1050
1051         self._is_populated = True
1052
1053     def list_tests(self):
1054         if self.tests:
1055             return self.tests
1056
1057         if self._run_defaults:
1058             scenarios = [self.scenarios_manager.get_scenario(scenario_name)
1059                          for scenario_name in self.get_scenarios()]
1060         else:
1061             scenarios = self.scenarios_manager.get_scenario(None)
1062         uris = self._list_uris()
1063
1064         for generator in self.get_generators():
1065             for test in generator.generate_tests(uris, scenarios):
1066                 self.add_test(test)
1067
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")
1070
1071         return self.tests
1072
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()
1078         else:
1079             media_descriptor = GstValidateMediaDescriptor(media_info)
1080
1081         try:
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()
1086
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:
1094                 if caps2 == caps:
1095                     media_descriptor.set_protocol(prot)
1096                     break
1097
1098             scenario_bname = media_descriptor.get_media_filepath()
1099             special_scenarios = self.scenarios_manager.find_special_scenarios(
1100                 scenario_bname)
1101             self._uris.append((uri,
1102                                NamedDic({"path": media_info,
1103                                          "media_descriptor": media_descriptor}),
1104                                special_scenarios))
1105         except configparser.NoOptionError as e:
1106             self.debug("Exception: %s for %s", e, media_info)
1107
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):
1112             try:
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):
1118                         continue
1119                 if is_push:
1120                     uri = "push" + uri
1121                 args = GstValidateBaseTestManager.MEDIA_CHECK_COMMAND.split(" ")
1122                 args.append(uri)
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)
1125                     continue
1126                 elif fpath.endswith(GstValidateMediaDescriptor.STREAM_INFO_EXT) and not is_skipped:
1127                     self._add_media(fpath)
1128                     continue
1129                 elif not self.options.generate_info and not self.options.update_media_info and not self.options.validate_uris:
1130                     continue
1131                 elif self.options.update_media_info and not os.path.isfile(media_info):
1132                     self.info(
1133                         "%s not present. Use --generate-media-info", media_info)
1134                     continue
1135                 elif os.path.islink(media_info):
1136                     self.info(
1137                         "%s is a symlink, not updating and hopefully the actual file gets updated!", media_info)
1138                     continue
1139
1140                 include_frames = 0
1141                 if self.options.update_media_info:
1142                     include_frames = 2
1143                 elif self.options.generate_info_full:
1144                     include_frames = 1
1145
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)
1150                 else:
1151                     self.warning("Could not get any descriptor for %s" % uri)
1152
1153             except subprocess.CalledProcessError as e:
1154                 if self.options.generate_info:
1155                     printc("Result: Failed", Colors.FAIL)
1156                 else:
1157                     self.error("Exception: %s", e)
1158                 return False
1159         return True
1160
1161     def _list_uris(self):
1162         if self._uris:
1163             return self._uris
1164
1165         if self.options.validate_uris:
1166             for uri in self.options.validate_uris:
1167                 self._discover_file(uri, uri)
1168             return self._uris
1169
1170         if not self.args:
1171             if isinstance(self.options.paths, str):
1172                 self.options.paths = [os.path.join(self.options.paths)]
1173
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)
1178                 else:
1179                     for root, dirs, files in os.walk(path):
1180                         for f in files:
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):
1185                                 continue
1186                             else:
1187                                 self._discover_file(path2url(fpath), fpath)
1188
1189         self.debug("Uris found: %s", self._uris)
1190
1191         return self._uris
1192
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
1199                 if uri:
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:
1206                     return True
1207         return False
1208
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", "")
1216
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)
1223         try:
1224             options.wanted_tests.remove("")
1225         except ValueError:
1226             pass
1227
1228         if options.validate_uris or options.validate_generate_ssim_reference_files:
1229             self.check_testslist = False
1230
1231         super(GstValidateTestManager, self).set_settings(
1232             options, args, reporter)
1233
1234     def register_defaults(self):
1235         """
1236         Registers the defaults:
1237             * Scenarios to be used
1238             * Encoding formats to be used
1239             * Blacklisted tests
1240             * Test generators
1241         """
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)
1248
1249     def register_default_scenarios(self):
1250         """
1251         Registers default test scenarios
1252         """
1253         if self.options.long_limit != 0:
1254             self.add_scenarios([
1255                 "play_15s",
1256                 "reverse_playback",
1257                 "fast_forward",
1258                 "seek_forward",
1259                 "seek_backward",
1260                 "seek_with_stop",
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"])
1268         else:
1269             self.add_scenarios([
1270                 "play_15s",
1271                 "reverse_playback",
1272                 "fast_forward",
1273                 "seek_forward",
1274                 "seek_backward",
1275                 "seek_with_stop",
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"])
1283
1284     def register_default_encoding_formats(self):
1285         """
1286         Registers default encoding formats
1287         """
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"),
1295         ])
1296
1297     def register_default_blacklist(self):
1298         self.set_default_blacklist([
1299             # testbin known issues
1300             ("testbin.media_check.*",
1301              "Not supported by GstDiscoverer."),
1302
1303             # dash known issues
1304             ("dash.media_check.*",
1305              "Caps are different depending on selected bitrates, etc"),
1306
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"),
1316
1317             # MPEG TS known issues:
1318             ('(?i)*playback.reverse_playback.*(?:_|.)(?:|m)ts$',
1319              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/issues/97"),
1320
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"),
1326
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)"),
1332
1333             # MXF known issues"
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"),
1338
1339             # WMV known issues"
1340             ("*reverse_playback.*wmv",
1341              "Reverse playback is not handled in wmv"),
1342             (".*reverse_playback.*asf",
1343              "Reverse playback is not handled in asf"),
1344
1345             # ogg known issues
1346             ("http.playback.seek.*vorbis_theora_1_ogg",
1347              "https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/issues/281"),
1348             # RTSP known issues
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'),
1355         ])
1356
1357     def register_default_test_generators(self):
1358         """
1359         Registers default test generators
1360         """
1361         if self._default_generators_registered:
1362             return
1363
1364         self.add_generators([GstValidatePlaybinTestsGenerator(self),
1365                              GstValidateMediaCheckTestsGenerator(self),
1366                              GstValidateTranscodingTestsGenerator(self)])
1367         self._default_generators_registered = True