Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/cwltool/tests/test_fetch.py @ 0:d30785e31577 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:18:57 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d30785e31577
1 import os
2
3 from six.moves import urllib
4
5 import pytest
6
7 import schema_salad.main
8 import schema_salad.ref_resolver
9 import schema_salad.schema
10
11 from cwltool.context import LoadingContext
12 from cwltool.load_tool import load_tool
13 from cwltool.main import main
14 from cwltool.resolver import Path, resolve_local
15 from cwltool.utils import onWindows
16 from cwltool.workflow import default_make_tool
17
18 from .util import get_data, working_directory
19
20 def test_fetcher():
21 class TestFetcher(schema_salad.ref_resolver.Fetcher):
22 def __init__(self, a, b):
23 pass
24
25 def fetch_text(self, url): # type: (unicode) -> unicode
26 if url == "baz:bar/foo.cwl":
27 return """
28 cwlVersion: v1.0
29 class: CommandLineTool
30 baseCommand: echo
31 inputs: []
32 outputs: []
33 """
34 raise RuntimeError("Not foo.cwl, was %s" % url)
35
36 def check_exists(self, url): # type: (unicode) -> bool
37 return url == "baz:bar/foo.cwl"
38
39 def urljoin(self, base, url):
40 urlsp = urllib.parse.urlsplit(url)
41 if urlsp.scheme:
42 return url
43 basesp = urllib.parse.urlsplit(base)
44
45 if basesp.scheme == "keep":
46 return base + "/" + url
47 return urllib.parse.urljoin(base, url)
48
49 def test_resolver(d, a):
50 if a.startswith("baz:bar/"):
51 return a
52 return "baz:bar/" + a
53
54 loadingContext = LoadingContext({"construct_tool_object": default_make_tool,
55 "resolver": test_resolver,
56 "fetcher_constructor": TestFetcher})
57
58 load_tool("foo.cwl", loadingContext)
59
60 assert main(["--print-pre", "--debug", "foo.cwl"], loadingContext=loadingContext) == 0
61
62 root = Path(os.path.join(get_data("")))
63
64 path_fragments = [
65 (os.path.join("tests", "echo.cwl"), "/tests/echo.cwl"),
66 (os.path.join("tests", "echo.cwl") + "#main", "/tests/echo.cwl#main"),
67 (str(root / "tests" / "echo.cwl"), "/tests/echo.cwl"),
68 (str(root / "tests" / "echo.cwl") + "#main", "/tests/echo.cwl#main")
69 ]
70
71 def norm(uri):
72 if onWindows():
73 return uri.lower()
74 return uri
75
76 @pytest.mark.parametrize('path,expected_path', path_fragments)
77 def test_resolve_local(path, expected_path):
78 with working_directory(root):
79 expected = norm(root.as_uri() + expected_path)
80 assert norm(resolve_local(None, path)) == expected