Package tlib :: Package base :: Module TestHelper |
[hide private]
[frames] | no frames]

# Source Code for Module tlib.base.TestHelper

      1  import os
      2  import socket
      3  import subprocess
      4  import logging
      5  import json
      6  import tempfile
      7  from glob import glob
      8  from requests.models import Response
      9
     10
     11 -class Singleton(type):
     12      _instances = {}
     13
     14 -    def __call__(cls, *args, **kwargs):
     15          if cls not in cls._instances:
     16              cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
     17          return cls._instances[cls]
     18
     19
     20 -def tlib_folder():
     21      """
     22      Returns tlib's absolute path
     23      """
     24      file_folder = os.path.dirname(__file__)
     25      return os.path.abspath(os.path.join(file_folder, os.pardir))
     26
     27
     28 -def tlib_asset_folder():
     29      """
     30      Returns absolute path of tlib's asset folder
     31      """
     32      return os.path.abspath(os.path.join(tlib_folder(), "asset"))
     33
     34
     35 -def tlib_modules_folder():
     36      """
     37      Returns absolute path of folder containing all modules
     38      """
     39      return os.path.abspath(os.path.join(tlib_folder(), "base"))
     40
     41
     42 -def tlib_config_folder():
     43      """
     44      Returns absolute path of tlib's config folder
     45      """
     46      return os.path.abspath(os.path.join(tlib_folder(), "config"))
     47
     48
     49 -def tlib_template_folder():
     50      """
     51      Returns absolute path of tlib's template folder\n
     52      Template folder contains jinja templates used for generation of reports\n
     53      like screenshots
     54      """
     55      return os.path.abspath(os.path.join(tlib_folder(), "templates"))
     56
     57
     58 -def webdriver_chrome_executable(version="latest"):
     59      """
     60      Returns path to Chrome WebDriver executable
     61      """
     62      if version == "latest":
     63          #Find all cromedriver files and sort by filename
     64          path = os.path.join(tlib_asset_folder(), "selenium", "chromedriver_*.exe")
     65          r = sorted(glob(path))
     66          return os.path.join(tlib_asset_folder(), "selenium", r[-1])
     67      else:
     68          return os.path.join(tlib_asset_folder(), "selenium", "chromedriver_%s.exe" % version)
     69
     70
     71 -def webdriver_ie_executable():
     72      """
     73      Alias for webdriver_ie32_executable
     74      """
     75      return webdriver_ie32_executable()
     76
     77
     78 -def webdriver_ie32_executable():
     79      """
     80      Returns path to IE WebDriver executable
     81      """
     82      return os.path.join(tlib_asset_folder(), "selenium", "IE", "2.41.0", "Win32", "IEDriverServer.exe")
     83
     84
     85 -def webdriver_ie64_executable():
     86      """
     87      Returns path to IE WebDriver executable
     88      """
     89      return os.path.join(tlib_asset_folder(), "selenium", "IE", "2.41.0", "x64", "IEDriverServer.exe")
     90
     91
     92 -def android_server_apk():
     93      """
     94      Returns path to Webdriver for android
     95      """
     96      return os.path.join(tlib_asset_folder(), "selenium", "android-server-2.21.0.apk")
     97
     98
     99 -def selendroid_server_jar():
    100      """
    101      Returns path to selendroid jar for android native app
    102      """
    103      return os.path.join(tlib_asset_folder(), "selenium", "selendroid-standalone-0.9.0-with-dependencies.jar")
    104
    105
    106 -def is_valid_ip(ip):
    107      """
    108      Returns true if IP parameter is a valid IP address.\n
    109      Currently IPs in this format are valid:\n
    110      X.X.X.X\n
    111      X.X.X.X:PORT
    112
    113      @type ip: str
    114      @return: bool
    115      """
    116
    117      if ip is None:
    118          return False
    119
    120      try:
    121          (ip, port) = ip.split(":", 1)
    122      except ValueError:
    123          #There is no ':' in the string
    124          port = None
    125
    126      try:
    127          socket.inet_aton(ip)
    128      except socket.error:
    129          return False
    130
    131      if port is not None:
    132          try:
    133              return (int(port) >= 1) and (int(port) <= 65535)
    134          except ValueError:
    135              #Not a valid integer
    136              return False
    137
    138      return True
    139
    140
    141 -def run_command(logger, command, shell=False, fail_on_error=True, cwd=None, wait_until_complete=True):
    142      """
    143      Run a command and skip test if exit code is not 0
    144
    145      Example:
    146      Run 'adb devices' and don't skip test if commands returns non-zero status
    147      run_command(logger, ["adb", "devices"], fail_on_error=False)
    148
    149      @param logger: logger for debugging purposes
    150      @type logger: logging.Logger
    151      @param command: Command to run
    152      @type command: list
    153      @param fail_on_error: When True, skip test if command returned a non-zero exit code
    154      @type fail_on_error: bool
    155      @type cwd: Working directory
    156      @type cwd: str
    157      @rtype: list
    158      @return: If wait_until_complete is True, Returns a list with stdout and stderr output
    159      otherwise returns process object
    160      """
    161
    162      if wait_until_complete:
    163          fd_out = tempfile.NamedTemporaryFile(delete=True)
    164          #With delete = True option, the files will be automatically removed after close
    165          fd_err = tempfile.NamedTemporaryFile(delete=True)
    166          try:
    167              process = subprocess.Popen(command, shell=shell, stdout=fd_out, stderr=fd_err, cwd=cwd)
    168          except WindowsError as e:
    169              logger.error(r"Problem running command.\n%s" % e)
    170              # noinspection PyUnresolvedReferences
    171              if fail_on_error:
    172                  raise RuntimeError(r"Problem running command.\n%s" % e)
    173
    174          # noinspection PyUnboundLocalVariable
    175          process.communicate()
    176          fd_out.seek(0)
    177          fd_err.seek(0)
    178          out = (fd_out.read(), fd_err.read())
    179          fd_out.close()
    180          fd_err.close()
    181          errcode = process.returncode
    182
    183          if (errcode != 0) and fail_on_error:
    184              logger.error(r"Program  exited with code {errcode}, skipping test\n{out}".format(errcode=errcode, out=out))
    185              # noinspection PyUnresolvedReferences
    186              raise RuntimeError(r"Program exited with code {errcode}, skipping test\n{out}".format(errcode=errcode, out=out))
    187
    188          return out
    189      else:
    190          try:
    191              process = subprocess.Popen(command, shell=shell, cwd=cwd)
    192          except WindowsError as e:
    193              logger.error(r"Problem running command.\n%s" % e)
    194              # noinspection PyUnresolvedReferences
    195              if fail_on_error:
    196                  raise RuntimeError(r"Problem running command.\n%s" % e)
    197          return process
    198
    199
    200 -def sort_list(l):
    201      """
    202      Sorts a list, and if the list has other objects inside, it will iterate over them
    203      @param l: list to sort
    204      @type l: list
    205      @return: sorted list
    206      @rtype: list
    207      """
    208      if type(l) is not list:
    209          raise RuntimeError("Parameter l is not a list")
    210
    211      for i, s in enumerate(l):
    212          if type(s) is list:
    213              l[i] = sort_list(s)
    214          if type(s) is dict:
    215              l[i] = sort_dict(s)
    216
    217      l.sort()
    218
    219      return l
    220
    221
    222 -def sort_dict(d):
    223      """
    224      Sorts a dictionary, and if the list has other objects inside, it will iterate over them
    225      @param d: dictionary to sort
    226      @type d: dict
    227      @return: sorted dictionary
    228      @rtype: dict
    229      """
    230      if type(d) is not dict:
    231          raise RuntimeError("Parameter d is not a dictionary")
    232
    233      for key in d.keys():
    234          if type(d[key]) is list:
    235              #Sort list
    236              d[key] = sort_list(d[key])
    237          elif type(d[key]) is dict:
    238              d[key] = sort_dict(d[key])
    239
    240      return d
    241
    242
    243 -def log_rest_request(response, logger):
    244      """
    245      Logs REST request and response
    246
    247      :param response: Object returned by requests class
    248      :param logger: Where to log request and response
    249      :type logger: logging.Logger
    250      """
    251
    252      logger.debug("Request\n\t{url} [{status_code} - {reason}]".
    253                   format(url=response.request.url, status_code=response.status_code, reason=response.reason))
    254
    255      try:
    256          #Try to get response as JSON
    257          json_resp = response.json()
    258          logger.debug("Response\n" + json.dumps(json_resp, sort_keys=True, indent=4, separators=(',', ': ')))
    259          return
    260      except ValueError:
    261          #not valid JSON
    262          pass
    263
    264      #log as text
    265      logger.debug("Response\n" + response.content)
    266