Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/cwltool/tests/test_fetch.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import os | |
2 from pathlib import Path | |
3 from typing import Any, List, Optional | |
4 from urllib.parse import urljoin, urlsplit | |
5 | |
6 import pytest | |
7 import requests | |
8 from schema_salad.fetcher import Fetcher | |
9 from schema_salad.utils import CacheType | |
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 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 | |
21 class CWLTestFetcher(Fetcher): | |
22 def __init__( | |
23 self, | |
24 cache: CacheType, | |
25 session: Optional[requests.sessions.Session], | |
26 ) -> None: | |
27 """Create a Fetcher that provides a fixed result for testing purposes.""" | |
28 | |
29 def fetch_text(self, url: str, content_types: Optional[List[str]] = None) -> str: | |
30 if url == "baz:bar/foo.cwl": | |
31 return """ | |
32 cwlVersion: v1.0 | |
33 class: CommandLineTool | |
34 baseCommand: echo | |
35 inputs: [] | |
36 outputs: [] | |
37 """ | |
38 raise RuntimeError("Not foo.cwl, was %s" % url) | |
39 | |
40 def check_exists(self, url): # type: (str) -> bool | |
41 return url == "baz:bar/foo.cwl" | |
42 | |
43 def urljoin(self, base: str, url: str) -> str: | |
44 urlsp = urlsplit(url) | |
45 if urlsp.scheme: | |
46 return url | |
47 basesp = urlsplit(base) | |
48 | |
49 if basesp.scheme == "keep": | |
50 return base + "/" + url | |
51 return urljoin(base, url) | |
52 | |
53 | |
54 def test_fetcher() -> None: | |
55 def test_resolver(d: Any, a: str) -> str: | |
56 if a.startswith("baz:bar/"): | |
57 return a | |
58 return "baz:bar/" + a | |
59 | |
60 loadingContext = LoadingContext( | |
61 { | |
62 "construct_tool_object": default_make_tool, | |
63 "resolver": test_resolver, | |
64 "fetcher_constructor": CWLTestFetcher, | |
65 } | |
66 ) | |
67 | |
68 load_tool("foo.cwl", loadingContext) | |
69 | |
70 assert ( | |
71 main(["--print-pre", "--debug", "foo.cwl"], loadingContext=loadingContext) == 0 | |
72 ) | |
73 | |
74 | |
75 root = Path(os.path.join(get_data(""))) | |
76 | |
77 path_fragments = [ | |
78 (os.path.join("tests", "echo.cwl"), "/tests/echo.cwl"), | |
79 (os.path.join("tests", "echo.cwl") + "#main", "/tests/echo.cwl#main"), | |
80 (str(root / "tests" / "echo.cwl"), "/tests/echo.cwl"), | |
81 (str(root / "tests" / "echo.cwl") + "#main", "/tests/echo.cwl#main"), | |
82 ] | |
83 | |
84 | |
85 def norm(uri: str) -> str: | |
86 if onWindows(): | |
87 return uri.lower() | |
88 return uri | |
89 | |
90 | |
91 @pytest.mark.parametrize("path,expected_path", path_fragments) | |
92 def test_resolve_local(path: str, expected_path: str) -> None: | |
93 with working_directory(root): | |
94 expected = norm(root.as_uri() + expected_path) | |
95 resolved = resolve_local(None, path) | |
96 assert resolved | |
97 assert norm(resolved) == expected |