Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use run_locally method in hypothesis

Best Python code snippet using hypothesis

comparative_tester.py

Source:comparative_tester.py

1#!/usr/bin/env python32# Copyright 2018 The Chromium Authors. All rights reserved.3# Use of this source code is governed by a BSD-style license that can be4# found in the LICENSE file.5# This script takes in a list of test targets to be run on both Linux and6# Fuchsia devices and then compares their output to each other, extracting the7# relevant performance data from the output of gtest.8import argparse9import logging10import os11import re12import subprocess13import sys14import time15from collections import defaultdict16from typing import Tuple, Dict, List17import target_spec18import test_results19def RunCommand(command: List[str], msg: str) -> str:20  """Runs a command and returns the standard output.21  Args:22    command (List[str]): The list of command chunks to use in subprocess.run.23        ex: ['git', 'grep', 'cat'] to find all instances of cat in a repo.24    msg (str): An error message in case the subprocess fails for some reason.25  Raises:26    subprocess.SubprocessError: Raises this with the command that failed in the27        event that the return code of the process is non-zero.28  Returns:29    str: the standard output of the subprocess.30  """31  command = [piece for piece in command if piece != ""]32  proc = subprocess.run(33      command,34      stdout=subprocess.PIPE,35      stderr=subprocess.PIPE,36      stdin=subprocess.DEVNULL)37  out = proc.stdout.decode("utf-8", errors="ignore")38  err = proc.stderr.decode("utf-8", errors="ignore")39  if proc.returncode != 0:40    sys.stderr.write("{}\nreturn code: {}\nstdout: {}\nstderr: {}".format(41        msg, proc.returncode, out, err))42    raise subprocess.SubprocessError(43        "Command failed to complete successfully. {}".format(command))44  return out45# TODO(crbug.com/848465): replace with --test-launcher-filter-file directly46def ParseFilterFile(filepath: str,47                    p_filts: List[str],48                    n_filts: List[str]) -> str:49  """Takes a path to a filter file, parses it, and constructs a gtest_filter50  string for test execution.51  Args:52    filepath (str): The path to the filter file to be parsed into a53        --gtest_filter flag.54    p_filts (List[str]): An initial set of positive filters passed in a flag.55    n_filts (List[str]): An initial set of negative filters passed in a flag.56  Returns:57    str: The properly-joined together gtest_filter flag.58  """59  positive_filters = p_filts60  negative_filters = n_filts61  with open(filepath, "r") as file:62    for line in file:63      # Only take the part of a line before a # sign64      line = line.split("#", 1)[0].strip()65      if line == "":66        continue67      elif line.startswith("-"):68        negative_filters.append(line[1:])69      else:70        positive_filters.append(line)71  return "--gtest_filter={}-{}".format(":".join(positive_filters),72                                       ":".join(negative_filters))73class TestTarget(object):74  """TestTarget encapsulates a single BUILD.gn target, extracts a name from the75  target string, and manages the building and running of the target for both76  Linux and Fuchsia.77  """78  def __init__(self, target: str, p_filts: List[str], n_filts: List[str]):79    self._target = target80    self._name = target.split(":")[-1]81    self._filter_file = "testing/buildbot/filters/fuchsia.{}.filter".format(82        self._name)83    if not os.path.isfile(self._filter_file):84      self._filter_flag = ""85      self._filter_file = ""86    else:87      self._filter_flag = ParseFilterFile(self._filter_file, p_filts, n_filts)88  def ExecFuchsia(self, out_dir: str, run_locally: bool) -> str:89    """Execute this test target's test on Fuchsia, either with QEMU or on actual90    hardware.91    Args:92      out_dir (str): The Fuchsia output directory.93      run_locally (bool): Whether to use QEMU(true) or a physical device(false)94    Returns:95      str: The standard output of the test process.96    """97    runner_name = "{}/bin/run_{}".format(out_dir, self._name)98    command = [runner_name, self._filter_flag, "--exclude-system-logs"]99    if not run_locally:100      command.append("-d")101    return RunCommand(command,102                      "Test {} failed on Fuchsia!".format(self._target))103  def ExecLinux(self, out_dir: str, run_locally: bool) -> str:104    """Execute this test target's test on Linux, either with QEMU or on actual105    hardware.106    Args:107      out_dir (str): The Linux output directory.108      run_locally (bool): Whether to use the host machine(true) or a physical109          device(false)110    Returns:111      str: The standard output of the test process.112    """113    command = []  # type: List[str]114    user = target_spec.linux_device_user115    ip = target_spec.linux_device_ip116    host_machine = "{0}@{1}".format(user, ip)117    if not run_locally:118      # Next is the transfer of all the directories to the destination device.119      self.TransferDependencies(out_dir, host_machine)120      command = [121          "ssh", "{}@{}".format(user, ip), "{1}/{0}/{1} -- {2}".format(122              out_dir, self._name, self._filter_flag)123      ]124    else:125      local_path = "{}/{}".format(out_dir, self._name)126      command = [local_path, "--", self._filter_flag]127    return RunCommand(command, "Test {} failed on linux!".format(self._target))128  def TransferDependencies(self, out_dir: str, host: str):129    """Transfer the dependencies of this target to the machine to execute the130    test.131    Args:132      out_dir (str): The output directory to find the dependencies in.133      host (str): The IP address of the host to receive the dependencies.134    """135    gn_desc = ["gn", "desc", out_dir, self._target, "runtime_deps"]136    out = RunCommand(137        gn_desc, "Failed to get dependencies of target {}".format(self._target))138    paths = []139    for line in out.split("\n"):140      if line == "":141        continue142      line = out_dir + "/" + line.strip()143      line = os.path.abspath(line)144      paths.append(line)145    common = os.path.commonpath(paths)146    paths = [os.path.relpath(path, common) for path in paths]147    archive_name = self._name + ".tar.gz"148    # Compress the dependencies of the test.149    command = ["tar", "-czf", archive_name] + paths150    if self._filter_file != "":151      command.append(self._filter_file)152    RunCommand(153        command,154        "{} dependency compression failed".format(self._target),155    )156    # Make sure the containing directory exists on the host, for easy cleanup.157    RunCommand(["ssh", host, "mkdir -p {}".format(self._name)],158               "Failed to create directory on host for {}".format(self._target))159    # Transfer the test deps to the host.160    RunCommand(161        [162            "scp", archive_name, "{}:{}/{}".format(host, self._name,163                                                   archive_name)164        ],165        "{} dependency transfer failed".format(self._target),166    )167    # Decompress the dependencies once they're on the host.168    RunCommand(169        [170            "ssh", host, "tar -xzf {0}/{1} -C {0}".format(171                self._name, archive_name)172        ],173        "{} dependency decompression failed".format(self._target),174    )175    # Clean up the local copy of the archive that is no longer needed.176    RunCommand(177        ["rm", archive_name],178        "{} dependency archive cleanup failed".format(self._target),179    )180def RunTest(target: TestTarget, run_locally: bool = False) -> None:181  """Run the given TestTarget on both Linux and Fuchsia182  Args:183    target (TestTarget): The TestTarget to run.184    run_locally (bool, optional): Defaults to False. Whether the test should be185        run on the host machine, or sent to remote devices for execution.186  Returns:187    None: Technically an IO (), as it writes to the results files188  """189  linux_out = target.ExecLinux(target_spec.linux_out_dir, run_locally)190  linux_result = test_results.TargetResultFromStdout(linux_out.splitlines(),191                                                     target._name)192  print("Ran Linux")193  fuchsia_out = target.ExecFuchsia(target_spec.fuchsia_out_dir, run_locally)194  fuchsia_result = test_results.TargetResultFromStdout(fuchsia_out.splitlines(),195                                                       target._name)196  print("Ran Fuchsia")197  outname = "{}.{}.json".format(target._name, time.time())198  linux_result.WriteToJson("{}/{}".format(target_spec.raw_linux_dir, outname))199  fuchsia_result.WriteToJson("{}/{}".format(target_spec.raw_fuchsia_dir,200                                            outname))201  print("Wrote result files")202def RunGnForDirectory(dir_name: str, target_os: str, is_debug: bool) -> None:203  """Create the output directory for test builds for an operating system.204  Args:205    dir_name (str): The name to use for the output directory. This will be206        created if it does not exist.207    target_os (str): The operating system to initialize this directory for.208    is_debug (bool): Whether or not this is a debug build of the tests in209        question.210  Returns:211    None: It has a side effect of replacing args.gn212  """213  if not os.path.exists(dir_name):214    os.makedirs(dir_name)215  debug_str = str(is_debug).lower()216  with open("{}/{}".format(dir_name, "args.gn"), "w") as args_file:217    args_file.write("is_debug = {}\n".format(debug_str))218    args_file.write("dcheck_always_on = false\n")219    args_file.write("is_component_build = false\n")220    args_file.write("use_goma = true\n")221    args_file.write("target_os = \"{}\"\n".format(target_os))222  subprocess.run(["gn", "gen", dir_name]).check_returncode()223def GenerateTestData(do_config: bool, do_build: bool, num_reps: int,224                     is_debug: bool, filter_flag: str):225  """Initializes directories, builds test targets, and repeatedly executes them226  on both operating systems227  Args:228    do_config (bool): Whether or not to run GN for the output directories229    do_build (bool): Whether or not to run ninja for the test targets.230    num_reps (int): How many times to run each test on a given device.231    is_debug (bool): Whether or not this should be a debug build of the tests.232    filter_flag (str): The --gtest_filter flag, to be parsed as such.233  """234  # Find and make the necessary directories235  DIR_SOURCE_ROOT = os.path.abspath(236      os.path.join(os.path.dirname(__file__), *([os.pardir] * 3)))237  os.chdir(DIR_SOURCE_ROOT)238  os.makedirs(target_spec.results_dir, exist_ok=True)239  os.makedirs(target_spec.raw_linux_dir, exist_ok=True)240  os.makedirs(target_spec.raw_fuchsia_dir, exist_ok=True)241  # Grab parameters from config file.242  linux_dir = target_spec.linux_out_dir243  fuchsia_dir = target_spec.fuchsia_out_dir244  # Parse filters passed in by flag245  pos_filter_chunk, neg_filter_chunk = filter_flag.split("-", 1)246  pos_filters = pos_filter_chunk.split(":")247  neg_filters = neg_filter_chunk.split(":")248  test_input = []  # type: List[TestTarget]249  for target in target_spec.test_targets:250    test_input.append(TestTarget(target, pos_filters, neg_filters))251  print("Test targets collected:\n{}".format(",".join(252      [test._target for test in test_input])))253  if do_config:254    RunGnForDirectory(linux_dir, "linux", is_debug)255    RunGnForDirectory(fuchsia_dir, "fuchsia", is_debug)256    print("Ran GN")257  elif is_debug:258    logging.warning("The --is_debug flag is ignored unless --do_config is also \259                     specified")260  if do_build:261    # Build test targets in both output directories.262    for directory in [linux_dir, fuchsia_dir]:263      build_command = ["autoninja", "-C", directory] \264                    + [test._target for test in test_input]265      RunCommand(build_command,266                 "autoninja failed in directory {}".format(directory))267    print("Builds completed.")268  # Execute the tests, one at a time, per system, and collect their results.269  for i in range(0, num_reps):270    print("Running Test set {}".format(i))271    for test_target in test_input:272      print("Running Target {}".format(test_target._name))273      RunTest(test_target)274      print("Finished {}".format(test_target._name))275  print("Tests Completed")276def main() -> int:277  cmd_flags = argparse.ArgumentParser(278      description="Execute tests repeatedly and collect performance data.")279  cmd_flags.add_argument(280      "--do-config",281      action="store_true",282      help="WARNING: This flag over-writes args.gn in the directories "283           "configured. GN is executed before running the tests.")284  cmd_flags.add_argument(285      "--do-build",286      action="store_true",287      help="Build the tests before running them.")288  cmd_flags.add_argument(289      "--is-debug",290      action="store_true",291      help="This config-and-build cycle is a debug build")292  cmd_flags.add_argument(293      "--num-repetitions",294      type=int,295      default=1,296      help="The number of times to execute each test target.")297  cmd_flags.add_argument(298      "--gtest_filter",299      type=str,300      default="",301  )302  cmd_flags.parse_args()303  GenerateTestData(cmd_flags.do_config, cmd_flags.do_build,304                   cmd_flags.num_repetitions, cmd_flags.is_debug,305                   cmd_flags.gtest_filter)306  return 0307if __name__ == "__main__":...

base.py

Source:base.py

1import logging2log = logging.getLogger('fabric.fabalicious.base')3from fabric.api import *4from fabric.state import output, env5from fabric.context_managers import env6from fabric.network import *7from fabric.contrib.files import exists8import re9class LocallyContext():10    def __init__(self, parent, config):11      self.parent= parent12      self.config = config13    def __enter__(self):14      self.saved = self.parent.run_locally15      self.parent.setRunLocally(self.config)16      return self17    def __exit__(self, type, value, traceback):18      self.parent.run_locally = self.saved19class BaseMethod(object):20  verbose_output = True21  run_locally = False22  @staticmethod23  def supports(methodName):24    return False25  @staticmethod26  def validateConfig(config):27    return {}28  @staticmethod29  def getDefaultConfig(config, settings, defaults):30    pass31  @staticmethod32  def applyConfig(config, settings):33    pass34  @staticmethod35  def getGlobalSettings():36    return {}37  @staticmethod38  def addExecutables(config, executables):39    for e in executables:40      if e not in config['executables']:41        config['executables'][e] = e42  def __init__(self, methodName, factory):43    self.methodName = methodName44    self.factory = factory45  @staticmethod46  def getOverrides():47    return False48  def preflight(self, task, config, **kwargs):49    pass50  def postflight(self, task, config, **kwargs):51    pass52  def fallback(self, task, config, **kwargs):53    pass54  def expandVariablesImpl(self, prefix, variables, result):55    for key in variables:56      if isinstance(variables[key], dict):57        self.expandVariablesImpl(prefix + "." + key, variables[key], result)58      elif isinstance(variables[key], list):59        pass # lists are not supported.60      else:61        result["%" + prefix + "." + key + "%"] = str(variables[key])62  def expandVariables(self, variables):63    results = {}64    for key in variables:65      self.expandVariablesImpl(key, variables[key], results)66    return results67  def expandCommands(self, commands, replacements):68    parsed_commands = []69    pattern = re.compile('|'.join(re.escape(key) for key in replacements.keys()))70    if commands:71      for line in commands:72        result = pattern.sub(lambda x: replacements[x.group()], line)73        parsed_commands.append(result)74    return parsed_commands75  def addPasswordToFabricCache(self, user, host, port, password, **kwargs):76    host_string = join_host_strings(user, host, port)77    env.passwords[host_string] = password78  def list_remote_files(self, base_folder, patterns):79    result = []80    with cd(base_folder), hide('running', 'output', 'warnings'), warn_only():81      for pattern in patterns:82        cmd = 'ls -l ' + pattern + ' 2>/dev/null'83        output = run(cmd)84        lines = output.stdout.splitlines()85        for line in lines:86          tokens = line.split()87          if(len(tokens) >= 9):88            result.append(tokens[8])89    return result90  def get_backup_result(self, config, file, hash, method):91    tokens = hash.split('--')92    if len(tokens) < 3:93      return False94    # be backwards compatible.95    if tokens[0] != config['config_name']:96      tokens[0], tokens[1] = tokens[1], tokens[0]97    if tokens[0] != config['config_name']:98      return False99    return {100      'config': tokens[0],101      'commit': tokens[1],102      'date':   tokens[2],103      'time':   tokens[3],104      'method': method,105      'hash':   hash,106      'file':   file107    }108  def get_backup_result_for_method(self, files, method):109    file = filter(lambda f: f['method'] == method, files)110    if len(file) != 1:111      return False112    return file[0]113  def runLocally(self, config):114    return LocallyContext(self, config)115  def setRunLocally(self, config):116    self.run_locally = config['runLocally']117    self.setExecutables(config)118  def setExecutables(self, config):119    self.executables = {}120    replacements = self.expandVariables({ "host": config })121    for key, command in config['executables'].iteritems():122      cmds = [ command ];123      cmds = self.expandCommands(cmds, replacements)124      self.executables[key] = cmds[0]125  def cd(self, path):126    # log.error('cd: %d %s'% (self.run_locally,  path))127    return lcd(path) if self.run_locally else cd(path)128  def expandCommand(self, in_cmd):129    cmd = in_cmd130    if len(self.executables) > 0:131      pattern = re.compile('|'.join(re.escape("#!" + key)+"\s" for key in self.executables.keys()))132      cmd = pattern.sub(lambda x: self.executables[x.group()[2:-1]] + ' ', cmd)133      if cmd.find('%arguments%') >= 0:134        arr = in_cmd.split(' ')135        command = arr.pop(0)136        arguments = ' '.join(arr)137        command = command.replace('#!', '');138        cmd = self.executables[command].replace('%arguments%', arguments)139    return cmd140  def run(self, cmd, **kwargs):141    # log.error("run: %d %s" % ( self.run_locally, cmd))142    cmd = self.expandCommand(cmd)143    if self.run_locally:144      return local(cmd, **kwargs)145    else:146      if 'capture' in kwargs:147        kwargs.pop('capture')148      return run(cmd, **kwargs)149  def exists(self, fname):150    return os.path.isfile(fname) if self.run_locally else exists(fname)151  def run_quietly(self, cmd, msg = '', hide_output = None, may_fail=False):152    if 'warn_only' in env and env['warn_only']:153      may_fail = True154    if msg != '':155      print msg156    if not hide_output:157      hide_output = ['running', 'output', 'warnings']158    if self.verbose_output:159      hide_output=[]160    with hide(*hide_output):161      try:162        result = self.run(cmd)163        if not may_fail and result.return_code != 0:164          log.error('%s failed:' %s)165          print result166        return result167      except:168        log.error('%s failed' % cmd)169        if output['aborts']:...

train.py

Source:train.py

...27        help = "The type of dataset.")28    parser.add_argument("-o", "--output-directory", default="predictions.csv",29        help = "The output directory the save the output of the inference run.")30    arguments = vars(parser.parse_args())31    run_locally(arguments)32################################################################################33## Guard Main34if __name__ == "__main__":35    main()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.