gstreamer/validate/launcher/utils.py
2017-01-03 16:24:19 -03:00

448 lines
13 KiB
Python

#!/usr/bin/env python3
#
# 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.
""" Some utilies. """
try:
import config
except ImportError:
from . import config
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.request
import urllib.error
import urllib.parse
from .loggable import Loggable
from operator import itemgetter
from xml.etree import ElementTree
GST_SECOND = int(1000000000)
DEFAULT_TIMEOUT = 30
DEFAULT_MAIN_DIR = os.path.join(os.path.expanduser("~"), "gst-validate")
DEFAULT_GST_QA_ASSETS = os.path.join(DEFAULT_MAIN_DIR, "gst-integration-testsuites")
DISCOVERER_COMMAND = "gst-discoverer-1.0"
# Use to set the duration from which a test is considered as being 'long'
LONG_TEST = 40
class Result(object):
NOT_RUN = "Not run"
FAILED = "Failed"
TIMEOUT = "Timeout"
PASSED = "Passed"
KNOWN_ERROR = "Known error"
class Protocols(object):
HTTP = "http"
FILE = "file"
HLS = "hls"
DASH = "dash"
@staticmethod
def needs_clock_sync(protocol):
if protocol in [Protocols.HLS, Protocols.DASH]:
return True
return False
class Colors(object):
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
def desactivate_colors():
Colors.HEADER = ''
Colors.OKBLUE = ''
Colors.OKGREEN = ''
Colors.WARNING = ''
Colors.FAIL = ''
Colors.ENDC = ''
def mkdir(directory):
try:
os.makedirs(directory)
except os.error:
pass
def which(name, extra_path=None):
exts = [_f for _f in os.environ.get('PATHEXT', '').split(os.pathsep) if _f]
path = os.environ.get('PATH', '')
if extra_path:
path = extra_path + os.pathsep + path
if not path:
return []
for p in path.split(os.pathsep):
p = os.path.join(p, name)
if os.access(p, os.X_OK):
return p
for e in exts:
pext = p + e
if os.access(pext, os.X_OK):
return pext
return None
def get_color_for_result(result):
if result is Result.FAILED:
color = Colors.FAIL
elif result is Result.TIMEOUT:
color = Colors.WARNING
elif result is Result.PASSED:
color = Colors.OKGREEN
else:
color = Colors.OKBLUE
return color
def printc(message, color="", title=False, title_char=''):
if title or title_char:
length = 0
for l in message.split("\n"):
if len(l) > length:
length = len(l)
if length == 0:
length = len(message)
if title is True:
message = length * "=" + "\n" + str(message) + "\n" + length * '='
else:
message = str(message) + "\n" + length * title_char
if hasattr(message, "result") and color == '':
color = get_color_for_result(message.result)
sys.stdout.write(color + str(message) + Colors.ENDC + "\n")
sys.stdout.flush()
def launch_command(command, color=None, fails=False):
printc(command, Colors.OKGREEN, True)
res = os.system(command)
if res != 0 and fails is True:
raise subprocess.CalledProcessError(res, "%s failed" % command)
def path2url(path):
return urllib.parse.urljoin('file:', urllib.request.pathname2url(path))
def is_windows():
platname = platform.system().lower()
return platname == 'windows' or 'mingw' in platname
def url2path(url):
path = urllib.parse.urlparse(url).path
if "win32" in sys.platform:
if path[0] == '/':
return path[1:] # We need to remove the first '/' on windows
path = urllib.parse.unquote(path)
return path
def isuri(string):
url = urllib.parse.urlparse(string)
if url.scheme != "" and url.scheme != "":
return True
return False
def touch(fname, times=None):
with open(fname, 'a'):
os.utime(fname, times)
def get_subclasses(klass, env):
subclasses = []
for symb in env.items():
try:
if issubclass(symb[1], klass) and not symb[1] is klass:
subclasses.append(symb[1])
except TypeError:
pass
return subclasses
def TIME_ARGS(time):
return "%u:%02u:%02u.%09u" % (time / (GST_SECOND * 60 * 60),
(time / (GST_SECOND * 60)) % 60,
(time / GST_SECOND) % 60,
time % GST_SECOND)
def look_for_file_in_source_dir(subdir, name):
root_dir = os.path.abspath(os.path.dirname(os.path.join(os.path.dirname(os.path.abspath(__file__)))))
p = os.path.join(root_dir, subdir, name)
if os.path.exists(p):
return p
return None
# Returns the path $top_src_dir/@subdir/@name if running from source, or
# $DATADIR/gstreamer-1.0/validate/@name if not
def get_data_file(subdir, name):
# Are we running from sources?
p = look_for_file_in_source_dir(subdir, name)
if p:
return p
# Look in system data dirs
p = os.path.join(config.DATADIR, 'gstreamer-1.0', 'validate', name)
if os.path.exists(p):
return p
return None
#
# Some utilities to parse gst-validate output #
#
def gsttime_from_tuple(stime):
return int((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
timeregex = re.compile(r'(?P<_0>.+):(?P<_1>.+):(?P<_2>.+)\.(?P<_3>.+)')
def parse_gsttimeargs(time):
stime = list(map(itemgetter(1), sorted(
timeregex.match(time).groupdict().items())))
return int((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
def get_duration(media_file):
duration = 0
res = ''
try:
res = subprocess.check_output([DISCOVERER_COMMAND, media_file]).decode()
except subprocess.CalledProcessError:
# gst-media-check returns !0 if seeking is not possible, we do not care
# in that case.
pass
for l in res.split('\n'):
if "Duration: " in l:
duration = parse_gsttimeargs(l.replace("Duration: ", ""))
break
return duration
def get_scenarios():
GST_VALIDATE_COMMAND = "gst-validate-1.0"
os.system("%s --scenarios-defs-output-file %s" % (GST_VALIDATE_COMMAND,
))
class BackTraceGenerator(Loggable):
__instance = None
_executable_regex = re.compile(r'Executable: (.*)\n')
_timestamp_regex = re.compile(r'Timestamp: .*\((\d*)s ago\)')
def __init__(self):
Loggable.__init__(self)
self.coredumpctl = shutil.which('coredumpctl')
self.gdb = shutil.which('gdb')
@classmethod
def get_default(cls):
if not cls.__instance:
cls.__instance = BackTraceGenerator()
return cls.__instance
def get_trace(self, test):
if not test.process.returncode:
return self.get_trace_on_running_process(test)
if self.coredumpctl:
return self.get_trace_from_systemd(test)
self.debug("coredumpctl not present, and it is the only"
" supported way to get backtraces for now.")
return None
def get_trace_on_running_process(self, test):
if not self.gdb:
return "Can not generate stack trace as `gdb` is not" \
"installed."
gdb = ['gdb', '-ex', 't a a bt', '-batch',
'-p', str(test.process.pid)]
try:
return subprocess.check_output(
gdb, stderr=subprocess.STDOUT, timeout=30).decode()
except Exception as e:
return "Could not run `gdb` on process (pid: %d):\n%s" % (
test.process.pid, e)
def get_trace_from_systemd(self, test):
for ntry in range(10):
if ntry != 0:
# Loopping, it means we conceder the logs might not be ready yet.
time.sleep(1)
try:
info = subprocess.check_output(['coredumpctl', 'info',
str(test.process.pid)],
stderr=subprocess.STDOUT)
except subprocess.CalledProcessError:
# The trace might not be ready yet
time.sleep(1)
continue
info = info.decode()
try:
executable = BackTraceGenerator._executable_regex.findall(info)[0]
except IndexError:
self.debug("Backtrace could not be found yet, trying harder.")
# The trace might not be ready yet
continue
if executable != test.application:
self.debug("PID: %s -- executable %s != test application: %s" % (
test.process.pid, executable, test.application))
# The trace might not be ready yet
continue
if not BackTraceGenerator._timestamp_regex.findall(info):
self.debug("Timestamp %s is more than 1min old",
re.findall(r'Timestamp: .*', info))
# The trace might not be ready yet
continue
bt_all = None
if self.gdb:
try:
tf = tempfile.NamedTemporaryFile()
subprocess.check_output(['coredumpctl', 'dump',
str(test.process.pid), '--output=' +
tf.name], stderr=subprocess.STDOUT)
gdb = ['gdb', '-ex', 't a a bt', '-ex', 'quit',
test.application, tf.name]
bt_all = subprocess.check_output(
gdb, stderr=subprocess.STDOUT).decode()
info += "\nThread apply all bt:\n\n%s" % (
bt_all.replace('\n', '\n' + 15 * ' '))
except Exception as e:
self.debug("Could not get backtrace from gdb: %s" % e)
return info
return None
def check_bugs_resolution(bugs_definitions):
bugz = {}
regexes = {}
for regex, bugs in bugs_definitions:
if isinstance(bugs, str):
bugs = [bugs]
for bug in bugs:
url = urllib.parse.urlparse(bug)
if "bugzilla" not in url.netloc:
printc(" + %s \n --> bug: %s\n --> Status: Not a bugzilla report\n" % (regex, bug),
Colors.WARNING)
continue
query = urllib.parse.parse_qs(url.query)
_id = query.get('id')
if not _id:
printc(" + '%s' -- Can't check bug '%s'\n" % (regex, bug), Colors.WARNING)
continue
if isinstance(_id, list):
_id = _id[0]
regexes[_id] = (regex, bug)
url_parts = tuple(list(url)[:3] + ['', '', ''])
ids = bugz.get(url_parts, [])
ids.append(_id)
bugz[url_parts] = ids
res = True
for url_parts, ids in bugz.items():
url_parts = list(url_parts)
query = {'id': ','.join(ids)}
query['ctype'] = 'xml'
url_parts[4] = urllib.parse.urlencode(query)
try:
res = urllib.request.urlopen(urllib.parse.urlunparse(url_parts))
except Exception as e:
printc(" + Could not properly check bugs status for: %s (%s)\n"
% (urllib.parse.urlunparse(url_parts), e), Colors.FAIL)
continue
root = ElementTree.fromstring(res.read())
bugs = root.findall('./bug')
if len(bugs) != len(ids):
printc(" + Could not properly check bugs status on server %s\n" %
urllib.parse.urlunparse(url_parts), Colors.FAIL)
continue
for bugelem in bugs:
status = bugelem.findtext('./bug_status')
bugid = bugelem.findtext('./bug_id')
regex, bug = regexes[bugid]
desc = bugelem.findtext('./short_desc')
if not status:
printc(" + %s \n --> bug: %s\n --> Status: UNKNOWN\n" % (regex, bug),
Colors.WARNING)
continue
if not status.lower() in ['new', 'verified']:
printc(" + %s \n --> bug: #%s: '%s'\n ==> Bug CLOSED already (status: %s)\n" % (
regex, bugid, desc, status), Colors.WARNING)
res = False
printc(" + %s \n --> bug: #%s: '%s'\n --> Status: %s\n" % (
regex, bugid, desc, status), Colors.OKGREEN)
return res