Best Python code snippet using tox_python
tests.py
Source:tests.py
...16if not which("snakemake"):17 raise Exception("snakemake not in PATH. For testing, install snakemake with "18 "'pip install -e .'. You should do this in a separate environment "19 "(via conda or virtualenv).")20def dpath(path):21 """get path to a data file (relative to the directory this22 test lives in)"""23 return os.path.realpath(join(os.path.dirname(__file__), path))24def md5sum(filename):25 data = open(filename, 'rb').read()26 return hashlib.md5(data).hexdigest()27def is_connected():28 try:29 urllib.request.urlopen("http://www.google.com", timeout=1)30 return True31 except urllib.request.URLError:32 return False33def run(path,34 shouldfail=False,35 needs_connection=False,36 snakefile="Snakefile",37 subpath=None,38 check_md5=True, cores=3, **params):39 """40 Test the Snakefile in path.41 There must be a Snakefile in the path and a subdirectory named42 expected-results.43 """44 if needs_connection and not is_connected():45 print("Skipping test because of missing internet connection",46 file=sys.stderr)47 return False48 results_dir = join(path, 'expected-results')49 snakefile = join(path, snakefile)50 assert os.path.exists(snakefile)51 assert os.path.exists(results_dir) and os.path.isdir(52 results_dir), '{} does not exist'.format(results_dir)53 with tempfile.TemporaryDirectory(prefix=".test", dir=os.path.abspath(".")) as tmpdir:54 config = {}55 if subpath is not None:56 # set up a working directory for the subworkflow and pass it in `config`57 # for now, only one subworkflow is supported58 assert os.path.exists(subpath) and os.path.isdir(59 subpath), '{} does not exist'.format(subpath)60 subworkdir = os.path.join(tmpdir, "subworkdir")61 os.mkdir(subworkdir)62 call('find {} -maxdepth 1 -type f -print0 | xargs -0 -I%% -n 1 cp %% {}'.format(63 quote(subpath), quote(subworkdir)),64 shell=True)65 config['subworkdir'] = subworkdir66 call('find {} -maxdepth 1 -type f -print0 | xargs -0 -I%% -n 1 cp -r %% {}'.format(67 quote(path), quote(tmpdir)),68 shell=True)69 success = snakemake(snakefile,70 cores=cores,71 workdir=tmpdir,72 stats="stats.txt",73 config=config, **params)74 if shouldfail:75 assert not success, "expected error on execution"76 else:77 assert success, "expected successful execution"78 for resultfile in os.listdir(results_dir):79 if resultfile == ".gitignore" or not os.path.isfile(80 os.path.join(results_dir, resultfile)):81 # this means tests cannot use directories as output files82 continue83 targetfile = join(tmpdir, resultfile)84 expectedfile = join(results_dir, resultfile)85 assert os.path.exists(86 targetfile), 'expected file "{}" not produced'.format(87 resultfile)88 if check_md5:89 # if md5sum(targetfile) != md5sum(expectedfile):90 # import pdb; pdb.set_trace()91 assert md5sum(targetfile) == md5sum(92 expectedfile), 'wrong result produced for file "{}"'.format(93 resultfile)94def test01():95 run(dpath("test01"))96def test02():97 run(dpath("test02"))98def test03():99 run(dpath("test03"), targets=['test.out'])100def test04():101 run(dpath("test04"), targets=['test.out'])102def test05():103 run(dpath("test05"))104def test06():105 run(dpath("test06"), targets=['test.bla.out'])106def test07():107 run(dpath("test07"), targets=['test.out', 'test2.out'])108def test08():109 run(dpath("test08"), targets=['test.out', 'test2.out'])110def test09():111 run(dpath("test09"), shouldfail=True)112def test10():113 run(dpath("test10"))114def test11():115 run(dpath("test11"))116def test12():117 run(dpath("test12"))118def test13():119 run(dpath("test13"))120def test14():121 run(dpath("test14"), snakefile="Snakefile.nonstandard", cluster="./qsub")122def test15():123 run(dpath("test15"))124def test_ancient():125 run(dpath("test_ancient"), targets=['D'])126def test_report():127 run(dpath("test_report"), check_md5=False)128def test_dynamic():129 run(dpath("test_dynamic"))130def test_params():131 run(dpath("test_params"))132def test_same_wildcard():133 run(dpath("test_same_wildcard"))134def test_conditional():135 run(dpath("test_conditional"),136 targets="test.out test.0.out test.1.out test.2.out".split())137def test_unpack_dict():138 run(dpath("test_unpack_dict"))139def test_unpack_list():140 run(dpath("test_unpack_list"))141def test_shell():142 run(dpath("test_shell"))143def test_temp():144 run(dpath("test_temp"),145 cluster="./qsub",146 targets="test.realigned.bam".split())147def test_keyword_list():148 run(dpath("test_keyword_list"))149def test_subworkflows():150 run(dpath("test_subworkflows"), subpath=dpath("test02"))151def test_globwildcards():152 run(dpath("test_globwildcards"))153def test_local_import():154 run(dpath("test_local_import"))155def test_ruledeps():156 run(dpath("test_ruledeps"))157def test_persistent_dict():158 try:159 import pytools160 run(dpath("test_persistent_dict"))161 except ImportError:162 pass163def test_url_include():164 run(dpath("test_url_include"), needs_connection=True)165def test_touch():166 run(dpath("test_touch"))167def test_config():168 run(dpath("test_config"))169def test_update_config():170 run(dpath("test_update_config"))171def test_wildcard_keyword():172 run(dpath("test_wildcard_keyword"))173def test_benchmark():174 run(dpath("test_benchmark"), check_md5=False)175def test_temp_expand():176 run(dpath("test_temp_expand"))177def test_wildcard_count_ambiguity():178 run(dpath("test_wildcard_count_ambiguity"))179def test_cluster_dynamic():180 run(dpath("test_cluster_dynamic"), cluster="./qsub")181def test_dynamic_complex():182 run(dpath("test_dynamic_complex"))183def test_srcdir():184 run(dpath("test_srcdir"))185def test_multiple_includes():186 run(dpath("test_multiple_includes"))187def test_yaml_config():188 run(dpath("test_yaml_config"))189# TODO reenable once S3Mocked works with boto3190# def test_remote():191# try:192# import moto193# import boto3194# import filechunkio195#196# # only run the remote file test if the dependencies197# # are installed, otherwise do nothing198# run(dpath("test_remote"), cores=1)199# except ImportError:200# pass201def test_cluster_sync():202 run(dpath("test14"),203 snakefile="Snakefile.nonstandard",204 cluster_sync="./qsub")205def test_symlink_temp():206 run(dpath("test_symlink_temp"), shouldfail=True)207def test_empty_include():208 run(dpath("test_empty_include"))209def test_script():210 run(dpath("test_script"))211def test_shadow():212 run(dpath("test_shadow"))213def test_until():214 run(dpath("test_until"),215 until=["leveltwo_first", # rule name216 "leveltwo_second.txt", # file name217 "second_wildcard"]) # wildcard rule218def test_omitfrom():219 run(dpath("test_omitfrom"),220 omit_from=["leveltwo_first", # rule name221 "leveltwo_second.txt", # file name222 "second_wildcard"]) # wildcard rule223def test_nonstr_params():224 run(dpath("test_nonstr_params"))225def test_delete_output():226 run(dpath("test_delete_output"), cores=1)227def test_input_generator():228 run(dpath("test_input_generator"))229def test_symlink_time_handling():230 #See Snakefile for notes on why this fails on some systems231 if os.utime in os.supports_follow_symlinks:232 run(dpath("test_symlink_time_handling"))233def test_issue328():234 try:235 import pytools236 run(dpath("test_issue328"), forcerun=["split"])237 except ImportError:238 # skip test if import fails239 pass240def test_conda():241 if conda_available():242 run(dpath("test_conda"), use_conda=True)243def test_conda_custom_prefix():244 if conda_available():245 run(dpath("test_conda_custom_prefix"),246 use_conda=True, conda_prefix="custom")247def test_wrapper():248 if conda_available():249 run(dpath("test_wrapper"), use_conda=True)250def conda_available():251 return which("conda")252def test_get_log_none():253 run(dpath("test_get_log_none"))254def test_get_log_both():255 run(dpath("test_get_log_both"))256def test_get_log_stderr():257 run(dpath("test_get_log_stderr"))258def test_get_log_stdout():259 run(dpath("test_get_log_stdout"))260def test_get_log_complex():261 run(dpath("test_get_log_complex"))262def test_spaces_in_fnames():263 run(dpath("test_spaces_in_fnames"),264 # cluster="./qsub",265 targets=["test bam file realigned.bam"],266 verbose=True,267 printshellcmds=True)268# TODO deactivate because of problems with moto and boto3.269# def test_static_remote():270# import importlib271# try:272# importlib.reload(boto3)273# importlib.reload(moto)274# # only run the remote file test if the dependencies275# # are installed, otherwise do nothing276# run(dpath("test_static_remote"), cores=1)277# except ImportError:278# pass279def test_remote_ncbi_simple():280 try:281 import Bio282 # only run the remote file test if the dependencies283 # are installed, otherwise do nothing284 run(dpath("test_remote_ncbi_simple"))285 except ImportError:286 pass287def test_remote_ncbi():288 try:289 import Bio290 # only run the remote file test if the dependencies291 # are installed, otherwise do nothing292 run(dpath("test_remote_ncbi"))293 except ImportError:294 pass295def test_deferred_func_eval():296 run(dpath("test_deferred_func_eval"))297def test_format_params():298 run(dpath("test_format_params"), check_md5=True)299def test_rule_defined_in_for_loop():300 # issue 257301 run(dpath("test_rule_defined_in_for_loop"))302def test_issue381():303 run(dpath("test_issue381"))304def test_format_wildcards():305 run(dpath("test_format_wildcards"))306def test_with_parentheses():307 run(dpath("test (with parentheses)"))308def test_dup_out_patterns():309 """Duplicate output patterns should emit an error310 Duplicate output patterns can be detected on the rule level311 """312 run(dpath("test_dup_out_patterns"), shouldfail=True)313def test_restartable_job_cmd_exit_1():314 """Test the restartable job feature on ``exit 1``315 The shell snippet in the Snakemake file will fail the first time316 and succeed the second time.317 """318 # Even two consecutive times should fail as files are cleared319 run(dpath("test_restartable_job_cmd_exit_1"), cluster="./qsub",320 restart_times=0, shouldfail=True)321 run(dpath("test_restartable_job_cmd_exit_1"), cluster="./qsub",322 restart_times=0, shouldfail=True)323 # Restarting once is enough324 run(dpath("test_restartable_job_cmd_exit_1"), cluster="./qsub",325 restart_times=1, printshellcmds=True)326def test_restartable_job_qsub_exit_1():327 """Test the restartable job feature when qsub fails328 The qsub in the sub directory will fail the first time and succeed the329 second time.330 """331 # Even two consecutive times should fail as files are cleared332 run(dpath("test_restartable_job_qsub_exit_1"), cluster="./qsub",333 restart_times=0, shouldfail=True)334 run(dpath("test_restartable_job_qsub_exit_1"), cluster="./qsub",335 restart_times=0, shouldfail=True)336 # Restarting once is enough337 run(dpath("test_restartable_job_qsub_exit_1"), cluster="./qsub",338 restart_times=1, shouldfail=False)339def test_threads():340 run(dpath("test_threads"), cores=20)341def test_dynamic_temp():342 run(dpath("test_dynamic_temp"))343# TODO this currently hangs. Has to be investigated (issue #660).344#def test_ftp_immediate_close():345# try:346# import ftputil347#348# # only run the remote file test if the dependencies349# # are installed, otherwise do nothing350# run(dpath("test_ftp_immediate_close"))351# except ImportError:352# pass353def test_issue260():354 run(dpath("test_issue260"))355# TODO reenable once S3Mocked works again with boto3356# def test_default_remote():357# run(dpath("test_default_remote"),358# default_remote_provider="S3Mocked",359# default_remote_prefix="test-remote-bucket")360def test_run_namedlist():361 run(dpath("test_run_namedlist"))362def test_remote_gs():363 run(dpath("test_remote_gs"))364def test_remote_log():365 run(dpath("test_remote_log"), shouldfail=True)366def test_profile():367 run(dpath("test_profile"))368# TODO reenable once we run tests in a VM instead of Docker (maybe go back to codeship)?369# def test_singularity():370# run(dpath("test_singularity"), use_singularity=True)371def test_issue612():372 run(dpath("test_issue612"), dryrun=True)373if __name__ == '__main__':374 import nose...
build.py
Source:build.py
1# Copyright (c) 2017-present, Facebook, Inc.2# All rights reserved.3# This source code is licensed under the BSD-style license found in the4# LICENSE file in the root directory of this source tree. An additional grant5# of patent rights can be found in the PATENTS file in the same directory.6# Download and build the data if it does not exist.7import parlai.core.build_data as build_data8import os9def buildImage(opt):10 dpath = os.path.join(opt['datapath'], 'COCO-IMG')11 version = '1'12 if not build_data.built(dpath, version_string=version):13 print('[building image data: ' + dpath + ']')14 if build_data.built(dpath):15 # An older version exists, so remove these outdated files.16 build_data.remove_dir(dpath)17 build_data.make_dir(dpath)18 # Download the image data.19 fname1 = 'train2014.zip'20 fname2 = 'val2014.zip'21 fname3 = 'test2015.zip'22 url = 'https://s3.amazonaws.com/fair-data/parlai/COCO-IMG/'23 build_data.download(url + fname1, dpath, fname1)24 build_data.download(url + fname2, dpath, fname2)25 build_data.download(url + fname3, dpath, fname3)26 build_data.untar(dpath, fname1)27 build_data.untar(dpath, fname2)28 build_data.untar(dpath, fname3)29 # Mark the data as built.30 build_data.mark_done(dpath, version_string=version)31def build(opt):32 dpath = os.path.join(opt['datapath'], 'VQA-v1')33 version = None34 if not build_data.built(dpath, version_string=version):35 print('[building data: ' + dpath + ']')36 if build_data.built(dpath):37 # An older version exists, so remove these outdated files.38 build_data.remove_dir(dpath)39 build_data.make_dir(dpath)40 # Download the data.41 fname1 = 'Questions_Train_mscoco.zip'42 fname2 = 'Questions_Val_mscoco.zip'43 fname3 = 'Questions_Test_mscoco.zip'44 fname4 = 'Annotations_Val_mscoco.zip'45 fname5 = 'Annotations_Train_mscoco.zip'46 url = 'http://visualqa.org/data/mscoco/vqa/'47 build_data.download(url + fname1, dpath, fname1)48 build_data.download(url + fname2, dpath, fname2)49 build_data.download(url + fname3, dpath, fname3)50 build_data.download(url + fname4, dpath, fname4)51 build_data.download(url + fname5, dpath, fname5)52 build_data.untar(dpath, fname1)53 build_data.untar(dpath, fname2)54 build_data.untar(dpath, fname3)55 build_data.untar(dpath, fname4)56 build_data.untar(dpath, fname5)57 # Mark the data as built....
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!!