Mercurial > repos > shellac > guppy_basecaller
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