diff env/lib/python3.7/site-packages/cwltool/tests/test_fetch.py @ 2:6af9afd405e9 draft

"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400
parents 26e78fe6e8c4
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.7/site-packages/cwltool/tests/test_fetch.py	Thu May 14 14:56:58 2020 -0400
@@ -0,0 +1,80 @@
+import os
+
+from six.moves import urllib
+
+import pytest
+
+import schema_salad.main
+import schema_salad.ref_resolver
+import schema_salad.schema
+
+from cwltool.context import LoadingContext
+from cwltool.load_tool import load_tool
+from cwltool.main import main
+from cwltool.resolver import Path, resolve_local
+from cwltool.utils import onWindows
+from cwltool.workflow import default_make_tool
+
+from .util import get_data, working_directory
+
+def test_fetcher():
+    class TestFetcher(schema_salad.ref_resolver.Fetcher):
+        def __init__(self, a, b):
+            pass
+
+        def fetch_text(self, url):  # type: (unicode) -> unicode
+            if url == "baz:bar/foo.cwl":
+                return """
+cwlVersion: v1.0
+class: CommandLineTool
+baseCommand: echo
+inputs: []
+outputs: []
+"""
+            raise RuntimeError("Not foo.cwl, was %s" % url)
+
+        def check_exists(self, url):  # type: (unicode) -> bool
+            return  url == "baz:bar/foo.cwl"
+
+        def urljoin(self, base, url):
+            urlsp = urllib.parse.urlsplit(url)
+            if urlsp.scheme:
+                return url
+            basesp = urllib.parse.urlsplit(base)
+
+            if basesp.scheme == "keep":
+                return base + "/" + url
+            return urllib.parse.urljoin(base, url)
+
+    def test_resolver(d, a):
+        if a.startswith("baz:bar/"):
+            return a
+        return "baz:bar/" + a
+
+    loadingContext = LoadingContext({"construct_tool_object": default_make_tool,
+                                     "resolver": test_resolver,
+                                     "fetcher_constructor": TestFetcher})
+
+    load_tool("foo.cwl", loadingContext)
+
+    assert main(["--print-pre", "--debug", "foo.cwl"], loadingContext=loadingContext) == 0
+
+root = Path(os.path.join(get_data("")))
+
+path_fragments = [
+    (os.path.join("tests", "echo.cwl"), "/tests/echo.cwl"),
+    (os.path.join("tests", "echo.cwl") + "#main", "/tests/echo.cwl#main"),
+    (str(root / "tests" / "echo.cwl"), "/tests/echo.cwl"),
+    (str(root / "tests" / "echo.cwl") + "#main", "/tests/echo.cwl#main")
+]
+
+def norm(uri):
+    if onWindows():
+        return uri.lower()
+    return uri
+
+@pytest.mark.parametrize('path,expected_path', path_fragments)
+def test_resolve_local(path, expected_path):
+    with working_directory(root):
+        expected = norm(root.as_uri() + expected_path)
+        assert norm(resolve_local(None, path)) == expected