|
1 |
| -from __future__ import absolute_import |
2 |
| - |
3 | 1 | import os
|
4 |
| -import unittest |
| 2 | + |
| 3 | +from six.moves import urllib |
| 4 | + |
| 5 | +import pytest |
5 | 6 |
|
6 | 7 | import schema_salad.main
|
7 | 8 | import schema_salad.ref_resolver
|
8 | 9 | import schema_salad.schema
|
9 |
| -from six.moves import urllib |
10 | 10 |
|
11 | 11 | from cwltool.context import LoadingContext
|
12 | 12 | from cwltool.load_tool import load_tool
|
|
15 | 15 | from cwltool.utils import onWindows
|
16 | 16 | from cwltool.workflow import default_make_tool
|
17 | 17 |
|
18 |
| -from .util import get_data |
| 18 | +from .util import get_data, working_directory |
19 | 19 |
|
| 20 | +def test_fetcher(): |
| 21 | + class TestFetcher(schema_salad.ref_resolver.Fetcher): |
| 22 | + def __init__(self, a, b): |
| 23 | + pass |
20 | 24 |
|
21 |
| -class FetcherTest(unittest.TestCase): |
22 |
| - """Test using custom schema_salad.ref_resolver.Fetcher.""" |
23 |
| - def test_fetcher(self): |
24 |
| - class TestFetcher(schema_salad.ref_resolver.Fetcher): |
25 |
| - def __init__(self, a, b): |
26 |
| - pass |
27 |
| - |
28 |
| - def fetch_text(self, url): # type: (unicode) -> unicode |
29 |
| - if url == "baz:bar/foo.cwl": |
30 |
| - return """ |
| 25 | + def fetch_text(self, url): # type: (unicode) -> unicode |
| 26 | + if url == "baz:bar/foo.cwl": |
| 27 | + return """ |
31 | 28 | cwlVersion: v1.0
|
32 | 29 | class: CommandLineTool
|
33 | 30 | baseCommand: echo
|
34 | 31 | inputs: []
|
35 | 32 | outputs: []
|
36 | 33 | """
|
37 |
| - else: |
38 |
| - raise RuntimeError("Not foo.cwl, was %s" % url) |
39 |
| - |
40 |
| - def check_exists(self, url): # type: (unicode) -> bool |
41 |
| - if url == "baz:bar/foo.cwl": |
42 |
| - return True |
43 |
| - else: |
44 |
| - return False |
45 |
| - |
46 |
| - def urljoin(self, base, url): |
47 |
| - urlsp = urllib.parse.urlsplit(url) |
48 |
| - if urlsp.scheme: |
49 |
| - return url |
50 |
| - basesp = urllib.parse.urlsplit(base) |
51 |
| - |
52 |
| - if basesp.scheme == "keep": |
53 |
| - return base + "/" + url |
54 |
| - return urllib.parse.urljoin(base, url) |
55 |
| - |
56 |
| - def test_resolver(d, a): |
57 |
| - if a.startswith("baz:bar/"): |
58 |
| - return a |
59 |
| - else: |
60 |
| - return "baz:bar/" + a |
61 |
| - |
62 |
| - loadingContext = LoadingContext({"construct_tool_object": default_make_tool, |
63 |
| - "resolver": test_resolver, |
64 |
| - "fetcher_constructor": TestFetcher}) |
65 |
| - |
66 |
| - load_tool("foo.cwl", loadingContext) |
67 |
| - |
68 |
| - self.assertEquals(0, main(["--print-pre", "--debug", "foo.cwl"], loadingContext=loadingContext)) |
69 |
| - |
70 |
| - |
71 |
| -class ResolverTest(unittest.TestCase): |
72 |
| - def test_resolve_local(self): |
73 |
| - origpath = os.getcwd() |
74 |
| - os.chdir(os.path.join(get_data(""))) |
75 |
| - |
76 |
| - def norm(uri): |
77 |
| - if onWindows(): |
78 |
| - return uri.lower() |
79 |
| - else: |
80 |
| - return uri |
81 |
| - try: |
82 |
| - root = Path.cwd() |
83 |
| - rooturi = root.as_uri() |
84 |
| - self.assertEqual(norm(rooturi+"/tests/echo.cwl"), |
85 |
| - norm(resolve_local(None, os.path.join("tests", |
86 |
| - "echo.cwl")))) |
87 |
| - self.assertEqual(norm(rooturi+"/tests/echo.cwl#main"), |
88 |
| - norm(resolve_local(None, os.path.join("tests", |
89 |
| - "echo.cwl")+"#main"))) |
90 |
| - self.assertEqual(norm(rooturi+"/tests/echo.cwl"), |
91 |
| - norm(resolve_local(None, str(root / "tests" / |
92 |
| - "echo.cwl")))) |
93 |
| - self.assertEqual(norm(rooturi+"/tests/echo.cwl#main"), |
94 |
| - norm(resolve_local(None, str(root / "tests" / |
95 |
| - "echo.cwl")+"#main"))) |
96 |
| - finally: |
97 |
| - os.chdir(origpath) |
| 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 | +@pytest.mark.parametrize('path,expected_path', path_fragments) |
| 72 | +def test_resolve_local(path, expected_path): |
| 73 | + def norm(uri): |
| 74 | + if onWindows(): |
| 75 | + return uri.lower() |
| 76 | + return uri |
| 77 | + |
| 78 | + with working_directory(root): |
| 79 | + expected = norm(root.as_uri() + expected_path) |
| 80 | + assert norm(resolve_local(None, path)) == expected |
0 commit comments