tests: add testdata retriever that returns a Path

Author: mr-c

tests/test_extensions.py

@@ -1,17 +1,11 @@
-from pathlib import Path
-
 from cwl_utils.parser import cwl_v1_0, cwl_v1_1, cwl_v1_2, load_document_by_uri
 
-from .util import get_data
+from .util import get_path
 
 
 def test_cuda_requirement_v1_0() -> None:
     """Test that CUDARequirement objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/cuda-requirement_v1_0.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/cuda-requirement_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.CUDARequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -20,11 +14,7 @@ def test_cuda_requirement_v1_0() -> None:
 
 def test_cuda_requirement_v1_1() -> None:
     """Test that CUDARequirement objects are correctly loaded for CWL v1.1."""
-    uri = (
-        Path(get_data("testdata/extensions/cuda-requirement_v1_1.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/cuda-requirement_v1_1.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_1.CUDARequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -33,11 +23,7 @@ def test_cuda_requirement_v1_1() -> None:
 
 def test_cuda_requirement_v1_2() -> None:
     """Test that CUDARequirement objects are correctly loaded for CWL v1.2."""
-    uri = (
-        Path(get_data("testdata/extensions/cuda-requirement_v1_2.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/cuda-requirement_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_2.CUDARequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -46,11 +32,7 @@ def test_cuda_requirement_v1_2() -> None:
 
 def test_load_listing_requirement_v1_0() -> None:
     """Test that LoadListingRequirement objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/load-listing-requirement_v1_0.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/load-listing-requirement_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.LoadListingRequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -59,11 +41,7 @@ def test_load_listing_requirement_v1_0() -> None:
 
 def test_loop_v1_2() -> None:
     """Test that Loop and LoopInput objects are correctly loaded for CWL v1.2."""
-    uri = (
-        Path(get_data("testdata/extensions/single-var-loop_v1_2.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/single-var-loop_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     cwl_step = next(iter(cwl_obj.steps))
     loop_req = next(iter(cwl_step.requirements))
@@ -75,11 +53,7 @@ def test_loop_v1_2() -> None:
 
 def test_inplace_update_requirement_v1_0() -> None:
     """Test that InplaceUpdateRequirement objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/inplace-update-requirement_v1_0.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/inplace-update-requirement_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(
         next(iter(cwl_obj.requirements)), cwl_v1_0.InplaceUpdateRequirement
@@ -90,11 +64,7 @@ def test_inplace_update_requirement_v1_0() -> None:
 
 def test_mpi_requirement_v1_0() -> None:
     """Test that MPIRequirement objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/mpi-requirement_v1_0.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/mpi-requirement_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.MPIRequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -103,11 +73,7 @@ def test_mpi_requirement_v1_0() -> None:
 
 def test_mpi_requirement_v1_1() -> None:
     """Test that MPIRequirement objects are correctly loaded for CWL v1.1."""
-    uri = (
-        Path(get_data("testdata/extensions/mpi-requirement_v1_1.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/mpi-requirement_v1_1.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_1.MPIRequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -116,11 +82,7 @@ def test_mpi_requirement_v1_1() -> None:
 
 def test_mpi_requirement_v1_2() -> None:
     """Test that MPIRequirement objects are correctly loaded for CWL v1.2."""
-    uri = (
-        Path(get_data("testdata/extensions/mpi-requirement_v1_2.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/mpi-requirement_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_2.MPIRequirement)
     cwl_dict = cwl_obj.save(top=True)
@@ -129,9 +91,7 @@ def test_mpi_requirement_v1_2() -> None:
 
 def test_network_access_v1_0() -> None:
     """Test that NetworkAccess objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/network-access_v1_0.cwl")).resolve().as_uri()
-    )
+    uri = get_path("testdata/extensions/network-access_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.NetworkAccess)
     cwl_dict = cwl_obj.save(top=True)
@@ -140,11 +100,7 @@ def test_network_access_v1_0() -> None:
 
 def test_process_generator_v1_0() -> None:
     """Test that ProcessGenerator objects are correctly loaded for CWL v1.0."""
-    uri = (
-        Path(get_data("testdata/extensions/process-generator_v1_0.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/process-generator_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(cwl_obj, cwl_v1_0.ProcessGenerator)
     cwl_dict = cwl_obj.save(top=True)
@@ -153,11 +109,7 @@ def test_process_generator_v1_0() -> None:
 
 def test_process_generator_v1_1() -> None:
     """Test that ProcessGenerator objects are correctly loaded for CWL v1.1."""
-    uri = (
-        Path(get_data("testdata/extensions/process-generator_v1_1.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/process-generator_v1_1.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(cwl_obj, cwl_v1_1.ProcessGenerator)
     cwl_dict = cwl_obj.save(top=True)
@@ -166,11 +118,7 @@ def test_process_generator_v1_1() -> None:
 
 def test_process_generator_v1_2() -> None:
     """Test that ProcessGenerator objects are correctly loaded for CWL v1.2."""
-    uri = (
-        Path(get_data("testdata/extensions/process-generator_v1_2.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/extensions/process-generator_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(cwl_obj, cwl_v1_2.ProcessGenerator)
     cwl_dict = cwl_obj.save(top=True)
@@ -179,7 +127,7 @@ def test_process_generator_v1_2() -> None:
 
 def test_secrets_v1_0() -> None:
     """Test that Secrets objects are correctly loaded for CWL v1.0."""
-    uri = Path(get_data("testdata/extensions/secrets_v1_0.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/secrets_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.Secrets)
     cwl_dict = cwl_obj.save(top=True)
@@ -188,7 +136,7 @@ def test_secrets_v1_0() -> None:
 
 def test_secrets_v1_1() -> None:
     """Test that Secrets objects are correctly loaded for CWL v1.1."""
-    uri = Path(get_data("testdata/extensions/secrets_v1_1.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/secrets_v1_1.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_1.Secrets)
     cwl_dict = cwl_obj.save(top=True)
@@ -197,7 +145,7 @@ def test_secrets_v1_1() -> None:
 
 def test_secrets_v1_2() -> None:
     """Test that Secrets objects are correctly loaded for CWL v1.2."""
-    uri = Path(get_data("testdata/extensions/secrets_v1_2.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/secrets_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_2.Secrets)
     cwl_dict = cwl_obj.save(top=True)
@@ -206,7 +154,7 @@ def test_secrets_v1_2() -> None:
 
 def test_shm_size_v1_0() -> None:
     """Test that ShmSize objects are correctly loaded for CWL v1.0."""
-    uri = Path(get_data("testdata/extensions/shm-size_v1_0.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/shm-size_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.ShmSize)
     cwl_dict = cwl_obj.save(top=True)
@@ -215,7 +163,7 @@ def test_shm_size_v1_0() -> None:
 
 def test_shm_size_v1_1() -> None:
     """Test that ShmSize objects are correctly loaded for CWL v1.1."""
-    uri = Path(get_data("testdata/extensions/shm-size_v1_1.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/shm-size_v1_1.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_1.ShmSize)
     cwl_dict = cwl_obj.save(top=True)
@@ -224,7 +172,7 @@ def test_shm_size_v1_1() -> None:
 
 def test_shm_size_v1_2() -> None:
     """Test that ShmSize objects are correctly loaded for CWL v1.2."""
-    uri = Path(get_data("testdata/extensions/shm-size_v1_2.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/shm-size_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_2.ShmSize)
     cwl_dict = cwl_obj.save(top=True)
@@ -233,7 +181,7 @@ def test_shm_size_v1_2() -> None:
 
 def test_time_limit_v1_0() -> None:
     """Test that TimeLimit objects are correctly loaded for CWL v1.0."""
-    uri = Path(get_data("testdata/extensions/time-limit_v1_0.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/time-limit_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.TimeLimit)
     cwl_dict = cwl_obj.save(top=True)
@@ -242,7 +190,7 @@ def test_time_limit_v1_0() -> None:
 
 def test_work_reuse_v1_0() -> None:
     """Test that WorkReuse objects are correctly loaded for CWL v1.0."""
-    uri = Path(get_data("testdata/extensions/work-reuse_v1_0.cwl")).resolve().as_uri()
+    uri = get_path("testdata/extensions/work-reuse_v1_0.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert isinstance(next(iter(cwl_obj.requirements)), cwl_v1_0.WorkReuse)
     cwl_dict = cwl_obj.save(top=True)

tests/test_format.py

@@ -17,7 +17,7 @@
 from cwl_utils.parser import load_document_by_uri
 from cwl_utils.types import CWLObjectType
 
-from .util import get_data
+from .util import get_path
 
 
 def _create_file(format_: Optional[str] = None) -> CWLObjectType:
@@ -34,8 +34,9 @@ def _create_file(format_: Optional[str] = None) -> CWLObjectType:
     return obj
 
 
-def _load_format(fetchurl: str) -> Graph:
+def _load_format(fetchpath: Path) -> Graph:
     fetcher = DefaultFetcher({}, requests.Session())
+    fetchurl = fetchpath.as_uri()
     content = fetcher.fetch_text(fetchurl)
     graph = Graph()
     for fmt in ["xml", "turtle", "rdfa"]:
@@ -47,8 +48,8 @@ def _load_format(fetchurl: str) -> Graph:
     return graph
 
 
-EDAM = _load_format(Path(get_data("testdata/EDAM_subset.owl")).as_uri())
-GX = _load_format(Path(get_data("testdata/gx_edam.ttl")).as_uri())
+EDAM = _load_format(get_path("testdata/EDAM_subset.owl"))
+GX = _load_format(get_path("testdata/gx_edam.ttl"))
 
 
 def test_check_format() -> None:
@@ -137,55 +138,47 @@ 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(get_data("testdata/formattest2_v1_0.cwl")).resolve().as_uri()
+    uri = get_path("testdata/formattest2_v1_0.cwl").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(get_data("testdata/formattest2_v1_1.cwl")).resolve().as_uri()
+    uri = get_path("testdata/formattest2_v1_1.cwl").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(get_data("testdata/formattest2.cwl")).resolve().as_uri()
+    uri = get_path("testdata/formattest2.cwl").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(get_data("testdata/workflow_input_format_expr.cwl")).resolve().as_uri()
+    uri = get_path("testdata/workflow_input_format_expr.cwl").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(get_data("testdata/workflow_input_format_expr_v1_1.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/workflow_input_format_expr_v1_1.cwl").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_2() -> None:
     """Affirm that v1.2 documents without $schema still produce an empty graph property."""
-    uri = (
-        Path(get_data("testdata/workflow_input_format_expr_v1_2.cwl"))
-        .resolve()
-        .as_uri()
-    )
+    uri = get_path("testdata/workflow_input_format_expr_v1_2.cwl").as_uri()
     cwl_obj = load_document_by_uri(uri)
     assert to_isomorphic(cwl_obj.loadingOptions.graph) == to_isomorphic(Graph())
 
 
 def test_loading_format_without_schema_v1_0() -> None:
     """Test that format fields without accompanying schemas are tolerated, CWL v1.0."""
-    uri = Path(get_data("testdata/revsort-run-1-packed.cwl")).resolve().as_uri()
+    uri = get_path("testdata/revsort-run-1-packed.cwl").as_uri()
     load_document_by_uri(uri)

tests/test_graph_split.py

@@ -7,7 +7,7 @@
 
 from cwl_utils.graph_split import graph_split
 
-from .util import get_data
+from .util import get_path
 
 URI = (
     "https://gist.githubusercontent.com/altairwei/"
@@ -25,5 +25,5 @@ def test_graph_split(tmp_path: Path) -> None:
 
 def test_graph_split_offline(tmp_path: Path) -> None:
     """Confirm that a local provided example produces no exception."""
-    with open(get_data("testdata/js-expr-req-wf.cwl")) as handle:
+    with get_path("testdata/js-expr-req-wf.cwl").open() as handle:
         graph_split(handle, tmp_path, "yaml", "main.cwl", True)