validate:launcher: Move the notion of test generator to the baseclasses

This can be very usefull for all the TestManager and thus exposes a
higher level API for test writers.
This commit is contained in:
Thibault Saunier 2014-06-26 12:42:38 +02:00
parent aeec108c88
commit af383ad0c0
4 changed files with 203 additions and 157 deletions

View file

@ -24,7 +24,8 @@ import ConfigParser
import xml.etree.ElementTree as ET
from loggable import Loggable
from baseclasses import GstValidateTest, TestsManager, Test, ScenarioManager, NamedDic
from baseclasses import GstValidateTest, TestsManager, Test, \
ScenarioManager, NamedDic, GstValidateTestsGenerator
from utils import MediaFormatCombination, get_profile,\
path2url, DEFAULT_TIMEOUT, which, GST_SECOND, Result, \
compare_rendered_with_original, Protocols
@ -51,29 +52,6 @@ G_V_STREAM_INFO_EXT = "stream_info"
# API to be used to create testsuites #
#################################################
"""
A list of tuple of the form:
(@regex_defining_blacklister_test_names, @reson_for_the_blacklisting)
"""
GST_VALIDATE_BLACKLISTED_TESTS = []
"""
A list of scenario names to be run
"""
GST_VALIDATE_SCENARIOS = []
"""
A list of #GstValidateTestGenerator to be used to generate tests
"""
GST_VALIDATE_TEST_GENERATORS = []
"""
A list of #MediaFormatCombinations describing wanted output
formats for transcoding test
"""
GST_VALIDATE_ENCODING_FORMATS = []
"""
Some info about protocols and how to handle them
"""
@ -146,32 +124,10 @@ class GstValidateMediaDescriptor(Loggable):
return True
class GstValidateTestGenerator(Loggable):
def __init__(self, name, tests=[]):
Loggable.__init__(self)
self.name = name
self._tests = tests
self.set_config(None, None)
def set_config(self, options, reporter):
self.options = options
self.reporter = reporter
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
pass
def generate_tests(self, uri_minfo_special_scenarios, scenarios):
self.populate_tests(uri_minfo_special_scenarios, scenarios)
return self._tests
def add_test(self, test):
self._tests.append(test)
class GstValidateMediaCheckTestGenerator(GstValidateTestGenerator):
def __init__(self):
GstValidateTestGenerator.__init__(self, "media_check")
class GstValidateMediaCheckTestsGenerator(GstValidateTestsGenerator):
def __init__(self, test_manager):
GstValidateTestsGenerator.__init__(self, "media_check", test_manager)
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
@ -184,37 +140,37 @@ class GstValidateMediaCheckTestGenerator(GstValidateTestGenerator):
classname = "validate.%s.media_check.%s" % (protocol,
os.path.basename(uri).replace(".", "_"))
self.add_test(GstValidateMediaCheckTest(classname,
self.options,
self.reporter,
self.test_manager.options,
self.test_manager.reporter,
mediainfo.media_descriptor,
uri,
mediainfo.path,
timeout=timeout))
class GstValidateTranscodingTestGenerator(GstValidateTestGenerator):
def __init__(self):
GstValidateTestGenerator.__init__(self, "transcode")
class GstValidateTranscodingTestsGenerator(GstValidateTestsGenerator):
def __init__(self, test_manager):
GstValidateTestsGenerator.__init__(self, "transcode", test_manager)
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
for uri, mediainfo, special_scenarios in uri_minfo_special_scenarios:
if mediainfo.media_descriptor.is_image():
continue
for comb in GST_VALIDATE_ENCODING_FORMATS:
for comb in self.test_manager.get_encoding_formats():
classname = "validate.%s.transcode.to_%s.%s" % (mediainfo.media_descriptor.get_protocol(),
str(comb).replace(' ', '_'),
os.path.basename(uri).replace(".", "_"))
self.add_test(GstValidateTranscodingTest(classname,
self.options,
self.reporter,
self.test_manager.options,
self.test_manager.reporter,
comb,
uri,
mediainfo.media_descriptor))
class GstValidatePipelineTestGenerator(GstValidateTestGenerator):
def __init__(self, name, pipeline_template=None, pipelines_descriptions=None,
class GstValidatePipelineTestsGenerator(GstValidateTestsGenerator):
def __init__(self, name, test_manager, pipeline_template=None, pipelines_descriptions=None,
valid_scenarios=[]):
"""
@name: The name of the generator
@ -223,7 +179,7 @@ class GstValidatePipelineTestGenerator(GstValidateTestGenerator):
(test_name, pipeline_description)
@valid_scenarios: A list of scenario name that can be used with that generator
"""
GstValidateTestGenerator.__init__(self, name)
GstValidateTestsGenerator.__init__(self, name, test_manager)
self._pipeline_template = pipeline_template
self._pipelines_descriptions = pipelines_descriptions
self._valid_scenarios = valid_scenarios
@ -248,7 +204,7 @@ class GstValidatePipelineTestGenerator(GstValidateTestGenerator):
scenarios = [scenario for scenario in scenarios if
scenario.name in self._valid_scenarios]
return super(GstValidatePipelineTestGenerator, self).generate_tests(
return super(GstValidatePipelineTestsGenerator, self).generate_tests(
uri_minfo_special_scenarios, scenarios)
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
@ -256,24 +212,24 @@ class GstValidatePipelineTestGenerator(GstValidateTestGenerator):
for scenario in scenarios:
fname = self.get_fname(scenario, name=name)
self.add_test(GstValidateLaunchTest(fname,
self.options,
self.reporter,
self.test_manager.options,
self.test_manager.reporter,
pipeline,
scenario=scenario)
)
class GstValidatePlaybinTestGenerator(GstValidatePipelineTestGenerator):
class GstValidatePlaybinTestsGenerator(GstValidatePipelineTestsGenerator):
def __init__(self):
GstValidatePipelineTestGenerator.__init__(self, "playback", "playbin")
def __init__(self, test_manager):
GstValidatePipelineTestsGenerator.__init__(self, "playback", test_manager, "playbin")
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
for uri, minfo, special_scenarios in uri_minfo_special_scenarios:
pipe = self._pipeline_template
protocol = minfo.media_descriptor.get_protocol()
if self.options.mute:
if self.test_manager.options.mute:
fakesink = "'fakesink sync=true'"
pipe += " audio-sink=%s video-sink=%s" %(fakesink, fakesink)
@ -292,8 +248,8 @@ class GstValidatePlaybinTestGenerator(GstValidatePipelineTestGenerator):
pipe += " ring-buffer-max-size=10485760"
self.add_test(GstValidateLaunchTest(fname,
self.options,
self.reporter,
self.test_manager.options,
self.test_manager.reporter,
pipe,
scenario=scenario,
media_descriptor=minfo.media_descriptor)
@ -364,7 +320,7 @@ class GstValidateMediaCheckTest(Test):
class GstValidateTranscodingTest(GstValidateTest):
_scenarios = ScenarioManager()
scenarios_manager = ScenarioManager()
def __init__(self, classname, options, reporter,
combination, uri, media_descriptor,
timeout=DEFAULT_TIMEOUT,
@ -437,18 +393,17 @@ class GstValidateTranscodingTest(GstValidateTest):
self.set_result(res, msg)
class GstValidateManager(TestsManager, Loggable):
class GstValidateTestManager(GstValidateBaseTestManager):
name = "validate"
_scenarios = ScenarioManager()
def __init__(self):
TestsManager.__init__(self)
Loggable.__init__(self)
super(GstValidateTestManager, self).__init__()
self._uris = []
self._run_defaults = True
self._is_populated = False
execfile(os.path.join(os.path.dirname(__file__), "apps",
"validate_default_testsuite.py"), globals())
def init(self):
if which(GST_VALIDATE_COMMAND) and which(GST_VALIDATE_TRANSCODING_COMMAND):
@ -461,38 +416,13 @@ class GstValidateManager(TestsManager, Loggable):
description="""When using --wanted-tests, all the scenarios can be used, even those which have
not been tested and explicitely activated if you set use --wanted-tests ALL""")
group.add_argument("-vc", "--validate-config", dest="validate_config",
default=None,
help="""Lets you specify a file where the testsuite to execute is defined.
In this file, your will be able to access the following variables:
* GST_VALIDATE_SCENARIOS: A list of scenario names to be run
* GST_VALIDATE_BLACKLISTED_TESTS: A list of tuple of the form:
(@regex_defining_blacklister_test_names, @reason_for_the_blacklisting)
* GST_VALIDATE_TEST_GENERATORS: A list of #GstValidateTestGenerator to be used to generate tests
* GST_VALIDATE_ENCODING_FORMATS: A list of #MediaFormatCombination to be used for transcoding tests
You can also set default values with:
* gst_validate_register_defaults: Sets default values for all parametters
* gst_validate_register_default_test_generators: Sets default values for the TestGenerators to be used
* gst_validate_register_default_scenarios: Sets default values for the scenarios to be executed
* gst_validate_register_default_encoding_formats: Sets default values for the encoding formats to be tested
Note: In the config file, you have acces to the options variable resulting from the parsing of the command line
user argument, you can thus overrides command line options using that.
""")
def _populate_testsuite(self, options):
def populate_testsuite(self):
if self._is_populated is True:
return
execfile(os.path.join(os.path.dirname(__file__), "apps",
"validate_default_testsuite.py"), globals())
if options.validate_config:
globals()["options"] = options
execfile(options.validate_config, globals())
else:
gst_validate_register_defaults()
if not self.options.config:
self.register_defaults()
self._is_populated = True
@ -500,17 +430,14 @@ user argument, you can thus overrides command line options using that.
if self.tests:
return self.tests
self._populate_testsuite(self.options)
if self._run_defaults:
scenarios = [self._scenarios.get_scenario(scenario_name)
for scenario_name in GST_VALIDATE_SCENARIOS]
scenarios = [self.scenarios_manager.get_scenario(scenario_name)
for scenario_name in self.get_scenarios()]
else:
scenarios = self._scenarios.get_scenario(None)
scenarios = self.scenarios_manager.get_scenario(None)
uris = self._list_uris()
for generator in GST_VALIDATE_TEST_GENERATORS:
generator.set_config(self.options, self.reporter)
for generator in self.get_generators():
for test in generator.generate_tests(uris, scenarios):
self.add_test(test)
@ -532,7 +459,7 @@ user argument, you can thus overrides command line options using that.
break
scenario_bname = media_descriptor.get_media_filepath()
special_scenarios = self._scenarios.find_special_scenarios(scenario_bname)
special_scenarios = self.scenarios_manager.find_special_scenarios(scenario_bname)
self._uris.append((uri,
NamedDic({"path": media_info,
"media_descriptor": media_descriptor}),
@ -556,13 +483,23 @@ user argument, you can thus overrides command line options using that.
else:
return True
subprocess.check_output(args)
if self.options.generate_info:
printc("Generating media info for %s\n"
" Command: '%s'" % (fpath, ' '.join(args)),
Colors.OKBLUE)
out = subprocess.check_output(args, stderr=open(os.devnull))
self._check_discovering_info(media_info, uri)
if self.options.generate_info:
printc("Result: Passed", Colors.OKGREEN)
return True
except subprocess.CalledProcessError as e:
self.debug("Exception: %s", e)
if self.options.generate_info:
printc("Result: Failed", Colors.FAIL)
else:
self.error("Exception: %s", e)
return False
def _list_uris(self):
@ -599,11 +536,6 @@ user argument, you can thus overrides command line options using that.
return True
return False
def get_blacklisted(self, options):
self._populate_testsuite(options)
return GST_VALIDATE_BLACKLISTED_TESTS
def set_settings(self, options, args, reporter):
if options.wanted_tests:
for i in range(len(options.wanted_tests)):
@ -615,7 +547,7 @@ user argument, you can thus overrides command line options using that.
except ValueError:
pass
TestsManager.set_settings(self, options, args, reporter)
super(GstValidateTestManager, self).set_settings(options, args, reporter)
def gst_validate_checkout_element_present(element_name):
null = open(os.devnull)

View file

@ -1,6 +1,6 @@
#!/usr/bin/env python2
#
# gst-validate-default-pipelines.py
# validate_default_testsuite.py
#
# Copyright (c) 2014, Thibault Saunier tsaunier@gnome.org
#
@ -20,19 +20,20 @@
# Boston, MA 02110-1301, USA.
def gst_validate_register_default_test_generators():
def register_default_test_generators(self):
"""
Registers default test generators
"""
GST_VALIDATE_TEST_GENERATORS.append(GstValidatePlaybinTestGenerator())
GST_VALIDATE_TEST_GENERATORS.append(GstValidateMediaCheckTestGenerator())
GST_VALIDATE_TEST_GENERATORS.append(GstValidateTranscodingTestGenerator())
self.add_generators([GstValidatePlaybinTestsGenerator(self),
GstValidateMediaCheckTestsGenerator(self),
GstValidateTranscodingTestsGenerator(self)])
def gst_validate_register_default_scenarios():
def register_default_scenarios(self):
"""
Registers default test scenarios
"""
GST_VALIDATE_SCENARIOS.extend([
self.add_scenarios([
"play_15s",
"reverse_playback",
"fast_forward",
@ -47,19 +48,19 @@ def gst_validate_register_default_scenarios():
"change_state_intensive",
"scrub_forward_seeking"])
def gst_validate_register_default_encoding_formats():
def register_default_encoding_formats(self):
"""
Registers default encoding formats
"""
GST_VALIDATE_ENCODING_FORMATS.extend([
self.add_encoding_formats([
MediaFormatCombination("ogg", "vorbis", "theora"),
MediaFormatCombination("webm", "vorbis", "vp8"),
MediaFormatCombination("mp4", "mp3", "h264"),
MediaFormatCombination("mkv", "vorbis", "h264"),
])
def gst_validate_define_default_blacklist():
GST_VALIDATE_BLACKLISTED_TESTS.extend([
def register_default_blacklist(self):
self.set_default_blacklist([
("validate.hls.playback.fast_forward.*",
"https://bugzilla.gnome.org/show_bug.cgi?id=698155"),
("validate.hls.playback.seek_with_stop.*",
@ -88,8 +89,18 @@ def gst_validate_define_default_blacklist():
("validate.http.*scrub_forward_seeking.*", "This is not stable enough for now."),
])
def gst_validate_register_defaults():
gst_validate_register_default_test_generators()
gst_validate_register_default_scenarios()
gst_validate_register_default_encoding_formats()
gst_validate_define_default_blacklist()
def register_defaults(self):
self.register_default_scenarios()
self.register_default_encoding_formats()
self.register_default_blacklist()
self.register_default_test_generators()
try:
GstValidateTestManager.register_defaults = register_defaults
GstValidateTestManager.register_default_blacklist = register_default_blacklist
GstValidateTestManager.register_default_test_generators = register_default_test_generators
GstValidateTestManager.register_default_scenarios = register_default_scenarios
GstValidateTestManager.register_compositing_tests = register_compositing_tests
GstValidateTestManager.register_default_encoding_formats = register_default_encoding_formats
except NameError:
pass

View file

@ -457,6 +457,7 @@ class TestsManager(Loggable):
self.reporter = None
self.wanted_tests_patterns = []
self.blacklisted_tests_patterns = []
self._generators = []
def init(self):
return False
@ -473,8 +474,29 @@ class TestsManager(Loggable):
def get_tests(self):
return self.tests
def get_blacklisted(self, options=None):
return []
def populate_testsuite(self):
pass
def add_generators(self, generators):
"""
@generators: A list of, or one single #TestsGenerator to be used to generate tests
"""
if isinstance(generators, list):
self._generators.extend(generators)
else:
self._generators.append(generators)
def get_generators(self):
return self._generators
def set_default_blacklist(self, default_blacklist):
msg = "\nCurrently 'hardcoded' %s blacklisted tests:\n\n" % self.name
for name, bug in default_blacklist:
self.options.blacklisted_tests.append(name)
msg += " + %s \n --> bug: %s\n" % (name, bug)
printc(msg, Colors.FAIL, True)
def add_options(self, parser):
""" Add more arguments. """
@ -486,6 +508,7 @@ class TestsManager(Loggable):
self.args = args
self.reporter = reporter
self.populate_testsuite()
if options.wanted_tests:
for patterns in options.wanted_tests:
for pattern in patterns.split(","):
@ -545,6 +568,34 @@ class TestsManager(Loggable):
return False
class TestsGenerator(Loggable):
def __init__(self, name, test_manager, tests=[]):
Loggable.__init__(self)
self.name = name
self.test_manager = test_manager
self._tests = {}
for test in tests:
self._tests[test.classname] = test
def generate_tests(self, *kwargs):
"""
Method that generates tests
"""
return list(self._tests.values())
def add_test(self, test):
self._tests[test.classname] = test
class GstValidateTestsGenerator(TestsGenerator):
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
pass
def generate_tests(self, uri_minfo_special_scenarios, scenarios):
self.populate_tests(uri_minfo_special_scenarios, scenarios)
return super(GstValidateTestsGenerator, self).generate_tests()
class _TestsLauncher(Loggable):
def __init__(self):
@ -593,9 +644,17 @@ class _TestsLauncher(Loggable):
self.testers.append(tester)
args.remove(tester.name)
if options.config:
for tester in self.testers:
tester.options = options
globals()[tester.name] = tester
globals()["options"] = options
execfile(self.options.config, globals())
for tester in self.testers:
tester.set_settings(options, args, self.reporter)
def list_tests(self):
for tester in self.testers:
tester.list_tests()
@ -638,16 +697,6 @@ class _TestsLauncher(Loggable):
if tester.needs_http_server():
return True
def get_blacklisted(self, options=None):
res = []
for tester in self.testers:
for blacklisted in tester.get_blacklisted(options):
if isinstance(blacklisted, str):
res.append(blacklisted, "Unknown")
else:
res.append(blacklisted)
return res
class NamedDic(object):
@ -779,3 +828,44 @@ class ScenarioManager(Loggable):
except IndexError:
self.warning("Scenario: %s not found" % name)
return None
class GstValidateBaseTestManager(TestsManager):
scenarios_manager = ScenarioManager()
def __init__(self):
super(GstValidateBaseTestManager, self).__init__()
self._scenarios = []
self._encoding_formats = []
def add_scenarios(self, scenarios):
"""
@scenarios: A list or a unic scenario name(s) to be run on the tests.
They are just the default scenarios, and then depending on
the TestsGenerator to be used you can have more fine grained
control on what to be run on each serie of tests.
"""
if isinstance(scenarios, list):
self._scenarios.extend(scenarios)
else:
self._scenarios.append(scenarios)
def get_scenarios(self):
return self._scenarios
def add_encoding_formats(self, encoding_formats):
"""
@encoding_formats: A list or one single #MediaFormatCombinations describing wanted output
formats for transcoding test.
They are just the default encoding formats, and then depending on
the TestsGenerator to be used you can have more fine grained
control on what to be run on each serie of tests.
"""
if isinstance(encoding_formats, list):
self._encoding_formats.extend(encoding_formats)
else:
self._encoding_formats.append(encoding_formats)
def get_encoding_formats(self):
return self._encoding_formats

View file

@ -181,7 +181,29 @@ def main():
help="Set it in order to generate the missing .media_infos files")
parser.add_argument("-lt", "--long-test-limit", dest="long_limit",
default=utils.LONG_TEST, action='store',
help="Defines the limite from which a test is concidered as long (is seconds)"),
help="Defines the limite from which a test is concidered as long (in seconds)"),
parser.add_argument("-c", "--config", dest="config",
default=None,
help="""Lets you specify a file where the testsuite to execute is defined.
In this file you will have acces to the TestManager objects that you can configure with
its various methods, for example you can find the 'validate' variable in case the GstValidateManager
launcher is avalaible. You should configure it using:
* validate.add_scenarios: which allows you to register a list of scenario names to be run
* validate.set_default_blacklist: Lets you set a list of tuple of the form:
(@regex_defining_blacklister_test_names, @reason_for_the_blacklisting)
* validate.add_generators: which allows you to register a list of #GstValidateTestsGenerator
to be used to generate tests
* validate.add_encoding_formats:: which allows you to register a list #MediaFormatCombination to be used for transcoding tests
You can also set default values with:
* validate.register_defaults: Sets default values for all parametters
* validate.register_default_test_generators: Sets default values for the TestsGenerators to be used
* gst_validate_register_default_scenarios: Sets default values for the scenarios to be executed
* gst_validate_register_default_encoding_formats: Sets default values for the encoding formats to be tested
Note: In the config file, you have acces to the options variable resulting from the parsing of the command line
user argument, you can thus overrides command line options using that.
""")
dir_group = parser.add_argument_group("Directories and files to be used by the launcher")
parser.add_argument('--xunit-file', action='store',
dest='xunit_file', metavar="FILE",
@ -279,15 +301,6 @@ def main():
% options.clone_dir, Colors.FAIL, True)
return -1
blacklisted = tests_launcher.get_blacklisted(options)
if blacklisted:
msg = "Currently 'hardcoded' blacklisted tests:\n"
for name, bug in blacklisted:
options.blacklisted_tests.append(name)
msg += " + %s \n --> bug: %s\n\n" % (name, bug)
printc(msg, Colors.FAIL, True)
tests_launcher.set_settings(options, args)
if options.remote_assets_url and options.sync: