Best Python code snippet using slash
watcher.py
Source:watcher.py  
...73    def agent_path(self):74        return self.path('agent', self.job['id'])75    76    @property77    def suite_path(self):78        return self.path('suite', self.task['id'])79    80    def maybe_download_suite(self):81        if not os.path.isfile(self.suite_path):82            logger.info('Suite not found, downloading...')83            response = self.api.download(self.task['file_url'], self.suite_path)84            if response.status_code != 200:85                raise Exception('Suite download failed')86        with open(self.suite_path, 'rb') as f:87            file_hash = utils.hash_file(f)88            if file_hash != self.task['file_hash']:89                logger.info('Suite hash mismatch ({} != {}), updating...'.format(file_hash, self.task['file_hash']))90                response = self.api.download(self.task['file_url'], self.suite_path)91                if response.status_code != 200:...csuite.py
Source:csuite.py  
1#!/usr/bin/python2# Copyright 2018 the V8 project authors. All rights reserved.3'''4C S u i t e                                         because who can remember?5-----------------------------------------------------------------------------6python csuite.py [options] <benchmark> <mode> <d8 path>7Arguments8  benchmark: one of octane, sunspider or kraken.9  mode: baseline or compare.10  d8 path: a valid path to the d8 executable you want to use.11CSuite is a wrapper around benchmark.py and compare-baseline.py, old12friends in the d8 benchmarking world. Unlike those tools, it can be13run in any directory. It's also opinionated about which benchmarks it14will run, currently SunSpider, Octane and Kraken. Furthermore, it15runs the versions we pull into ./test/benchmarks/data.16Examples:17Say you want to see how much optimization buys you:18  ./csuite.py kraken baseline ~/src/v8/out/d8 -x="--noopt"19  ./csuite.py kraken compare ~/src/v8/out/d820Suppose you are comparing two binaries, quick n' dirty style:21  ./csuite.py -r 3 octane baseline ~/src/v8/out-master/d822  ./csuite.py -r 3 octane compare ~/src/v8/out-mine/d823You can run from any place:24  ../../somewhere-strange/csuite.py sunspider baseline ./d825  ../../somewhere-strange/csuite.py sunspider compare ./d8-better26'''27# for py2/py3 compatibility28from __future__ import print_function29import os30from optparse import OptionParser31import subprocess32import sys33if __name__ == '__main__':34  parser = OptionParser(usage=__doc__)35  parser.add_option("-r", "--runs", dest="runs",36      help="Override the default number of runs for the benchmark.")37  parser.add_option("-x", "--extra-arguments", dest="extra_args",38      help="Pass these extra arguments to d8.")39  parser.add_option("-v", "--verbose", action="store_true", dest="verbose",40      help="See more output about what magic csuite is doing.")41  (opts, args) = parser.parse_args()42  if len(args) < 3:43    print('not enough arguments')44    sys.exit(1)45  suite = args[0]46  mode = args[1]47  if suite not in ['octane', 'jetstream', 'sunspider', 'kraken']:48    print('Suite must be octane, jetstream, sunspider or kraken. Aborting.')49    sys.exit(1)50  if mode != 'baseline' and mode != 'compare':51    print('mode must be baseline or compare. Aborting.')52    sys.exit(1)53  # Set up paths.54  d8_path = os.path.abspath(args[2])55  if not os.path.exists(d8_path):56    print(d8_path + " is not valid.")57    sys.exit(1)58  csuite_path = os.path.dirname(os.path.abspath(__file__))59  if not os.path.exists(csuite_path):60    print("The csuite directory is invalid.")61    sys.exit(1)62  benchmark_py_path = os.path.join(csuite_path, "benchmark.py")63  if not os.path.exists(benchmark_py_path):64    print("Unable to find benchmark.py in " + csuite_path \65        + ". Aborting.")66    sys.exit(1)67  compare_baseline_py_path = os.path.join(csuite_path,68      "compare-baseline.py")69  if not os.path.exists(compare_baseline_py_path):70    print("Unable to find compare-baseline.py in " + csuite_path \71        + ". Aborting.")72    sys.exit(1)73  benchmark_path = os.path.abspath(os.path.join(csuite_path, "../data"))74  if not os.path.exists(benchmark_path):75    print("I can't find the benchmark data directory. Aborting.")76    sys.exit(1)77  # Gather the remaining arguments into a string of extra args for d8.78  extra_args = ""79  if opts.extra_args:80    extra_args = opts.extra_args81  if suite == "octane":82    runs = 1083    suite_path = os.path.join(benchmark_path, "octane")84    cmd = "run.js"85  elif suite == "jetstream":86    runs = 587    suite_path = os.path.join(benchmark_path, "JetStream2.0")88    cmd = "-m all_setup.js cli.js"89  elif suite == "kraken":90    runs = 8091    suite_path = os.path.join(benchmark_path, "kraken")92    cmd = os.path.join(csuite_path, "run-kraken.js")93  else:94    runs = 10095    suite_path = os.path.join(benchmark_path, "sunspider")96    cmd = os.path.join(csuite_path, "sunspider-standalone-driver.js")97  if opts.runs:98    if (float(opts.runs) / runs) < 0.6:99      print("Normally, %s requires %d runs to get stable results." \100          % (suite, runs))101    runs = int(opts.runs)102  if opts.verbose:103    print("Running and averaging %s %d times." % (suite, runs))104  # Ensure output directory is setup105  output_path_base = os.path.abspath(os.getcwd())106  output_path = os.path.join(output_path_base, "_results")107  output_file = os.path.join(output_path, "master_" + suite)108  if not os.path.exists(output_path):109    if opts.verbose:110      print("Creating directory %s." % output_path)111    os.mkdir(output_path)112  if opts.verbose:113    print("Working directory for runs is %s." % suite_path)114  inner_command = " -c \"%s --expose-gc %s %s \"" \115      % (d8_path, extra_args, cmd)116  if opts.verbose:117    print("calling d8 like so: %s." % inner_command)118  cmdline_base = "python %s %s -fv -r %d -d %s" \119      % (benchmark_py_path, inner_command, runs, output_path_base)120  if mode == "baseline":121    cmdline = "%s > %s" % (cmdline_base, output_file)122  else:123    output_file_compare = output_file + "_compare"124    cmdline = "%s > %s" % (cmdline_base, output_file_compare)125  if opts.verbose:126    print("Spawning subprocess: %s." % cmdline)127  return_code = subprocess.call(cmdline, shell=True, cwd=suite_path)128  if return_code < 0:129    print("Error return code: %d." % return_code)130  if mode == "baseline":131    print("Wrote %s." % output_file)132    print("Run %s again with compare mode to see results." % suite)133  else:134    print("Wrote %s." % output_file_compare)135    if suite == "octane" or suite == "jetstream":136      cmdline = "python %s  %s -f %s" % (compare_baseline_py_path, output_file, output_file_compare)137    else:138      cmdline = "python %s  %s -t -f %s" % (compare_baseline_py_path, output_file, output_file_compare)139    if opts.verbose:140      print("Spawning subprocess: %s." % cmdline)141    return_code = subprocess.call(cmdline, shell=True, cwd=suite_path)142    if return_code < 0:...swell_launch_experiment.py
Source:swell_launch_experiment.py  
1#!/usr/bin/env python2# (C) Copyright 2021-2022 United States Government as represented by the Administrator of the3# National Aeronautics and Space Administration. All Rights Reserved.4#5# This software is licensed under the terms of the Apache Licence Version 2.06# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.7# --------------------------------------------------------------------------------------------------8# standard imports9import click10import os11import subprocess12# local imports13from swell.utilities.logger import Logger14# --------------------------------------------------------------------------------------------------15class DeployWorkflow():16    def __init__(self, suite_path, experiment_name):17        self.logger = Logger('DeployWorkflow')18        self.suite_path = suite_path19        self.experiment_name = experiment_name20    # ----------------------------------------------------------------------------------------------21    def cylc_run_experiment(self):  # NB: Could be a factory based on workflow_manager22        # Move to the suite path23        os.chdir(self.suite_path)24        # Check flow.cylc is present in the provided suite path25        flow_path = os.path.join(self.suite_path, 'flow.cylc')26        if not os.path.exists(flow_path):27            self.logger.abort('In cylc_run_experiment the suite_path contains no flow.cylc file. ' +28                              'i.e. ' + flow_path + ' does not exist')29        # Check for user provided global.cylc30        if os.path.exists(self.suite_path + 'global.cylc'):31            os.environ['CYLC_CONF_PATH'] = self.suite_path32        # Install the suite33        subprocess.run(['cylc', 'install'], check=True)34        # Start the workflow35        subprocess.run(['cylc', 'play', self.experiment_name], check=True)36        # Pre TUI messages37        self.logger.info(' ')38        self.logger.info('Workflow is now running... ')39        self.logger.info(' ')40        self.logger.info('Use \'cylc scan\' to see running workflows.')41        self.logger.info(' ')42        self.logger.info('If the workflow needs to be stopped, close the TUI (if open)')43        self.logger.info('by pressing \'q\' and issue either:')44        self.logger.info('  cylc stop ' + self.experiment_name)45        self.logger.info('or to kill running tasks and stop:')46        self.logger.info('  cylc stop --kill ' + self.experiment_name)47        self.logger.info(' ')48        # Launch the job monitor49        self.logger.input('Launching the TUI, press \'q\' at any time to exit the TUI')50        subprocess.run(['cylc', 'tui', self.experiment_name], check=True)51# --------------------------------------------------------------------------------------------------52@click.command()53@click.option('-p', '--suite_path', 'suite_path', default=None,54              help='Directory containing the suite file needed by the workflow manager')55@click.option('-w', '--workflow_manager', 'workflow_manager', default='cylc',56              help='Workflow manager to be used')57def main(suite_path, workflow_manager):58    # Get the path to where the suite files are located59    # -------------------------------------------------60    if suite_path is None:61        suite_path = os.getcwd()62    else:63        suite_path = os.path.realpath(suite_path)  # Full absolute path64    # Get suite name65    # --------------66    experiment_name = os.path.basename(os.path.normpath(suite_path))67    # Create the deployment object68    # ----------------------------69    deploy_workflow = DeployWorkflow(suite_path, experiment_name)70    # Write some info for the user71    # ----------------------------72    deploy_workflow.logger.info('Launching workflow defined by files in ' + suite_path + '``.')73    deploy_workflow.logger.info('Experiment name: ' + experiment_name)74    deploy_workflow.logger.info('Workflow manager: ' + workflow_manager)75    # Launch the workflow76    # -------------------77    if workflow_manager == 'cylc':78        deploy_workflow.cylc_run_experiment()79# --------------------------------------------------------------------------------------------------80if __name__ == '__main__':81    main()...system_test.py
Source:system_test.py  
1import sys2import os3import importlib4import subprocess5import filecmp6import argparse7import json8def setup_output_dir(output_dir):9    if not os.path.exists(output_dir):10        os.mkdir(output_dir)11def execute_testsuite(exe, suite_path, suite_out_path, suite_ans_path):12    correct_count = 013    suite_name = os.path.basename(suite_path)14    suite_mod_path = os.path.join(suite_path, "suite.py")15    suite = None16    if os.path.exists(suite_mod_path):17        module_path = ".".join(["testcases", suite_name, "suite"])18        mod = importlib.import_module(module_path)19        suite = getattr(mod, "Suite")20    case_files = list()21    for name in os.listdir(suite_path):22        if name.startswith("t") and name.endswith(".txt"):23            case_files.append(name)24    print("Starting test suite ``{suite}``".format(suite=suite_name))25    failed_cases = list()26    for case in sorted(case_files):27        if suite:28            suite.setUp()29        case_path = os.path.join(suite_path, case)30        out_path = os.path.join(suite_out_path, case)31        ans_path = os.path.join(suite_ans_path, case)32        with open(case_path) as fp:33            content = fp.readlines()34        content.insert(0, ".output {out_path}\n".format(out_path=out_path))35        p = subprocess.Popen([exe], stdin=subprocess.PIPE) 36        for line in content:37            p.stdin.write(line.encode())38        p.stdin.close()39        try:40            p.wait(timeout=30)41        except subprocess.TimeoutExpired:42            print("Timeout")43        if suite:44            suite.tearDown()45        if os.path.isfile(out_path):46            is_result_match = filecmp.cmp(out_path, ans_path)47        else:48            is_result_match = False49        if is_result_match:50            print("The test file {case} passed".format(case=case))51            correct_count += 152        else:53            print("The test file {case} failed".format(case=case))54            failed_cases.append(case)55    print("The test suite ``{}`` total passed {}/{}".format(suite_name,56            correct_count, len(case_files)))57    print("Failed cases: {}".format(failed_cases))58    print()59    return correct_count, len(case_files), failed_cases60def main():61    parser = argparse.ArgumentParser()62    parser.add_argument("shell", help="The path of the compiled shell")63    parser.add_argument("test_case", help="The test case to be run, default: all", nargs="*", default="all")64    args = parser.parse_args()65    if not isinstance(args.test_case, list):66        args.test_case = "all"67    68    file_path = os.path.dirname(__file__)69    testcase_path = os.path.join(file_path, "testcases")70    output_path = os.path.join(file_path, "output")71    answer_path = os.path.join(file_path, "answer")72    if args.test_case == "all":73        args.test_case = os.listdir(testcase_path)74    75    result = dict()76    for test_suite in args.test_case:77        if test_suite == "__pycache__":78            continue79        suite_path = os.path.join(testcase_path, test_suite)80        suite_out_path = os.path.join(output_path, test_suite)81        suite_ans_path = os.path.join(answer_path, test_suite)82        if os.path.isdir(suite_path):83            result[test_suite] = dict()84            setup_output_dir(suite_out_path)85            ret = execute_testsuite(sys.argv[1], suite_path, suite_out_path, suite_ans_path)86            correct_count, total_count, failed_cases = ret87            result[test_suite]["correct"] = correct_count88            result[test_suite]["total"] = total_count89            result[test_suite]["failed"] = failed_cases90    with open("result.json", "w") as fp:91        json.dump(result, fp)92if __name__ == "__main__":...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
