use get_data instead of __file__

Author: mr-c

tests/test_cite_extract.py

@@ -1,15 +1,12 @@
 # SPDX-License-Identifier: Apache-2.0
 """Tests for cwl-cite-extract."""
-from pathlib import Path
-
 import cwl_utils.parser.cwl_v1_0 as parser
 from cwl_utils.cite_extract import traverse_workflow
 
-HERE = Path(__file__).resolve().parent
-TEST_CWL = HERE / "../testdata/md5sum.cwl"
+from .util import get_data
 
 
 def test_traverse_workflow() -> None:
     """Test the citation extraction, simply."""
-    loaded = parser.load_document(str(TEST_CWL.resolve()))
+    loaded = parser.load_document(get_data("testdata/md5sum.cwl"))
     traverse_workflow(loaded)

tests/test_docker_extract.py

@@ -10,14 +10,15 @@
 from cwl_utils.docker_extract import traverse
 from cwl_utils.image_puller import DockerImagePuller, SingularityImagePuller
 
-HERE = Path(__file__).resolve().parent
-TEST_CWL = HERE / "../testdata/md5sum.cwl"
+from .util import get_data
+
+TEST_CWL = get_data("testdata/md5sum.cwl")
 
 
 @mark.skipif(which("docker") is None, reason="docker is not available")
 def test_traverse_workflow() -> None:
     """Test container extraction tool using Docker."""
-    loaded = parser.load_document(str(TEST_CWL.resolve()))
+    loaded = parser.load_document(TEST_CWL)
 
     with TemporaryDirectory() as tmpdir:
         reqs = set(traverse(loaded))
@@ -32,7 +33,7 @@ def test_traverse_workflow() -> None:
 @mark.skipif(which("singularity") is None, reason="singularity is not available")
 def test_traverse_workflow_singularity() -> None:
     """Test container extraction tool using Singularity."""
-    loaded = parser.load_document(str(TEST_CWL.resolve()))
+    loaded = parser.load_document(TEST_CWL)
 
     with TemporaryDirectory() as tmpdir:
         reqs = set(traverse(loaded))

tests/test_etools_to_clt.py

@@ -20,16 +20,14 @@
 from cwl_utils.errors import WorkflowException
 from cwl_utils.expression_refactor import run as expression_refactor
 
-HERE = Path(__file__).resolve().parent
+from .util import get_data
 
 
 def test_v1_0_workflow_top_level_format_expr() -> None:
     """Test for the correct error when converting a format expression in a workflow level input."""
     with raises(WorkflowException, match=r".*format specification.*"):
         result, modified = traverse0(
-            parser.load_document(
-                str(HERE / "../testdata/workflow_input_format_expr.cwl")
-            ),
+            parser.load_document(get_data("testdata/workflow_input_format_expr.cwl")),
             False,
             False,
             False,
@@ -41,7 +39,7 @@ def test_v1_0_workflow_top_level_sf_expr() -> None:
     """Test for the correct error when converting a secondaryFiles expression in a workflow level input."""
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse0(
-            parser.load_document(str(HERE / "../testdata/workflow_input_sf_expr.cwl")),
+            parser.load_document(get_data("testdata/workflow_input_sf_expr.cwl")),
             False,
             False,
             False,
@@ -53,9 +51,7 @@ def test_v1_0_workflow_top_level_sf_expr_array() -> None:
     """Test for the correct error when converting a secondaryFiles expression (array form) in a workflow level input."""
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse0(
-            parser.load_document(
-                str(HERE / "../testdata/workflow_input_sf_expr_array.cwl")
-            ),
+            parser.load_document(get_data("testdata/workflow_input_sf_expr_array.cwl")),
             False,
             False,
             False,
@@ -65,11 +61,10 @@ def test_v1_0_workflow_top_level_sf_expr_array() -> None:
 
 def test_v1_1_workflow_top_level_format_expr() -> None:
     """Test for the correct error when converting a format expression in a workflow level input."""
-    # import ipdb; ipdb.set_trace()
     with raises(WorkflowException, match=r".*format specification.*"):
         result, modified = traverse1(
             parser1.load_document(
-                str(HERE / "../testdata/workflow_input_format_expr_v1_1.cwl")
+                get_data("testdata/workflow_input_format_expr_v1_1.cwl")
             ),
             False,
             False,
@@ -82,9 +77,7 @@ def test_v1_1_workflow_top_level_sf_expr() -> None:
     """Test for the correct error when converting a secondaryFiles expression in a workflow level input."""
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse1(
-            parser1.load_document(
-                str(HERE / "../testdata/workflow_input_sf_expr_v1_1.cwl")
-            ),
+            parser1.load_document(get_data("testdata/workflow_input_sf_expr_v1_1.cwl")),
             False,
             False,
             False,
@@ -97,7 +90,7 @@ def test_v1_1_workflow_top_level_sf_expr_array() -> None:
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse1(
             parser1.load_document(
-                str(HERE / "../testdata/workflow_input_sf_expr_array_v1_1.cwl")
+                get_data("testdata/workflow_input_sf_expr_array_v1_1.cwl")
             ),
             False,
             False,
@@ -111,7 +104,7 @@ def test_v1_2_workflow_top_level_format_expr() -> None:
     with raises(WorkflowException, match=r".*format specification.*"):
         result, modified = traverse2(
             parser2.load_document(
-                str(HERE / "../testdata/workflow_input_format_expr_v1_2.cwl")
+                get_data("testdata/workflow_input_format_expr_v1_2.cwl")
             ),
             False,
             False,
@@ -124,9 +117,7 @@ def test_v1_2_workflow_top_level_sf_expr() -> None:
     """Test for the correct error when converting a secondaryFiles expression in a workflow level input."""
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse2(
-            parser2.load_document(
-                str(HERE / "../testdata/workflow_input_sf_expr_v1_2.cwl")
-            ),
+            parser2.load_document(get_data("testdata/workflow_input_sf_expr_v1_2.cwl")),
             False,
             False,
             False,
@@ -139,7 +130,7 @@ def test_v1_2_workflow_top_level_sf_expr_array() -> None:
     with raises(WorkflowException, match=r".*secondaryFiles.*"):
         result, modified = traverse2(
             parser2.load_document(
-                str(HERE / "../testdata/workflow_input_sf_expr_array_v1_2.cwl")
+                get_data("testdata/workflow_input_sf_expr_array_v1_2.cwl")
             ),
             False,
             False,
@@ -151,7 +142,7 @@ def test_v1_2_workflow_top_level_sf_expr_array() -> None:
 def test_v1_0_step_valuefrom_expr_multisource() -> None:
     """Convert a valueFrom expression that has multiple sources."""
     result, modified = traverse0(
-        parser.load_document(str(HERE / "../testdata/step-valuefrom2-wf_v1_0.cwl")),
+        parser.load_document(get_data("testdata/step-valuefrom2-wf_v1_0.cwl")),
         False,
         False,
         False,
@@ -162,7 +153,7 @@ def test_v1_0_step_valuefrom_expr_multisource() -> None:
 def test_v1_1_step_valuefrom_expr_multisource() -> None:
     """Convert a valueFrom expression that has multiple sources."""
     result, modified = traverse1(
-        parser1.load_document(str(HERE / "../testdata/step-valuefrom2-wf_v1_1.cwl")),
+        parser1.load_document(get_data("testdata/step-valuefrom2-wf_v1_1.cwl")),
         False,
         False,
         False,
@@ -173,7 +164,7 @@ def test_v1_1_step_valuefrom_expr_multisource() -> None:
 def test_v1_2_step_valuefrom_expr_multisource() -> None:
     """Convert a valueFrom expression that has multiple sources."""
     result, modified = traverse2(
-        parser2.load_document(str(HERE / "../testdata/step-valuefrom2-wf_v1_2.cwl")),
+        parser2.load_document(get_data("testdata/step-valuefrom2-wf_v1_2.cwl")),
         False,
         False,
         False,
@@ -184,7 +175,7 @@ def test_v1_2_step_valuefrom_expr_multisource() -> None:
 def test_v1_0_step_valuefrom_expr_sibling_inputs() -> None:
     """Convert a valueFrom expression from a step input that has uninvolved sibling inputs."""
     result, modified = traverse0(
-        parser.load_document(str(HERE / "../testdata/step-valuefrom3-wf_v1_0.cwl")),
+        parser.load_document(get_data("testdata/step-valuefrom3-wf_v1_0.cwl")),
         False,
         False,
         False,
@@ -195,7 +186,7 @@ def test_v1_0_step_valuefrom_expr_sibling_inputs() -> None:
 def test_v1_1_step_valuefrom_expr_sibling_inputs() -> None:
     """Convert a valueFrom expression from a step input that has uninvolved sibling inputs."""
     result, modified = traverse1(
-        parser1.load_document(str(HERE / "../testdata/step-valuefrom3-wf_v1_1.cwl")),
+        parser1.load_document(get_data("testdata/step-valuefrom3-wf_v1_1.cwl")),
         False,
         False,
         False,
@@ -206,7 +197,7 @@ def test_v1_1_step_valuefrom_expr_sibling_inputs() -> None:
 def test_v1_2_step_valuefrom_expr_sibling_inputs() -> None:
     """Convert a valueFrom expression from a step input that has uninvolved sibling inputs."""
     result, modified = traverse2(
-        parser2.load_document(str(HERE / "../testdata/step-valuefrom3-wf_v1_2.cwl")),
+        parser2.load_document(get_data("testdata/step-valuefrom3-wf_v1_2.cwl")),
         False,
         False,
         False,
@@ -217,7 +208,7 @@ def test_v1_2_step_valuefrom_expr_sibling_inputs() -> None:
 def test_v1_2_workflow_output_pickvalue_expr() -> None:
     """Convert a workflow output pickValue expression."""
     result, modified = traverse2(
-        parser2.load_document(str(HERE / "../testdata/cond-wf-003.1.cwl")),
+        parser2.load_document(get_data("testdata/cond-wf-003.1.cwl")),
         False,
         False,
         False,
@@ -227,22 +218,22 @@ def test_v1_2_workflow_output_pickvalue_expr() -> None:
 
 def test_expression_refactor(tmp_path: Path) -> None:
     """Functional test."""
-    input_path = str(HERE / "../testdata/cond-wf-003.1.cwl")
+    input_path = get_data("testdata/cond-wf-003.1.cwl")
     result = expression_refactor([str(tmp_path), input_path])
     assert result == 0
 
 
 def test_expression_refactor_noop_solo(tmp_path: Path) -> None:
     """Functional test."""
-    input_path = str(HERE / "../testdata/dockstore-tool-md5sum.cwl")
+    input_path = get_data("testdata/dockstore-tool-md5sum.cwl")
     result = expression_refactor([str(tmp_path), input_path])
     assert result == 7
 
 
 def test_expression_refactor_noop(tmp_path: Path) -> None:
     """Functional test."""
-    input_path1 = str(HERE / "../testdata/dockstore-tool-md5sum.cwl")
-    input_path2 = str(HERE / "../testdata/echo-tool-packed.cwl")
+    input_path1 = get_data("testdata/dockstore-tool-md5sum.cwl")
+    input_path2 = get_data("testdata/echo-tool-packed.cwl")
     result = expression_refactor([str(tmp_path), input_path1, input_path2])
     assert result == 0
 

tests/test_format.py

@@ -17,6 +17,8 @@
 from cwl_utils.parser import load_document_by_uri
 from cwl_utils.types import CWLObjectType
 
+from .util import get_data
+
 
 def _create_file(format_: Optional[str] = None) -> CWLObjectType:
     obj: CWLObjectType = {
@@ -45,9 +47,8 @@ def _load_format(fetchurl: str) -> Graph:
     return graph
 
 
-HERE = Path(__file__).resolve().parent
-EDAM = _load_format((HERE / "../testdata/EDAM_subset.owl").absolute().as_uri())
-GX = _load_format((HERE / "../testdata/gx_edam.ttl").absolute().as_uri())
+EDAM = _load_format(Path(get_data("testdata/EDAM_subset.owl")).as_uri())
+GX = _load_format(Path(get_data("testdata/gx_edam.ttl")).as_uri())
 
 
 def test_check_format() -> None:
@@ -136,36 +137,36 @@ def test_check_format_no_ontology() -> None:
 
 def test_loading_options_graph_property_v1_0() -> None:
     """Test that RDFLib Graph representations of $schema properties are correctly loaded, CWL v1.0."""
-    uri = Path(HERE / "../testdata/formattest2_v1_0.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/formattest2_v1_0.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert to_isomorphic(cwl_obj.loadingOptions.graph) == to_isomorphic(EDAM)
 
 
 def test_loading_options_graph_property_v1_1() -> None:
     """Test that RDFLib Graph representations of $schema properties are correctly loaded, CWL v1.1."""
-    uri = Path(HERE / "../testdata/formattest2_v1_1.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/formattest2_v1_1.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert to_isomorphic(cwl_obj.loadingOptions.graph) == to_isomorphic(EDAM)
 
 
 def test_loading_options_graph_property_v1_2() -> None:
     """Test that RDFLib Graph representations of $schema properties are correctly loaded, CWL v1.2."""
-    uri = Path(HERE / "../testdata/formattest2.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/formattest2.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert to_isomorphic(cwl_obj.loadingOptions.graph) == to_isomorphic(EDAM)
 
 
 def test_loading_options_missing_graph_v1_0() -> None:
     """Affirm that v1.0 documents without $schema still produce an empty graph property."""
-    uri = Path(HERE / "../testdata/workflow_input_format_expr.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/workflow_input_format_expr.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert to_isomorphic(cwl_obj.loadingOptions.graph) == to_isomorphic(Graph())
 
 
 def test_loading_options_missing_graph_v1_1() -> None:
     """Affirm that v1.1 documents without $schema still produce an empty graph property."""
     uri = (
-        Path(HERE / "../testdata/workflow_input_format_expr_v1_1.cwl")
+        Path(get_data("testdata/workflow_input_format_expr_v1_1.cwl"))
         .resolve()
         .as_uri()
     )
@@ -176,7 +177,7 @@ def test_loading_options_missing_graph_v1_1() -> None:
 def test_loading_options_missing_graph_v1_2() -> None:
     """Affirm that v1.2 documents without $schema still produce an empty graph property."""
     uri = (
-        Path(HERE / "../testdata/workflow_input_format_expr_v1_2.cwl")
+        Path(get_data("testdata/workflow_input_format_expr_v1_2.cwl"))
         .resolve()
         .as_uri()
     )

tests/test_parser.py

@@ -15,10 +15,11 @@
     save,
 )
 
-HERE = Path(__file__).resolve().parent
-TEST_v1_0_CWL = HERE / "../testdata/md5sum.cwl"
+from .util import get_data
+
+TEST_v1_0_CWL = get_data("testdata/md5sum.cwl")
 TEST_v1_0_CWL_REMOTE = "https://raw.githubusercontent.com/common-workflow-language/cwl-utils/main/testdata/md5sum.cwl"
-TEST_v1_2_CWL = HERE / "../testdata/workflow_input_format_expr_v1_2.cwl"
+TEST_v1_2_CWL = get_data("testdata/workflow_input_format_expr_v1_2.cwl")
 yaml = YAML(typ="rt")
 yaml.preserve_quotes = True  # type: ignore[assignment]
 
@@ -93,20 +94,20 @@ def test_shortname() -> None:
 
 def test_get_id_from_graph() -> None:
     """Test loading an explicit id of a CWL document with $graph property."""
-    uri = Path(HERE / "../testdata/echo-tool-packed.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/echo-tool-packed.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri + "#main")
     assert cwl_obj.id == uri + "#main"
 
 
 def test_get_default_id_from_graph() -> None:
     """Test that loading the default id of a CWL document with $graph property returns the `#main` id."""
-    uri = Path(HERE / "../testdata/echo-tool-packed.cwl").resolve().as_uri()
+    uri = Path(get_data("testdata/echo-tool-packed.cwl")).resolve().as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert cwl_obj.id == uri + "#main"
 
 
 def test_get_default_id_from_graph_without_main() -> None:
     """Test that loading the default id of a CWL document with $graph property and no `#main` id throws an error."""
     with raises(GraphTargetMissingException):
-        uri = Path(HERE / "../testdata/js-expr-req-wf.cwl").resolve().as_uri()
+        uri = Path(get_data("testdata/js-expr-req-wf.cwl")).resolve().as_uri()
         load_document_by_uri(uri)