validate/launcher/utils.py

   1 #!/usr/bin/env python2
   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 """ Some utilies. """
  20
  21 import config
  22 import os
  23 import re
  24 import sys
  25 import urllib
  26 import urlparse
  27 import subprocess
  28
  29 from operator import itemgetter
  30
  31
  32 GST_SECOND = long(1000000000)
  33 DEFAULT_TIMEOUT = 30
  34 DEFAULT_MAIN_DIR = os.path.join(os.path.expanduser("~"), "gst-validate")
  35 DEFAULT_GST_QA_ASSETS = os.path.join(DEFAULT_MAIN_DIR, "gst-integration-testsuites")
  36 DISCOVERER_COMMAND = "gst-discoverer-1.0"
  37 # Use to set the duration from which a test is concidered as being 'long'
  38 LONG_TEST = 40
  39
  40
  41 class Result(object):
  42     NOT_RUN = "Not run"
  43     FAILED = "Failed"
  44     TIMEOUT = "Timeout"
  45     PASSED = "Passed"
  46     KNOWN_ERROR = "Known error"
  47
  48
  49 class Protocols(object):
  50     HTTP = "http"
  51     FILE = "file"
  52     HLS = "hls"
  53     DASH = "dash"
  54
  55     @staticmethod
  56     def needs_clock_sync(protocol):
  57         if protocol == Protocols.HLS:
  58             return True
  59
  60         return False
  61
  62
  63 class Colors(object):
  64     HEADER = '\033[95m'
  65     OKBLUE = '\033[94m'
  66     OKGREEN = '\033[92m'
  67     WARNING = '\033[93m'
  68     FAIL = '\033[91m'
  69     ENDC = '\033[0m'
  70
  71
  72 def desactivate_colors():
  73     Colors.HEADER = ''
  74     Colors.OKBLUE = ''
  75     Colors.OKGREEN = ''
  76     Colors.WARNING = ''
  77     Colors.FAIL = ''
  78     Colors.ENDC = ''
  79
  80
  81 def mkdir(directory):
  82     try:
  83         os.makedirs(directory)
  84     except os.error:
  85         pass
  86
  87
  88 def which(name):
  89     result = []
  90     exts = filter(None, os.environ.get('PATHEXT', '').split(os.pathsep))
  91     path = os.environ.get('PATH', None)
  92     if path is None:
  93         return []
  94     for p in os.environ.get('PATH', '').split(os.pathsep):
  95         p = os.path.join(p, name)
  96         if os.access(p, os.X_OK):
  97             result.append(p)
  98         for e in exts:
  99             pext = p + e
 100             if os.access(pext, os.X_OK):
 101                 result.append(pext)
 102     return result
 103
 104
 105 def get_color_for_result(result):
 106     if result is Result.FAILED:
 107         color = Colors.FAIL
 108     elif result is Result.TIMEOUT:
 109         color = Colors.WARNING
 110     elif result is Result.PASSED:
 111         color = Colors.OKGREEN
 112     else:
 113         color = Colors.OKBLUE
 114
 115     return color
 116
 117
 118 def printc(message, color="", title=False):
 119     if title:
 120         length = 0
 121         for l in message.split("\n"):
 122             if len(l) > length:
 123                 length = len(l)
 124         if length == 0:
 125             length = len(message)
 126         message = length * '=' + "\n" + str(message) + "\n" + length * '='
 127
 128     if hasattr(message, "result") and color == '':
 129         color = get_color_for_result(message.result)
 130
 131     sys.stdout.write(color + str(message) + Colors.ENDC + "\n")
 132     sys.stdout.flush()
 133
 134
 135 def launch_command(command, color=None, fails=False):
 136     printc(command, Colors.OKGREEN, True)
 137     res = os.system(command)
 138     if res != 0 and fails is True:
 139         raise subprocess.CalledProcessError(res, "%s failed" % command)
 140
 141
 142 def path2url(path):
 143     return urlparse.urljoin('file:', urllib.pathname2url(path))
 144
 145
 146 def url2path(url):
 147     path = urlparse.urlparse(url).path
 148     if "win32" in sys.platform:
 149         if path[0] == '/':
 150             return path[1:]  # We need to remove the first '/' on windows
 151     return path
 152
 153
 154 def isuri(string):
 155     url = urlparse.urlparse(string)
 156     if url.scheme != "" and url.scheme != "":
 157         return True
 158
 159     return False
 160
 161
 162 def touch(fname, times=None):
 163     with open(fname, 'a'):
 164         os.utime(fname, times)
 165
 166
 167 def get_subclasses(klass, env):
 168     subclasses = []
 169     for symb in env.iteritems():
 170         try:
 171             if issubclass(symb[1], klass) and not symb[1] is klass:
 172                 subclasses.append(symb[1])
 173         except TypeError:
 174             pass
 175
 176     return subclasses
 177
 178
 179 def TIME_ARGS(time):
 180     return "%u:%02u:%02u.%09u" % (time / (GST_SECOND * 60 * 60),
 181                                   (time / (GST_SECOND * 60)) % 60,
 182                                   (time / GST_SECOND) % 60,
 183                                   time % GST_SECOND)
 184
 185
 186 def look_for_file_in_source_dir(subdir, name):
 187     root_dir = os.path.abspath(os.path.dirname(os.path.join(os.path.dirname(os.path.abspath(__file__)))))
 188     p = os.path.join(root_dir, subdir, name)
 189     if os.path.exists(p):
 190         return p
 191
 192     return None
 193
 194
 195 # Returns the path $top_src_dir/@subdir/@name if running from source, or
 196 # $DATADIR/gstreamer-1.0/validate/@name if not
 197 def get_data_file(subdir, name):
 198     # Are we running from sources?
 199     p = look_for_file_in_source_dir(subdir, name)
 200     if p:
 201         return p
 202
 203     # Look in system data dirs
 204     p = os.path.join(config.DATADIR, 'gstreamer-1.0', 'validate', name)
 205     if os.path.exists(p):
 206         return p
 207
 208     return None
 209
 210 #
 211 # Some utilities to parse gst-validate output   #
 212 #
 213
 214
 215 def gsttime_from_tuple(stime):
 216     return long((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
 217
 218 timeregex = re.compile(r'(?P<_0>.+):(?P<_1>.+):(?P<_2>.+)\.(?P<_3>.+)')
 219
 220
 221 def parse_gsttimeargs(time):
 222     stime = map(itemgetter(1), sorted(
 223         timeregex.match(time).groupdict().items()))
 224     return long((int(stime[0]) * 3600 + int(stime[1]) * 60 + int(stime[2])) * GST_SECOND + int(stime[3]))
 225
 226
 227 def get_duration(media_file):
 228
 229     duration = 0
 230     res = ''
 231     try:
 232         res = subprocess.check_output([DISCOVERER_COMMAND, media_file])
 233     except subprocess.CalledProcessError:
 234         # gst-media-check returns !0 if seeking is not possible, we do not care
 235         # in that case.
 236         pass
 237
 238     for l in res.split('\n'):
 239         if "Duration: " in l:
 240             duration = parse_gsttimeargs(l.replace("Duration: ", ""))
 241             break
 242
 243     return duration
 244
 245
 246 def get_scenarios():
 247     GST_VALIDATE_COMMAND = "gst-validate-1.0"
 248     os.system("%s --scenarios-defs-output-file %s" % (GST_VALIDATE_COMMAND,
 249                                                       ))