How to use dpath method in tox

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....

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.