mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-06-08 00:59:26 +00:00
Code Issues 1 Projects Releases Wiki Activity

validate: tools: Cleanup test launcher tool

Browse source 
Previous commit was not meant to be pushed and those two should have
been fixed up together, sorry for the mistake
This commit is contained in:
Thibault Saunier 2013-12-31 11:45:07 +01:00
parent 8c39dbfaa8
commit e9b2967504
5 changed files with 166 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
validate/tools/apps/ges-projects-tests.py
View file

@ -21,8 +21,8 @@ import os
import time
from urllib import unquote
from urlparse import urlsplit
from utils import launch_command
from gi.repository import GES, Gst, GLib
from testdefinitions import Test, DEFAULT_QA_SAMPLE_PATH, TestsManager
DURATION_TOLERANCE = Gst.SECOND / 2
DEFAULT_GES_LAUNCH = "ges-launch-1.0"
@ -110,8 +110,8 @@ def quote_uri(uri):
class GESTest(Test):
def __init__(self, classname, options, reporter, project_uri, scenario,
combination=None):
super(GESTest, self).__init__(DEFAULT_GES_LAUNCH, classname, options, reporter)
self.scenario = scenario
super(GESTest, self).__init__(DEFAULT_GES_LAUNCH, classname, options, reporter,
scenario)
self.project_uri = project_uri
self.combination = combination
proj = GES.Project.new(project_uri)
@ -160,9 +160,7 @@ class GESTest(Test):
self.add_arguments("--sample-paths", "file://" + path)
def build_arguments(self):
print "\OOO %s" % self.combination
if self.scenario is not None:
self.add_arguments("--set-scenario", self.scenario)
Test.build_arguments(self)
if self.combination is not None:
self.set_rendering_info()
@ -205,19 +203,7 @@ class GESTest(Test):
self.set_result(Result.TIMEOUT, "The rendered file add right duration, MISSING EOS?\n",
"failure", e)
else:
if self.result == Result.TIMEOUT:
self.set_result(Result.TIMEOUT, "Application timed out", "timeout")
else:
if self.process.returncode == 139:
self.get_backtrace("SEGFAULT")
self.set_result("Application segfaulted")
else:
self.set_result(Result.FAILED,
"Application returned %d (issues: %s)" % (
self.process.returncode,
self.get_validate_criticals_errors()),
"error")
Test.check_results(self)
def wait_process(self):
last_val = 0
@ -263,41 +249,29 @@ class GESTest(Test):
class GESTestsManager(TestsManager):
name = "ges"
def __init__(self):
super(GESTestsManager, self).__init__()
Gst.init(None)
GES.init()
default_opath = GLib.get_user_special_dir(
GLib.UserDirectory.DIRECTORY_VIDEOS)
if default_opath:
self.default_path = os.path.join(default_opath, "ges-projects")
else:
self.default_path = os.path.join(os.path.expanduser('~'), "Video",
"ges-projects")
def add_options(self, parser):
parser.add_option("-o", "--output-path", dest="dest",
default=os.path.join(self.default_path, "rendered"),
help="Set the path to which projects should be"
" renderd")
parser.add_option("-P", "--sample-path", dest="paths",
default=[],
help="Paths in which to look for moved assets")
parser.add_option("-r", "--recurse-paths", dest="recurse_paths",
default=False, action="store_true",
help="Whether to recurse into paths to find assets")
parser.add_option("-m", "--mute", dest="mute",
action="store_true", default=False,
help="Mute playback output, which mean that we use "
"a fakesink")
def add_options(self, group):
group.add_option("-o", "--output-path", dest="dest",
default=None,
help="Set the path to which projects should be"
" renderd")
group.add_option("-P", "--projects-paths", dest="projects_paths",
default=os.path.join(DEFAULT_QA_SAMPLE_PATH, "ges-projects"),
help="Paths in which to look for moved medias")
group.add_option("-r", "--recurse-paths", dest="recurse_paths",
default=False, action="store_true",
help="Whether to recurse into paths to find medias")
def set_settings(self, options, args, reporter):
TestsManager.set_settings(self, options, args, reporter)
if not args and not os.path.exists(self.default_path):
launch_command("git clone %s" % DEFAULT_ASSET_REPO,
"Getting assets")
if options.dest is None:
options.dest = os.path.join(options.logsdir, "rendered")
if not Gst.uri_is_valid(options.dest):
options.dest = GLib.filename_to_uri(options.dest, None)
@ -311,8 +285,8 @@ class GESTestsManager(TestsManager):
def list_tests(self):
projects = list()
if not self.args:
self.options.paths = [os.path.join(self.default_path, "assets")]
path = os.path.join(self.default_path, "projects")
path = self.options.projects_paths
print path
for root, dirs, files in os.walk(path):
for f in files:
if not f.endswith(".xges"):

60
validate/tools/gst-validate-launcher.py
View file

@ -1,20 +1,42 @@
#!/usr//bin/python
#
# Copyright (c) 2013,Thibault Saunier <thibault.saunier@collabora.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301, USA.
import os
from testdefinitions import _TestsLauncher
import logging
from testdefinitions import _TestsLauncher, DEFAULT_QA_SAMPLE_PATH
from utils import printc
from optparse import OptionParser
def main():
parser = OptionParser()
parser.add_option("-g", "--gdb", dest="gdb",
action="store_true",
default=False,
help="Run applications into gdb")
parser.add_option("-f", "--forever", dest="forever",
action="store_true", default=False,
help="Keep running tests until one fails")
parser.add_option("-F", "--fatal-error", dest="fatal_error",
action="store_true", default=False,
help="Stop on first fail")
# FIXME:
#parser.add_option("-g", "--gdb", dest="gdb",
#action="store_true",
#default=False,
#help="Run applications into gdb")
#parser.add_option("-f", "--forever", dest="forever",
#action="store_true", default=False,
#help="Keep running tests until one fails")
#parser.add_option("-F", "--fatal-error", dest="fatal_error",
#action="store_true", default=False,
#help="Stop on first fail")
parser.add_option('--xunit-file', action='store',
dest='xunit_file', metavar="FILE",
default=None,
@ -31,6 +53,20 @@ def main():
parser.add_option("-l", "--logs-dir", dest="logsdir",
action="store_true", default=os.path.expanduser("~/gst-validate/logs/"),
help="Directory where to store logs")
parser.add_option("-p", "--medias-paths", dest="paths",
default=[os.path.join(DEFAULT_QA_SAMPLE_PATH, "medias")],
help="Paths in which to look for media files")
parser.add_option("-m", "--mute", dest="mute",
action="store_true", default=False,
help="Mute playback output, which mean that we use "
"a fakesink")
try:
level = getattr(logging,
os.environ["GST_VALIDATE_LAUNCHER_DEBUG"].upper(),
None)
logging.basicConfig(level=level)
except:
pass
tests_launcher = _TestsLauncher()
tests_launcher.add_options(parser)
@ -41,7 +77,7 @@ def main():
tests_launcher.list_tests()
if options.list_tests:
for test in tests_launcher.tests:
print test
printc(test)
return 0
tests_launcher.run_tests()
tests_launcher.final_report()

18
validate/tools/reporters.py
View file

@ -22,9 +22,9 @@
import os
import re
import codecs
import testdefinitions
import logging
from xml.sax import saxutils
from utils import mkdir, Result
from utils import mkdir, Result, printc
UNICODE_STRINGS = (type(unicode()) == type(str()))
@ -75,6 +75,7 @@ class Reporter(object):
self.stats["passed"] += 1
def add_results(self, test):
logging.debug("%s", test)
if test.result == Result.PASSED:
self.set_passed(test)
elif test.result == Result.FAILED or \
@ -84,14 +85,15 @@ class Reporter(object):
raise UnknownResult("%s" % test.result)
def after_test(self):
self.out.close()
self.out = None
self.results.append(self._current_test)
self.add_results(self._current_test)
self.out.close()
self.out = None
self._current_test = None
def final_report(self):
pass
for test in self.results:
printc(test)
class XunitReporter(Reporter):
@ -106,6 +108,7 @@ class XunitReporter(Reporter):
def final_report(self):
self.report()
super(XunitReporter, self).final_report()
def _get_captured(self):
if self.out:
@ -129,13 +132,12 @@ class XunitReporter(Reporter):
The file includes a report of test errors and failures.
"""
print "Writing XML file to: %s" % self.options.xunit_file
logging.debug("Writing XML file to: %s", self.options.xunit_file)
self.xml_file = codecs.open(self.options.xunit_file, 'w',
self.encoding, 'replace')
self.stats['encoding'] = self.encoding
self.stats['total'] = (self.stats['timeout'] + self.stats['failures']
+ self.stats['passes'] + self.stats['skipped'])
print self.stats
self.xml_file.write( u'<?xml version="1.0" encoding="%(encoding)s"?>'
u'<testsuite name="gesprojectslauncher" tests="%(total)d" '
u'errors="%(timeout)d" failures="%(failures)d" '
@ -149,7 +151,6 @@ class XunitReporter(Reporter):
"""Add failure output to Xunit report.
"""
self.stats['failures'] += 1
self.results.insert(0, test)
self.errorlist.append(
'<testcase classname=%(cls)s name=%(name)s time="%(taken).3f">'
'<failure type=%(errtype)s message=%(message)s>'
@ -166,7 +167,6 @@ class XunitReporter(Reporter):
"""Add success output to Xunit report.
"""
self.stats['passes'] += 1
self.results.append(test)
self.errorlist.append(
'<testcase classname=%(cls)s name=%(name)s '
'time="%(taken).3f">%(systemout)s</testcase>' %

91
validate/tools/testdefinitions.py
View file

@ -24,37 +24,46 @@ import re
import time
import subprocess
import reporters
import logging
from optparse import OptionGroup
from utils import mkdir, Result
from utils import mkdir, Result, Colors, printc
DEFAULT_TIMEOUT = 10
DEFAULT_QA_SAMPLE_PATH = os.path.join(os.path.expanduser('~'), "Videos",
"gst-qa-samples")
class Test(object):
""" A class representing a particular test. """
def __init__(self, application_name, classname, options, reporter):
self.timeout = DEFAULT_TIMEOUT
def __init__(self, application_name, classname, options, reporter, scenario=None, timeout=DEFAULT_TIMEOUT):
self.timeout = timeout
self.classname = classname
self.options = options
self.application = application_name
self.command = ""
self.reporter = reporter
self.process = None
if scenario.lower() == "none":
self.scenario = None
else:
self.scenario = scenario
self.message = None
self.error = None
self.time_taken = None
self.message = ""
self.error = ""
self.time_taken = 0.0
self._starting_time = None
self.result = Result.NOT_RUN
def __str__(self):
string = self.classname
if self.result:
if self.result != Result.NOT_RUN:
string += ": " + self.result
if "FAILED" in self.result:
string += "\n You can reproduce with: " + self.command
if self.result == Result.FAILED:
string += "'%s'\n You can reproduce with: %s" \
% (self.message, self.command)
return string
@ -63,19 +72,32 @@ class Test(object):
self.command += " " + arg
def build_arguments(self):
pass
if self.scenario is not None:
self.add_arguments("--set-scenario", self.scenario)
def set_result(self, result, message=None, error=None):
print "SETTING TER"
def set_result(self, result, message="", error=""):
self.result = result
self.message = message
self.error = error
def check_results(self):
if self.process.returncode == 0:
logging.debug("%s returncode: %d", self, self.process.returncode)
if self.result == Result.TIMEOUT:
self.set_result(Result.TIMEOUT, "Application timed out", "timeout")
elif self.process.returncode == 0:
self.result = Result.PASSED
self.result = Result.FAILED
else:
if self.process.returncode == 139:
self.get_backtrace("SEGFAULT")
self.set_result(Result.FAILED,
"Application segfaulted",
"segfault")
else:
self.set_result(Result.FAILED,
"Application returned %d (issues: %s)" % (
self.process.returncode,
self.get_validate_criticals_errors()),
"error")
def wait_process(self):
last_change_ts = time.time()
@ -96,11 +118,16 @@ class Test(object):
def get_validate_criticals_errors(self):
self.reporter.out.seek(0)
ret = "["
errors = []
for l in self.reporter.out.readlines():
if "critical : " in l:
if ret != "[":
ret += ", "
ret += l.split("critical : ")[1].replace("\n", '')
error = l.split("critical : ")[1].replace("\n", '')
print "%s -- %s" %(error, errors)
if error not in errors:
ret += error
errors.append(error)
if ret == "[":
return "No critical"
@ -111,7 +138,8 @@ class Test(object):
self.command = "%s " % (self.application)
self._starting_time = time.time()
self.build_arguments()
print "Launching %s" % self.command
printc("Launching:%s '%s' -- logs are in %s" % (Colors.ENDC, self.classname,
self.reporter.out.name), Colors.OKBLUE)
try:
self.process = subprocess.Popen(self.command,
stderr=self.reporter.out,
@ -134,6 +162,8 @@ class TestsManager(object):
""" A class responsible for managing tests. """
name = ""
def __init__(self):
self.tests = []
self.options = None
@ -163,6 +193,9 @@ class TestsManager(object):
def _is_test_wanted(self, test):
if not self.wanted_tests_patterns:
return True
for pattern in self.wanted_tests_patterns:
if pattern.findall(test.classname):
return True
@ -190,7 +223,7 @@ class _TestsLauncher(object):
subclasses = []
for symb in env.iteritems():
try:
if issubclass(symb[1], c):
if issubclass(symb[1], c) and not symb[1] is c:
subclasses.append(symb[1])
except TypeError:
pass
@ -201,16 +234,34 @@ class _TestsLauncher(object):
d = os.path.dirname(__file__)
for f in os.listdir(os.path.join(d, "apps")):
execfile(os.path.join(d, "apps", f), env)
self.testers = [i() for i in get_subclasses(TestsManager, env)]
print self.testers
def add_options(self, parser):
for tester in self.testers:
tester.add_options(parser)
group = OptionGroup(parser, "%s Options" % tester.name,
"Options specific to the %s test manager"
% tester.name)
tester.add_options(group)
parser.add_option_group(group)
def set_settings(self, options, args):
self.reporter = reporters.XunitReporter(options)
mkdir(options.logsdir)
wanted_testers = None
for tester in self.testers:
if tester.name in args:
wanted_testers = tester.name
if wanted_testers:
testers = self.testers
self.testers = []
for tester in testers:
if tester.name in args:
self.testers.append(tester)
args.remove(tester.name)
for tester in self.testers:
tester.set_settings(options, args, self.reporter)

23
validate/tools/utils.py
View file

@ -53,11 +53,20 @@ def mkdir(directory):
pass
def launch_command(command, name="", color=None):
if name != "":
if color is not None:
print "%s%s" % (color, len(name) * "=")
print name
if color is not None:
print "%s%s" % (len(name) * "=", Colors.ENDC)
def printc (message, color="", title=False):
if title:
message = len(message) * '=' + message + len(message) * '='
if hasattr(message, "result") and color == '':
if message.result == Result.FAILED:
color = Colors.FAIL
elif message.result == Result.PASSED:
color = Colors.OKGREEN
else:
color = Colors.OKBLUE
print color + str(message) + Colors.ENDC
def launch_command(command, color=None):
printc(command, Colors.OKGREEN, True)
os.system(command)