KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run hypothesis automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
