Mercurial > repos > guerler > springsuite
comparison planemo/lib/python3.7/site-packages/schema_salad/tests/test_fetch.py @ 1:56ad4e20f292 draft
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author | guerler |
---|---|
date | Fri, 31 Jul 2020 00:32:28 -0400 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
0:d30785e31577 | 1:56ad4e20f292 |
---|---|
1 from __future__ import absolute_import, print_function | |
2 | |
3 import os | |
4 from typing import Text | |
5 | |
6 import pytest | |
7 from six.moves import urllib | |
8 | |
9 import schema_salad.main | |
10 import schema_salad.ref_resolver | |
11 import schema_salad.schema | |
12 | |
13 | |
14 def test_fetcher(): | |
15 class TestFetcher(schema_salad.ref_resolver.Fetcher): | |
16 def __init__(self, a, b): | |
17 pass | |
18 | |
19 def fetch_text(self, url): # type: (Text) -> Text | |
20 if url == "keep:abc+123/foo.txt": | |
21 return "hello: keepfoo" | |
22 if url.endswith("foo.txt"): | |
23 return "hello: foo" | |
24 else: | |
25 raise RuntimeError("Not foo.txt") | |
26 | |
27 def check_exists(self, url): # type: (Text) -> bool | |
28 if url.endswith("foo.txt"): | |
29 return True | |
30 else: | |
31 return False | |
32 | |
33 def urljoin(self, base, url): | |
34 urlsp = urllib.parse.urlsplit(url) | |
35 if urlsp.scheme: | |
36 return url | |
37 basesp = urllib.parse.urlsplit(base) | |
38 | |
39 if basesp.scheme == "keep": | |
40 return base + "/" + url | |
41 return urllib.parse.urljoin(base, url) | |
42 | |
43 loader = schema_salad.ref_resolver.Loader({}, fetcher_constructor=TestFetcher) | |
44 assert {"hello": "foo"} == loader.resolve_ref("foo.txt")[0] | |
45 assert {"hello": "keepfoo"} == loader.resolve_ref( | |
46 "foo.txt", base_url="keep:abc+123" | |
47 )[0] | |
48 assert loader.check_exists("foo.txt") | |
49 | |
50 with pytest.raises(RuntimeError): | |
51 loader.resolve_ref("bar.txt") | |
52 assert not loader.check_exists("bar.txt") | |
53 | |
54 | |
55 def test_cache(): | |
56 loader = schema_salad.ref_resolver.Loader({}) | |
57 foo = os.path.join(os.getcwd(), "foo.txt") | |
58 foo = schema_salad.ref_resolver.file_uri(foo) | |
59 loader.cache.update({foo: "hello: foo"}) | |
60 print(loader.cache) | |
61 assert {"hello": "foo"} == loader.resolve_ref("foo.txt")[0] | |
62 assert loader.check_exists(foo) |