KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_factors method in tox

Best Python code snippet using tox_python

test_examples.py

Source:test_examples.py

1import json2import logging3import os4import stat5import subprocess6import sys7from io import StringIO8from pathlib import Path9from typing import Any, Dict, List, Union, cast10import py.path11import pytest  # type: ignore12from ruamel.yaml.comments import CommentedMap, CommentedSeq13from schema_salad.exceptions import ValidationException14import cwltool.checker15import cwltool.expression as expr16import cwltool.factory17import cwltool.pathmapper18import cwltool.process19import cwltool.workflow20from cwltool.checker import can_assign_src_to_sink21from cwltool.context import RuntimeContext22from cwltool.errors import WorkflowException23from cwltool.main import main24from cwltool.process import CWL_IANA25from cwltool.sandboxjs import JavascriptException26from cwltool.utils import CWLObjectType, dedup, onWindows27from .util import (28    get_data,29    get_main_output,30    get_windows_safe_factory,31    needs_docker,32    needs_singularity,33    temp_dir,34    windows_needs_docker,35    working_directory,36)37sys.argv = [""]38expression_match = [39    ("(foo)", True),40    ("(foo.bar)", True),41    ("(foo['bar'])", True),42    ('(foo["bar"])', True),43    ("(foo.bar.baz)", True),44    ("(foo['bar'].baz)", True),45    ("(foo['bar']['baz'])", True),46    ("(foo['b\\'ar']['baz'])", True),47    ("(foo['b ar']['baz'])", True),48    ("(foo_bar)", True),49    ('(foo.["bar"])', False),50    ('(.foo["bar"])', False),51    ('(foo ["bar"])', False),52    ('( foo["bar"])', False),53    ("(foo[bar].baz)", False),54    ("(foo['bar\"].baz)", False),55    ("(foo['bar].baz)", False),56    ("{foo}", False),57    ("(foo.bar", False),58    ("foo.bar)", False),59    ("foo.b ar)", False),60    ("foo.b'ar)", False),61    ("(foo+bar", False),62    ("(foo bar", False),63]64@pytest.mark.parametrize("expression,expected", expression_match)  # type: ignore65def test_expression_match(expression: str, expected: bool) -> None:66    match = expr.param_re.match(expression)67    assert (match is not None) == expected68interpolate_input = {69    "foo": {70        "bar": {"baz": "zab1"},71        "b ar": {"baz": 2},72        "b'ar": {"baz": True},73        'b"ar': {"baz": None},74    },75    "lst": ["A", "B"],76}  # type: Dict[str, Any]77interpolate_parameters = [78    ("$(foo)", interpolate_input["foo"]),79    ("$(foo.bar)", interpolate_input["foo"]["bar"]),80    ("$(foo['bar'])", interpolate_input["foo"]["bar"]),81    ('$(foo["bar"])', interpolate_input["foo"]["bar"]),82    ("$(foo.bar.baz)", interpolate_input["foo"]["bar"]["baz"]),83    ("$(foo['bar'].baz)", interpolate_input["foo"]["bar"]["baz"]),84    ("$(foo['bar'][\"baz\"])", interpolate_input["foo"]["bar"]["baz"]),85    ("$(foo.bar['baz'])", interpolate_input["foo"]["bar"]["baz"]),86    ("$(foo['b\\'ar'].baz)", True),87    ('$(foo["b\'ar"].baz)', True),88    ("$(foo['b\\\"ar'].baz)", None),89    ("$(lst[0])", "A"),90    ("$(lst[1])", "B"),91    ("$(lst.length)", 2),92    ("$(lst['length'])", 2),93    ("-$(foo.bar)", """-{"baz": "zab1"}"""),94    ("-$(foo['bar'])", """-{"baz": "zab1"}"""),95    ('-$(foo["bar"])', """-{"baz": "zab1"}"""),96    ("-$(foo.bar.baz)", "-zab1"),97    ("-$(foo['bar'].baz)", "-zab1"),98    ("-$(foo['bar'][\"baz\"])", "-zab1"),99    ("-$(foo.bar['baz'])", "-zab1"),100    ("-$(foo['b ar'].baz)", "-2"),101    ("-$(foo['b\\'ar'].baz)", "-true"),102    ('-$(foo["b\\\'ar"].baz)', "-true"),103    ("-$(foo['b\\\"ar'].baz)", "-null"),104    ("$(foo.bar) $(foo.bar)", """{"baz": "zab1"} {"baz": "zab1"}"""),105    ("$(foo['bar']) $(foo['bar'])", """{"baz": "zab1"} {"baz": "zab1"}"""),106    ('$(foo["bar"]) $(foo["bar"])', """{"baz": "zab1"} {"baz": "zab1"}"""),107    ("$(foo.bar.baz) $(foo.bar.baz)", "zab1 zab1"),108    ("$(foo['bar'].baz) $(foo['bar'].baz)", "zab1 zab1"),109    ("$(foo['bar'][\"baz\"]) $(foo['bar'][\"baz\"])", "zab1 zab1"),110    ("$(foo.bar['baz']) $(foo.bar['baz'])", "zab1 zab1"),111    ("$(foo['b ar'].baz) $(foo['b ar'].baz)", "2 2"),112    ("$(foo['b\\'ar'].baz) $(foo['b\\'ar'].baz)", "true true"),113    ('$(foo["b\\\'ar"].baz) $(foo["b\\\'ar"].baz)', "true true"),114    ("$(foo['b\\\"ar'].baz) $(foo['b\\\"ar'].baz)", "null null"),115]116@pytest.mark.parametrize("pattern,expected", interpolate_parameters)  # type: ignore117def test_expression_interpolate(pattern: str, expected: Any) -> None:118    assert expr.interpolate(pattern, interpolate_input) == expected119interpolate_bad_parameters = [120    ("$(fooz)"),121    ("$(foo.barz)"),122    ("$(foo['barz'])"),123    ('$(foo["barz"])'),124    ("$(foo.bar.bazz)"),125    ("$(foo['bar'].bazz)"),126    ("$(foo['bar'][\"bazz\"])"),127    ("$(foo.bar['bazz'])"),128    ("$(foo['b\\'ar'].bazz)"),129    ('$(foo["b\'ar"].bazz)'),130    ("$(foo['b\\\"ar'].bazz)"),131    ("$(lst[O])"),  # not "0" the number, but the letter O132    ("$(lst[2])"),133    ("$(lst.lengthz)"),134    ("$(lst['lengthz'])"),135    ("-$(foo.barz)"),136    ("-$(foo['barz'])"),137    ('-$(foo["barz"])'),138    ("-$(foo.bar.bazz)"),139    ("-$(foo['bar'].bazz)"),140    ("-$(foo['bar'][\"bazz\"])"),141    ("-$(foo.bar['bazz'])"),142    ("-$(foo['b ar'].bazz)"),143    ("-$(foo['b\\'ar'].bazz)"),144    ('-$(foo["b\\\'ar"].bazz)'),145    ("-$(foo['b\\\"ar'].bazz)"),146]147@pytest.mark.parametrize("pattern", interpolate_bad_parameters)  # type: ignore148def test_expression_interpolate_failures(pattern: str) -> None:149    result = None150    try:151        result = expr.interpolate(pattern, interpolate_input)152    except JavascriptException:153        return154    assert False, 'Should have produced a JavascriptException, got "{}".'.format(result)155@windows_needs_docker  # type: ignore156def test_factory() -> None:157    factory = get_windows_safe_factory()158    echo = factory.make(get_data("tests/echo.cwl"))159    assert echo(inp="foo") == {"out": "foo\n"}160def test_factory_bad_outputs() -> None:161    factory = cwltool.factory.Factory()162    with pytest.raises(ValidationException):163        factory.make(get_data("tests/echo_broken_outputs.cwl"))164def test_factory_default_args() -> None:165    factory = cwltool.factory.Factory()166    assert factory.runtime_context.use_container is True167    assert factory.runtime_context.on_error == "stop"168def test_factory_redefined_args() -> None:169    runtime_context = RuntimeContext()170    runtime_context.use_container = False171    runtime_context.on_error = "continue"172    factory = cwltool.factory.Factory(runtime_context=runtime_context)173    assert factory.runtime_context.use_container is False174    assert factory.runtime_context.on_error == "continue"175def test_factory_partial_scatter() -> None:176    runtime_context = RuntimeContext()177    runtime_context.on_error = "continue"178    factory = cwltool.factory.Factory(runtime_context=runtime_context)179    with pytest.raises(cwltool.factory.WorkflowStatus) as err_info:180        factory.make(get_data("tests/wf/scatterfail.cwl"))()181    err = err_info.value182    assert (183        err.out["out"][0]["checksum"] == "sha1$e5fa44f2b31c1fb553b6021e7360d07d5d91ff5e"184    )185    assert err.out["out"][1] is None186    assert (187        err.out["out"][2]["checksum"] == "sha1$a3db5c13ff90a36963278c6a39e4ee3c22e2a436"188    )189def test_factory_partial_output() -> None:190    runtime_context = RuntimeContext()191    runtime_context.on_error = "continue"192    factory = cwltool.factory.Factory(runtime_context=runtime_context)193    with pytest.raises(cwltool.factory.WorkflowStatus) as err_info:194        factory.make(get_data("tests/wf/wffail.cwl"))()195    err = err_info.value196    assert (197        err.out["out1"]["checksum"] == "sha1$e5fa44f2b31c1fb553b6021e7360d07d5d91ff5e"198    )199    assert err.out["out2"] is None200def test_scandeps() -> None:201    obj = {202        "id": "file:///example/foo.cwl",203        "steps": [204            {205                "id": "file:///example/foo.cwl#step1",206                "inputs": [207                    {208                        "id": "file:///example/foo.cwl#input1",209                        "default": {210                            "class": "File",211                            "location": "file:///example/data.txt",212                        },213                    }214                ],215                "run": {216                    "id": "file:///example/bar.cwl",217                    "inputs": [218                        {219                            "id": "file:///example/bar.cwl#input2",220                            "default": {221                                "class": "Directory",222                                "location": "file:///example/data2",223                                "listing": [224                                    {225                                        "class": "File",226                                        "location": "file:///example/data3.txt",227                                        "secondaryFiles": [228                                            {229                                                "class": "File",230                                                "location": "file:///example/data5.txt",231                                            }232                                        ],233                                    }234                                ],235                            },236                        },237                        {238                            "id": "file:///example/bar.cwl#input3",239                            "default": {240                                "class": "Directory",241                                "listing": [242                                    {243                                        "class": "File",244                                        "location": "file:///example/data4.txt",245                                    }246                                ],247                            },248                        },249                        {250                            "id": "file:///example/bar.cwl#input4",251                            "default": {"class": "File", "contents": "file literal"},252                        },253                    ],254                },255            }256        ],257    }258    def loadref(base: str, p: str) -> Union[CommentedMap, CommentedSeq, str, None]:259        if isinstance(p, dict):260            return p261        raise Exception("test case can't load things")262    scanned_deps = cwltool.process.scandeps(263        cast(str, obj["id"]),264        obj,265        {"$import", "run"},266        {"$include", "$schemas", "location"},267        loadref,268    )269    scanned_deps.sort(key=lambda k: k["basename"])270    expected_deps = [271        {272            "basename": "bar.cwl",273            "nameroot": "bar",274            "class": "File",275            "format": CWL_IANA,276            "nameext": ".cwl",277            "location": "file:///example/bar.cwl",278        },279        {280            "basename": "data.txt",281            "nameroot": "data",282            "class": "File",283            "nameext": ".txt",284            "location": "file:///example/data.txt",285        },286        {287            "basename": "data2",288            "class": "Directory",289            "location": "file:///example/data2",290            "listing": [291                {292                    "basename": "data3.txt",293                    "nameroot": "data3",294                    "class": "File",295                    "nameext": ".txt",296                    "location": "file:///example/data3.txt",297                    "secondaryFiles": [298                        {299                            "class": "File",300                            "basename": "data5.txt",301                            "location": "file:///example/data5.txt",302                            "nameext": ".txt",303                            "nameroot": "data5",304                        }305                    ],306                }307            ],308        },309        {310            "basename": "data4.txt",311            "nameroot": "data4",312            "class": "File",313            "nameext": ".txt",314            "location": "file:///example/data4.txt",315        },316    ]317    assert scanned_deps == expected_deps318    scanned_deps = cwltool.process.scandeps(319        cast(str, obj["id"]), obj, set(("run"),), set(), loadref320    )321    scanned_deps.sort(key=lambda k: k["basename"])322    expected_deps = [323        {324            "basename": "bar.cwl",325            "nameroot": "bar",326            "format": CWL_IANA,327            "class": "File",328            "nameext": ".cwl",329            "location": "file:///example/bar.cwl",330        }331    ]332    assert scanned_deps == expected_deps333def test_trick_scandeps() -> None:334    stream = StringIO()335    main(336        ["--print-deps", "--debug", get_data("tests/wf/trick_defaults.cwl")],337        stdout=stream,338    )339    assert json.loads(stream.getvalue())["secondaryFiles"][0]["location"][:2] != "_:"340def test_input_deps() -> None:341    stream = StringIO()342    main(343        [344            "--print-input-deps",345            get_data("tests/wf/count-lines1-wf.cwl"),346            get_data("tests/wf/wc-job.json"),347        ],348        stdout=stream,349    )350    expected = {351        "class": "File",352        "location": "wc-job.json",353        "format": CWL_IANA,354        "secondaryFiles": [355            {356                "class": "File",357                "location": "whale.txt",358                "basename": "whale.txt",359                "nameroot": "whale",360                "nameext": ".txt",361            }362        ],363    }364    assert json.loads(stream.getvalue()) == expected365def test_input_deps_cmdline_opts() -> None:366    stream = StringIO()367    main(368        [369            "--print-input-deps",370            get_data("tests/wf/count-lines1-wf.cwl"),371            "--file1",372            get_data("tests/wf/whale.txt"),373        ],374        stdout=stream,375    )376    expected = {377        "class": "File",378        "location": "",379        "format": CWL_IANA,380        "secondaryFiles": [381            {382                "class": "File",383                "location": "whale.txt",384                "basename": "whale.txt",385                "nameroot": "whale",386                "nameext": ".txt",387            }388        ],389    }390    assert json.loads(stream.getvalue()) == expected391def test_input_deps_cmdline_opts_relative_deps_cwd() -> None:392    stream = StringIO()393    data_path = get_data("tests/wf/whale.txt")394    main(395        [396            "--print-input-deps",397            "--relative-deps",398            "cwd",399            get_data("tests/wf/count-lines1-wf.cwl"),400            "--file1",401            data_path,402        ],403        stdout=stream,404    )405    goal = {406        "class": "File",407        "location": "",408        "format": CWL_IANA,409        "secondaryFiles": [410            {411                "class": "File",412                "location": str(Path(os.path.relpath(data_path, os.path.curdir))),413                "basename": "whale.txt",414                "nameroot": "whale",415                "nameext": ".txt",416            }417        ],418    }419    assert json.loads(stream.getvalue()) == goal420def test_dedupe() -> None:421    not_deduped = [422        {"class": "File", "location": "file:///example/a"},423        {"class": "File", "location": "file:///example/a"},424        {"class": "File", "location": "file:///example/d"},425        {426            "class": "Directory",427            "location": "file:///example/c",428            "listing": [{"class": "File", "location": "file:///example/d"}],429        },430    ]  # type: List[CWLObjectType]431    expected = [432        {"class": "File", "location": "file:///example/a"},433        {434            "class": "Directory",435            "location": "file:///example/c",436            "listing": [{"class": "File", "location": "file:///example/d"}],437        },438    ]439    assert dedup(not_deduped) == expected440record = {441    "fields": [442        {443            "type": {"items": "string", "type": "array"},444            "name": "file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec/description",445        },446        {447            "type": {"items": "File", "type": "array"},448            "name": "file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec/vrn_file",449        },450    ],451    "type": "record",452    "name": "file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec",453}454source_to_sink = [455    (456        "0",457        {"items": ["string", "null"], "type": "array"},458        {"items": ["string", "null"], "type": "array"},459        True,460    ),461    (462        "1",463        {"items": ["string"], "type": "array"},464        {"items": ["string", "null"], "type": "array"},465        True,466    ),467    (468        "2",469        {"items": ["string", "null"], "type": "array"},470        {"items": ["string"], "type": "array"},471        True,472    ),473    (474        "3",475        {"items": ["string"], "type": "array"},476        {"items": ["int"], "type": "array"},477        False,478    ),479    ("record 0", record, record, True),480    ("record 1", record, {"items": "string", "type": "array"}, False),481]482@pytest.mark.parametrize("name, source, sink, expected", source_to_sink)  # type: ignore483def test_compare_types(484    name: str, source: Dict[str, Any], sink: Dict[str, Any], expected: bool485) -> None:486    assert can_assign_src_to_sink(source, sink) == expected, name487source_to_sink_strict = [488    ("0", ["string", "null"], ["string", "null"], True),489    ("1", ["string"], ["string", "null"], True),490    ("2", ["string", "int"], ["string", "null"], False),491    (492        "3",493        {"items": ["string"], "type": "array"},494        {"items": ["string", "null"], "type": "array"},495        True,496    ),497    (498        "4",499        {"items": ["string", "int"], "type": "array"},500        {"items": ["string", "null"], "type": "array"},501        False,502    ),503]504@pytest.mark.parametrize("name, source, sink, expected", source_to_sink_strict)  # type: ignore505def test_compare_types_strict(506    name: str, source: Dict[str, Any], sink: Dict[str, Any], expected: bool507) -> None:508    assert can_assign_src_to_sink(source, sink, strict=True) == expected, name509typechecks = [510    (["string", "int"], ["string", "int", "null"], None, None, "pass"),511    (["string", "int"], ["string", "null"], None, None, "warning"),512    (["File", "int"], ["string", "null"], None, None, "exception"),513    (514        {"items": ["string", "int"], "type": "array"},515        {"items": ["string", "int", "null"], "type": "array"},516        None,517        None,518        "pass",519    ),520    (521        {"items": ["string", "int"], "type": "array"},522        {"items": ["string", "null"], "type": "array"},523        None,524        None,525        "warning",526    ),527    (528        {"items": ["File", "int"], "type": "array"},529        {"items": ["string", "null"], "type": "array"},530        None,531        None,532        "exception",533    ),534    # check linkMerge when sinktype is not an array535    (["string", "int"], ["string", "int", "null"], "merge_nested", None, "exception"),536    # check linkMerge: merge_nested537    (538        ["string", "int"],539        {"items": ["string", "int", "null"], "type": "array"},540        "merge_nested",541        None,542        "pass",543    ),544    (545        ["string", "int"],546        {"items": ["string", "null"], "type": "array"},547        "merge_nested",548        None,549        "warning",550    ),551    (552        ["File", "int"],553        {"items": ["string", "null"], "type": "array"},554        "merge_nested",555        None,556        "exception",557    ),558    # check linkMerge: merge_nested and sinktype is "Any"559    (["string", "int"], "Any", "merge_nested", None, "pass"),560    # check linkMerge: merge_flattened561    (562        ["string", "int"],563        {"items": ["string", "int", "null"], "type": "array"},564        "merge_flattened",565        None,566        "pass",567    ),568    (569        ["string", "int"],570        {"items": ["string", "null"], "type": "array"},571        "merge_flattened",572        None,573        "warning",574    ),575    (576        ["File", "int"],577        {"items": ["string", "null"], "type": "array"},578        "merge_flattened",579        None,580        "exception",581    ),582    (583        {"items": ["string", "int"], "type": "array"},584        {"items": ["string", "int", "null"], "type": "array"},585        "merge_flattened",586        None,587        "pass",588    ),589    (590        {"items": ["string", "int"], "type": "array"},591        {"items": ["string", "null"], "type": "array"},592        "merge_flattened",593        None,594        "warning",595    ),596    (597        {"items": ["File", "int"], "type": "array"},598        {"items": ["string", "null"], "type": "array"},599        "merge_flattened",600        None,601        "exception",602    ),603    # check linkMerge: merge_flattened and sinktype is "Any"604    (["string", "int"], "Any", "merge_flattened", None, "pass"),605    (606        {"items": ["string", "int"], "type": "array"},607        "Any",608        "merge_flattened",609        None,610        "pass",611    ),612    # check linkMerge: merge_flattened when srctype is a list613    (614        [{"items": "string", "type": "array"}],615        {"items": "string", "type": "array"},616        "merge_flattened",617        None,618        "pass",619    ),620    # check valueFrom621    (622        {"items": ["File", "int"], "type": "array"},623        {"items": ["string", "null"], "type": "array"},624        "merge_flattened",625        "special value",626        "pass",627    ),628]629@pytest.mark.parametrize(  # type: ignore630    "src_type,sink_type,link_merge,value_from,expected_type", typechecks631)632def test_typechecking(633    src_type: Any, sink_type: Any, link_merge: str, value_from: Any, expected_type: str634) -> None:635    assert (636        cwltool.checker.check_types(637            src_type, sink_type, linkMerge=link_merge, valueFrom=value_from638        )639        == expected_type640    )641def test_lifting() -> None:642    # check that lifting the types of the process outputs to the workflow step643    # fails if the step 'out' doesn't match.644    factory = cwltool.factory.Factory()645    with pytest.raises(ValidationException):646        echo = factory.make(get_data("tests/test_bad_outputs_wf.cwl"))647        assert echo(inp="foo") == {"out": "foo\n"}648def test_malformed_outputs() -> None:649    # check that tool validation fails if one of the outputs is not a valid CWL type650    factory = cwltool.factory.Factory()651    with pytest.raises(ValidationException):652        factory.make(get_data("tests/wf/malformed_outputs.cwl"))()653def test_separate_without_prefix() -> None:654    # check that setting 'separate = false' on an inputBinding without prefix fails the workflow655    factory = cwltool.factory.Factory()656    with pytest.raises(WorkflowException):657        factory.make(get_data("tests/wf/separate_without_prefix.cwl"))()658def test_static_checker() -> None:659    # check that the static checker raises exception when a source type660    # mismatches its sink type.661    factory = cwltool.factory.Factory()662    with pytest.raises(ValidationException):663        factory.make(get_data("tests/checker_wf/broken-wf.cwl"))664    with pytest.raises(ValidationException):665        factory.make(get_data("tests/checker_wf/broken-wf2.cwl"))666    with pytest.raises(ValidationException):667        factory.make(get_data("tests/checker_wf/broken-wf3.cwl"))668def test_var_spool_cwl_checker1() -> None:669    """Confirm that references to /var/spool/cwl are caught."""670    stream = StringIO()671    streamhandler = logging.StreamHandler(stream)672    _logger = logging.getLogger("cwltool")673    _logger.addHandler(streamhandler)674    factory = cwltool.factory.Factory()675    try:676        factory.make(get_data("tests/non_portable.cwl"))677        assert (678            "non_portable.cwl:18:4: Non-portable reference to /var/spool/cwl detected"679            in stream.getvalue()680        )681    finally:682        _logger.removeHandler(streamhandler)683def test_var_spool_cwl_checker2() -> None:684    """Confirm that references to /var/spool/cwl are caught."""685    stream = StringIO()686    streamhandler = logging.StreamHandler(stream)687    _logger = logging.getLogger("cwltool")688    _logger.addHandler(streamhandler)689    factory = cwltool.factory.Factory()690    try:691        factory.make(get_data("tests/non_portable2.cwl"))692        assert (693            "non_portable2.cwl:19:4: Non-portable reference to /var/spool/cwl detected"694            in stream.getvalue()695        )696    finally:697        _logger.removeHandler(streamhandler)698def test_var_spool_cwl_checker3() -> None:699    """Confirm that references to /var/spool/cwl are caught."""700    stream = StringIO()701    streamhandler = logging.StreamHandler(stream)702    _logger = logging.getLogger("cwltool")703    _logger.addHandler(streamhandler)704    factory = cwltool.factory.Factory()705    try:706        factory.make(get_data("tests/portable.cwl"))707        assert (708            "Non-portable reference to /var/spool/cwl detected" not in stream.getvalue()709        )710    finally:711        _logger.removeHandler(streamhandler)712def test_print_dot() -> None:713    assert main(["--print-dot", get_data("tests/wf/revsort.cwl")]) == 0714test_factors = [(""), ("--parallel"), ("--debug"), ("--parallel --debug")]715@pytest.mark.parametrize("factor", test_factors)  # type: ignore716def test_js_console_cmd_line_tool(factor: str) -> None:717    for test_file in ("js_output.cwl", "js_output_workflow.cwl"):718        commands = factor.split()719        commands.extend(720            ["--js-console", "--no-container", get_data("tests/wf/" + test_file)]721        )722        error_code, _, stderr = get_main_output(commands)723        assert "[log] Log message" in stderr724        assert "[err] Error message" in stderr725        assert error_code == 0, stderr726@pytest.mark.parametrize("factor", test_factors)  # type: ignore727def test_no_js_console(factor: str) -> None:728    for test_file in ("js_output.cwl", "js_output_workflow.cwl"):729        commands = factor.split()730        commands.extend(["--no-container", get_data("tests/wf/" + test_file)])731        _, _, stderr = get_main_output(commands)732        assert "[log] Log message" not in stderr733        assert "[err] Error message" not in stderr734@needs_docker  # type: ignore735@pytest.mark.parametrize("factor", test_factors)  # type: ignore736def test_cid_file_dir(tmpdir: py.path.local, factor: str) -> None:737    test_file = "cache_test_workflow.cwl"738    cwd = tmpdir.chdir()739    commands = factor.split()740    commands.extend(["--cidfile-dir", str(tmpdir), get_data("tests/wf/" + test_file)])741    error_code, stdout, stderr = get_main_output(commands)742    assert "completed success" in stderr743    assert error_code == 0744    cidfiles_count = sum(1 for _ in tmpdir.visit(fil="*"))745    assert cidfiles_count == 2746    cwd.chdir()747    tmpdir.remove(ignore_errors=True)748@needs_docker  # type: ignore749@pytest.mark.parametrize("factor", test_factors)  # type: ignore750def test_cid_file_dir_arg_is_file_instead_of_dir(751    tmpdir: py.path.local, factor: str752) -> None:753    test_file = "cache_test_workflow.cwl"754    bad_cidfile_dir = str(tmpdir.ensure("cidfile-dir-actually-a-file"))755    commands = factor.split()756    commands.extend(757        ["--cidfile-dir", bad_cidfile_dir, get_data("tests/wf/" + test_file)]758    )759    error_code, _, stderr = get_main_output(commands)760    assert "is not a directory, please check it first" in stderr, stderr761    assert error_code == 2 or error_code == 1, stderr762    tmpdir.remove(ignore_errors=True)763@needs_docker  # type: ignore764@pytest.mark.parametrize("factor", test_factors)  # type: ignore765def test_cid_file_non_existing_dir(tmpdir: py.path.local, factor: str) -> None:766    test_file = "cache_test_workflow.cwl"767    bad_cidfile_dir = str(tmpdir.join("cidfile-dir-badpath"))768    commands = factor.split()769    commands.extend(770        [771            "--record-container-id",772            "--cidfile-dir",773            bad_cidfile_dir,774            get_data("tests/wf/" + test_file),775        ]776    )777    error_code, _, stderr = get_main_output(commands)778    assert "directory doesn't exist, please create it first" in stderr, stderr779    assert error_code == 2 or error_code == 1, stderr780    tmpdir.remove(ignore_errors=True)781@needs_docker  # type: ignore782@pytest.mark.parametrize("factor", test_factors)  # type: ignore783def test_cid_file_w_prefix(tmpdir: py.path.local, factor: str) -> None:784    test_file = "cache_test_workflow.cwl"785    cwd = tmpdir.chdir()786    try:787        commands = factor.split()788        commands.extend(789            [790                "--record-container-id",791                "--cidfile-prefix=pytestcid",792                get_data("tests/wf/" + test_file),793            ]794        )795        error_code, stdout, stderr = get_main_output(commands)796    finally:797        listing = tmpdir.listdir()798        cwd.chdir()799        cidfiles_count = sum(1 for _ in tmpdir.visit(fil="pytestcid*"))800        tmpdir.remove(ignore_errors=True)801    assert "completed success" in stderr802    assert error_code == 0803    assert cidfiles_count == 2, "{}/n{}".format(listing, stderr)804@needs_docker  # type: ignore805@pytest.mark.parametrize("factor", test_factors)  # type: ignore806def test_secondary_files_v1_1(factor: str) -> None:807    test_file = "secondary-files.cwl"808    test_job_file = "secondary-files-job.yml"809    try:810        old_umask = os.umask(stat.S_IWOTH)  # test run with umask 002811        commands = factor.split()812        commands.extend(813            [814                "--enable-dev",815                get_data(os.path.join("tests", test_file)),816                get_data(os.path.join("tests", test_job_file)),817            ]818        )819        error_code, _, stderr = get_main_output(commands)820    finally:821        # 664 in octal, '-rw-rw-r--'822        assert stat.S_IMODE(os.stat("lsout").st_mode) == 436823        os.umask(old_umask)  # revert back to original umask824    assert "completed success" in stderr825    assert error_code == 0826@needs_docker  # type: ignore827@pytest.mark.parametrize("factor", test_factors)  # type: ignore828def test_secondary_files_v1_0(factor: str) -> None:829    test_file = "secondary-files-string-v1.cwl"830    test_job_file = "secondary-files-job.yml"831    try:832        old_umask = os.umask(stat.S_IWOTH)  # test run with umask 002833        commands = factor.split()834        commands.extend(835            [836                get_data(os.path.join("tests", test_file)),837                get_data(os.path.join("tests", test_job_file)),838            ]839        )840        error_code, _, stderr = get_main_output(commands)841    finally:842        # 664 in octal, '-rw-rw-r--'843        assert stat.S_IMODE(os.stat("lsout").st_mode) == 436844        os.umask(old_umask)  # revert back to original umask845    assert "completed success" in stderr846    assert error_code == 0847@needs_docker  # type: ignore848@pytest.mark.parametrize("factor", test_factors)  # type: ignore849def test_wf_without_container(tmpdir: py.path.local, factor: str) -> None:850    test_file = "hello-workflow.cwl"851    with temp_dir("cwltool_cache") as cache_dir:852        commands = factor.split()853        commands.extend(854            [855                "--cachedir",856                cache_dir,857                "--outdir",858                str(tmpdir),859                get_data("tests/wf/" + test_file),860                "--usermessage",861                "hello",862            ]863        )864        error_code, _, stderr = get_main_output(commands)865    assert "completed success" in stderr866    assert error_code == 0867@needs_docker  # type: ignore868@pytest.mark.parametrize("factor", test_factors)  # type: ignore869def test_issue_740_fixed(factor: str) -> None:870    test_file = "cache_test_workflow.cwl"871    with temp_dir("cwltool_cache") as cache_dir:872        commands = factor.split()873        commands.extend(["--cachedir", cache_dir, get_data("tests/wf/" + test_file)])874        error_code, _, stderr = get_main_output(commands)875        assert "completed success" in stderr876        assert error_code == 0877        commands = factor.split()878        commands.extend(["--cachedir", cache_dir, get_data("tests/wf/" + test_file)])879        error_code, _, stderr = get_main_output(commands)880        assert "Output of job will be cached in" not in stderr881        assert error_code == 0, stderr882@needs_docker  # type: ignore883def test_compute_checksum() -> None:884    runtime_context = RuntimeContext()885    runtime_context.compute_checksum = True886    runtime_context.use_container = onWindows()887    factory = cwltool.factory.Factory(runtime_context=runtime_context)888    echo = factory.make(get_data("tests/wf/cat-tool.cwl"))889    output = echo(890        file1={"class": "File", "location": get_data("tests/wf/whale.txt")},891        reverse=False,892    )893    assert isinstance(output, dict)894    result = output["output"]895    assert isinstance(result, dict)896    assert result["checksum"] == "sha1$327fc7aedf4f6b69a42a7c8b808dc5a7aff61376"897@needs_docker  # type: ignore898@pytest.mark.parametrize("factor", test_factors)  # type: ignore899def test_no_compute_chcksum(tmpdir: py.path.local, factor: str) -> None:900    test_file = "tests/wf/wc-tool.cwl"901    job_file = "tests/wf/wc-job.json"902    commands = factor.split()903    commands.extend(904        [905            "--no-compute-checksum",906            "--outdir",907            str(tmpdir),908            get_data(test_file),909            get_data(job_file),910        ]911    )912    error_code, stdout, stderr = get_main_output(commands)913    assert "completed success" in stderr914    assert error_code == 0915    assert "checksum" not in stdout916@pytest.mark.skipif(onWindows(), reason="udocker is Linux/macOS only")  # type: ignore917@pytest.mark.parametrize("factor", test_factors)  # type: ignore918def test_bad_userspace_runtime(factor: str) -> None:919    test_file = "tests/wf/wc-tool.cwl"920    job_file = "tests/wf/wc-job.json"921    commands = factor.split()922    commands.extend(923        [924            "--user-space-docker-cmd=quaquioN",925            "--default-container=debian",926            get_data(test_file),927            get_data(job_file),928        ]929    )930    error_code, stdout, stderr = get_main_output(commands)931    assert "or quaquioN is missing or broken" in stderr, stderr932    assert error_code == 1933@windows_needs_docker  # type: ignore934@pytest.mark.parametrize("factor", test_factors)  # type: ignore935def test_bad_basecommand(factor: str) -> None:936    test_file = "tests/wf/missing-tool.cwl"937    commands = factor.split()938    commands.extend([get_data(test_file)])939    error_code, stdout, stderr = get_main_output(commands)940    assert "'neenooGo' not found" in stderr, stderr941    assert error_code == 1942@needs_docker  # type: ignore943@pytest.mark.parametrize("factor", test_factors)  # type: ignore944def test_bad_basecommand_docker(factor: str) -> None:945    test_file = "tests/wf/missing-tool.cwl"946    commands = factor.split()947    commands.extend(["--debug", "--default-container", "debian", get_data(test_file)])948    error_code, stdout, stderr = get_main_output(commands)949    assert "permanentFail" in stderr, stderr950    assert error_code == 1951@pytest.mark.parametrize("factor", test_factors)  # type: ignore952def test_v1_0_position_expression(factor: str) -> None:953    test_file = "tests/echo-position-expr.cwl"954    test_job = "tests/echo-position-expr-job.yml"955    commands = factor.split()956    commands.extend(["--debug", get_data(test_file), get_data(test_job)])957    error_code, stdout, stderr = get_main_output(commands)958    assert "is not int" in stderr, stderr959    assert error_code == 1960@windows_needs_docker  # type: ignore961@pytest.mark.parametrize("factor", test_factors)  # type: ignore962def test_optional_numeric_output_0(factor: str) -> None:963    test_file = "tests/wf/optional-numerical-output-0.cwl"964    commands = factor.split()965    commands.extend([get_data(test_file)])966    error_code, stdout, stderr = get_main_output(commands)967    assert "completed success" in stderr968    assert error_code == 0969    assert json.loads(stdout)["out"] == 0970@pytest.mark.parametrize("factor", test_factors)  # type: ignore971@windows_needs_docker  # type: ignore972def test_env_filtering(factor: str) -> None:973    test_file = "tests/env.cwl"974    commands = factor.split()975    commands.extend([get_data(test_file)])976    error_code, stdout, stderr = get_main_output(commands)977    process = subprocess.Popen(978        [979            "sh",980            "-c",981            r"""getTrueShellExeName() {982  local trueExe nextTarget 2>/dev/null983  trueExe=$(ps -o comm= $$) || return 1984  [ "${trueExe#-}" = "$trueExe" ] || trueExe=${trueExe#-}985  [ "${trueExe#/}" != "$trueExe" ] || trueExe=$([ -n "$ZSH_VERSION" ] && which -p "$trueExe" || which "$trueExe")986  while nextTarget=$(readlink "$trueExe"); do trueExe=$nextTarget; done987  printf '%s\n' "$(basename "$trueExe")"988} ; getTrueShellExeName""",989        ],990        stdout=subprocess.PIPE,991        stderr=subprocess.PIPE,992        env=None,993    )994    sh_name_b, sh_name_err = process.communicate()995    sh_name = sh_name_b.decode("utf-8").strip()996    assert "completed success" in stderr, (error_code, stdout, stderr)997    assert error_code == 0, (error_code, stdout, stderr)998    if onWindows():999        target = 51000    elif sh_name == "dash":1001        target = 41002    else:  # bash adds "SHLVL" and "_" environment variables1003        target = 61004    result = json.loads(stdout)["env_count"]1005    details = ""1006    if result != target:1007        _, details, _ = get_main_output(["--quiet", get_data("tests/env2.cwl")])1008        print(sh_name)1009        print(sh_name_err)1010        print(details)1011    assert result == target, (error_code, sh_name, sh_name_err, details, stdout, stderr)1012@windows_needs_docker  # type: ignore1013def test_v1_0_arg_empty_prefix_separate_false() -> None:1014    test_file = "tests/arg-empty-prefix-separate-false.cwl"1015    error_code, stdout, stderr = get_main_output(1016        ["--debug", get_data(test_file), "--echo"]1017    )1018    assert "completed success" in stderr...

decode_behavioral_state.py

Source:decode_behavioral_state.py

1# ---2# jupyter:3#   jupytext:4#     formats: py:percent,ipynb5#     text_representation:6#       extension: .py7#       format_name: percent8#       format_version: '1.3'9#       jupytext_version: 1.13.010#   kernelspec:11#     display_name: Python (hidenseek)12#     language: python13#     name: hidenseek14# ---15# %% [markdown]16# Supervised decoding of the tagged behavioral states from the GPFA factors.17#18# Generates data plotted in Supplementary Figure 2.19# %%20import numpy as np21import pandas as pd22import xarray as xr23import matplotlib.pyplot as plt24import seaborn as sns25from tqdm.autonotebook import tqdm26import os27import pickle28from joblib import Parallel, delayed29import ssm30import autograd.numpy.random as npr31# %%32from hidenseek.db_interface import *33connect_to_db(os.path.join(os.getenv('INTERIM_DATA_DIR'), 'database.db'))34# %%35figures_root_dir = os.path.join(os.getenv('ROOT_DIR'), 'reports', 'figures')36# %%37from hidenseek.figure_util.load_results import load_results, load_factors38K = 1139transitions = 'sticky'40bin_length = load_results(K, transitions)41# %%42from hidenseek.figure_util.add_behavioral_states import add_behavioral_states43add_behavioral_states()44# %%45for session in tqdm(Session.select()):46    session.factors = np.row_stack([trial.factors.values.T for trial in session.trials])47    session.hmm_states = np.concatenate([trial.states.values for trial in session.trials])48    session.behavioral_states = np.concatenate([trial.behavioral_states.values for trial in session.trials])49# %% [markdown]50# # Decode from the factors51# %%52from sklearn.naive_bayes import GaussianNB53from sklearn.discriminant_analysis import LinearDiscriminantAnalysis54from sklearn.ensemble import RandomForestClassifier55from sklearn.dummy import DummyClassifier56from sklearn.model_selection import StratifiedKFold, KFold, LeaveOneOut, cross_val_score57# %% [markdown]58# ## CV trial-wise 59#60# Separate training and test data on the level of trials, so that it cannot happen that time points from the same trial are both in the test and training sets.61#62# In each iteration, leave one trial as the test set and train on the rest.63# %%64np.random.seed(123)65# %% tags=[]66cv = KFold(10, shuffle=True, random_state=123)67#cv = LeaveOneOut()68cv_tuples = []69for session in tqdm(Session.select()):70    n_trials = len(session.trials)71    for (fold, (train_ind, test_ind)) in enumerate(cv.split(np.arange(n_trials))):72        train_trials = [trial for (i, trial) in enumerate(session.trials) if i in train_ind]73        test_trials = [trial for (i, trial) in enumerate(session.trials) if i in test_ind]74        train_factors = np.row_stack([trial.factors.values.T for trial in train_trials])75        test_factors = np.row_stack([trial.factors.values.T for trial in test_trials])76        train_behavioral_states = np.concatenate([trial.behavioral_states for trial in train_trials])77        test_behavioral_states = np.concatenate([trial.behavioral_states for trial in test_trials])78        lda = LinearDiscriminantAnalysis().fit(train_factors, train_behavioral_states)79        nb = GaussianNB().fit(train_factors, train_behavioral_states)80        rf = RandomForestClassifier().fit(train_factors, train_behavioral_states)81        dummy_stratified = DummyClassifier(strategy='stratified').fit(train_factors, train_behavioral_states)82        dummy_most_frequent = DummyClassifier(strategy='most_frequent').fit(train_factors, train_behavioral_states)83        cv_tuples.append((session.id, 'lda', len(session.recorded_cells), lda.score(test_factors, test_behavioral_states), fold))84        cv_tuples.append((session.id, 'naive_bayes', len(session.recorded_cells), nb.score(test_factors, test_behavioral_states), fold))85        cv_tuples.append((session.id, 'random_forest', len(session.recorded_cells), rf.score(test_factors, test_behavioral_states), fold))86        cv_tuples.append((session.id, 'dummy_stratified', len(session.recorded_cells), dummy_stratified.score(test_factors, test_behavioral_states), fold))87        cv_tuples.append((session.id, 'dummy_most_frequent', len(session.recorded_cells), dummy_most_frequent.score(test_factors, test_behavioral_states), fold))88# %%89cv_df = pd.DataFrame(cv_tuples, columns = ['session', 'method', 'num_cells', 'score', 'fold'])90# %%91if isinstance(cv, KFold):92    cv_type = f'cv_{cv.n_splits}'93else:94    cv_type = 'leave_one_out'95cv_df.to_csv(f'decode_behavior_cv_results_K_{K}_{cv_type}.csv')96# %% [markdown]97# Plot performance98# %%99sns.catplot(x = 'session', y = 'score', col = 'method', data = cv_df, kind = 'swarm')100sns.catplot(x = 'session', y = 'score', col = 'method', data = cv_df, kind = 'point')...

search_test.py

Source:search_test.py

1#!/usr/bin/env python2# -*- coding:utf-83"""4Author:  Hao Li5Email: howardleeh@gmail.com6Github: https://github.com/SAmmer07Created: 2018/3/88"""9from shutil import rmtree10import os.path as os_path11from pprint import pprint12import fmanager13from database.const import DataClassification, DataFormatCategory, DataValueCategory14from database.db import Database15test_factors = [{'name': 'ZX_IND',16                 'store format': (DataClassification.STRUCTURED, DataValueCategory.CHAR, DataFormatCategory.PANEL),17                 'dtype': None,18                 'rel_path': 'ind.zx'},19                {'name': 'CLOSE',20                 'store format': (DataClassification.STRUCTURED, DataValueCategory.NUMERIC, DataFormatCategory.PANEL),21                 'dtype': 'float64',22                 'rel_path': 'quote.close'},23                {'name': 'ADJ_CLOSE',24                 'store format': (DataClassification.STRUCTURED, DataValueCategory.NUMERIC, DataFormatCategory.PANEL),25                 'dtype': 'float64',26                 'rel_path': 'quote.adj_close'},27                {'name': 'BETA',28                 'store format': (DataClassification.STRUCTURED, DataValueCategory.NUMERIC, DataFormatCategory.PANEL),29                 'dtype': 'float64',30                 'rel_path': 'basicfactor.beta'}]31start_time = '2014-01-01'32end_time = '2018-03-01'33db_path = r'C:\Users\c\Desktop\test\db_test'34if os_path.exists(db_path):35    rmtree(db_path)36    rmtree(r'C:\Users\c\Documents\DatabaseMetadata')37db = Database(db_path)38for factor in test_factors:39    factor_data = fmanager.query(factor['name'], (start_time, end_time))40    result = db.insert(factor_data, factor['rel_path'], factor['store format'], factor['dtype'])41    print(result)42unstruct_data = list(range(1000))43print(db.insert(unstruct_data, 'unstruct_data.test', (DataClassification.UNSTRUCTURED, )))44db.print_collections()45pprint(db.find_data('beta'))46print(db.find_collection('quote'))47db.remove_data('ind.zx', test_factors[0]['store format'])48db.print_collections()49db.move_to('basicfactor.beta', 'quote.beta', test_factors[3]['store format'])50db.print_collections()...

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.