How to use test_factors method in tox

Best Python code snippet using tox_python

test_examples.py

Source:test_examples.py Github

copy

Full Screen

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

Full Screen

Full Screen

decode_behavioral_state.py

Source:decode_behavioral_state.py Github

copy

Full Screen

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

Full Screen

Full Screen

search_test.py

Source:search_test.py Github

copy

Full Screen

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

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tox automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful