sdk/platform-tools/systrace/catapult/telemetry/telemetry/core/util.py

   1 # Copyright 2012 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4 import glob
   5 import imp
   6 import os
   7 import socket
   8 import sys
   9
  10 from telemetry import decorators
  11
  12 import py_utils as catapult_util  # pylint: disable=import-error
  13
  14
  15 IsRunningOnCrosDevice = catapult_util.IsRunningOnCrosDevice
  16 GetCatapultDir = catapult_util.GetCatapultDir
  17
  18
  19 def GetBaseDir():
  20   main_module = sys.modules['__main__']
  21   if hasattr(main_module, '__file__'):
  22     return os.path.dirname(os.path.abspath(main_module.__file__))
  23   else:
  24     return os.getcwd()
  25
  26
  27 def GetCatapultThirdPartyDir():
  28   return os.path.normpath(os.path.join(GetCatapultDir(), 'third_party'))
  29
  30
  31 def GetTelemetryDir():
  32   return os.path.normpath(os.path.join(
  33       os.path.abspath(__file__), '..', '..', '..'))
  34
  35
  36 def GetTelemetryThirdPartyDir():
  37   return os.path.join(GetTelemetryDir(), 'third_party')
  38
  39
  40 def GetUnittestDataDir():
  41   return os.path.join(GetTelemetryDir(), 'telemetry', 'internal', 'testing')
  42
  43
  44 def GetChromiumSrcDir():
  45   return os.path.normpath(os.path.join(GetTelemetryDir(), '..', '..', '..'))
  46
  47
  48 _counter = [0]
  49
  50
  51 def _GetUniqueModuleName():
  52   _counter[0] += 1
  53   return "page_set_module_" + str(_counter[0])
  54
  55
  56 def GetPythonPageSetModule(file_path):
  57   return imp.load_source(_GetUniqueModuleName(), file_path)
  58
  59
  60 @decorators.Deprecated(
  61     2017, 6, 1,
  62     'telemetry.core.utils.WaitFor() is being deprecated. Please use '
  63     'catapult.common.py_utils.WaitFor() instead.')
  64 def WaitFor(condition, timeout):
  65   return catapult_util.WaitFor(condition, timeout)
  66
  67
  68 class PortKeeper(object):
  69   """Port keeper hold an available port on the system.
  70
  71   Before actually use the port, you must call Release().
  72   """
  73
  74   def __init__(self):
  75     self._temp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  76     self._temp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  77     self._temp_socket.bind(('', 0))
  78     self._port = self._temp_socket.getsockname()[1]
  79
  80   @property
  81   def port(self):
  82     return self._port
  83
  84   def Release(self):
  85     assert self._temp_socket, 'Already released'
  86     self._temp_socket.close()
  87     self._temp_socket = None
  88
  89
  90 def GetUnreservedAvailableLocalPort():
  91   """Returns an available port on the system.
  92
  93   WARNING: This method does not reserve the port it returns, so it may be used
  94   by something else before you get to use it. This can lead to flake.
  95   """
  96   tmp = socket.socket()
  97   tmp.bind(('', 0))
  98   port = tmp.getsockname()[1]
  99   tmp.close()
 100
 101   return port
 102
 103
 104 def GetBuildDirectories(chrome_root=None):
 105   """Yields all combination of Chromium build output directories."""
 106   # chrome_root can be set to something else via --chrome-root.
 107   if not chrome_root:
 108     chrome_root = GetChromiumSrcDir()
 109
 110   # CHROMIUM_OUTPUT_DIR can be set by --chromium-output-directory.
 111   output_dir = os.environ.get('CHROMIUM_OUTPUT_DIR')
 112   if output_dir:
 113     yield os.path.join(chrome_root, output_dir)
 114   elif os.path.exists('build.ninja'):
 115     yield os.getcwd()
 116   else:
 117     out_dir = os.environ.get('CHROMIUM_OUT_DIR')
 118     if out_dir:
 119       build_dirs = [out_dir]
 120     else:
 121       build_dirs = ['build',
 122                     'out',
 123                     'xcodebuild']
 124
 125     build_types = ['Debug', 'Debug_x64', 'Release', 'Release_x64', 'Default']
 126
 127     for build_dir in build_dirs:
 128       for build_type in build_types:
 129         yield os.path.join(chrome_root, build_dir, build_type)
 130
 131
 132 def GetSequentialFileName(base_name):
 133   """Returns the next sequential file name based on |base_name| and the
 134   existing files. base_name should not contain extension.
 135   e.g: if base_name is /tmp/test, and /tmp/test_000.json,
 136   /tmp/test_001.mp3 exist, this returns /tmp/test_002. In case no
 137   other sequential file name exist, this will return /tmp/test_000
 138   """
 139   name, ext = os.path.splitext(base_name)
 140   assert ext == '', 'base_name cannot contain file extension.'
 141   index = 0
 142   while True:
 143     output_name = '%s_%03d' % (name, index)
 144     if not glob.glob(output_name + '.*'):
 145       break
 146     index = index + 1
 147   return output_name