1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
# (C) Copyright 2021 ECMWF.
#
# This software is licensed under the terms of the Apache Licence Version 2.0
# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
# In applying this licence, ECMWF does not waive the privileges and immunities
# granted to it by virtue of its status as an intergovernmental organisation
# nor does it submit to any jurisdiction.
#
import logging
import os
from multiurl import download
# NOTE: we just test if the auth object is properly called with the
# requests when using download()
class Auth:
def __init__(self):
from collections import defaultdict
self.calls = defaultdict(set)
def __call__(self, r):
method = r.method.lower()
self.calls[method].add(r.url)
return r
def test_auth_single():
auth = Auth()
url = "http://get.ecmwf.int/test-data/metview/gallery/temp.bufr"
download(url=url, target="out.data", auth=auth)
assert auth.calls["head"] == set([url])
assert auth.calls["get"] == set([url])
def test_auth_single_fake_headers():
auth = Auth()
url = "http://get.ecmwf.int/test-data/metview/gallery/temp.bufr"
download(url=url, target="out.data", auth=auth, fake_headers={})
assert auth.calls["head"] == set()
assert auth.calls["get"] == set([url])
def test_auth_single_parts():
auth = Auth()
url = "http://get.ecmwf.int/test-data/metview/gallery/temp.bufr"
download(url=url, target="out.data", parts=((0, 4),), auth=auth)
assert auth.calls["head"] == set([url])
assert auth.calls["get"] == set([url])
assert os.path.getsize("out.data") == 4
with open("out.data", "rb") as f:
assert f.read() == b"BUFR"
def test_auth_single_multi_parts():
auth = Auth()
url = "http://get.ecmwf.int/test-data/metview/gallery/temp.bufr"
download(url=url, target="out.data", parts=((0, 4), (20, 4)), auth=auth)
assert auth.calls["head"] == set([url])
assert auth.calls["get"] == set([url])
assert os.path.getsize("out.data") == 8
with open("out.data", "rb") as f:
assert f.read(4) == b"BUFR"
def test_auth_multi():
auth = Auth()
urls = [
"http://get.ecmwf.int/test-data/earthkit-data/examples/test.grib",
"http://get.ecmwf.int/test-data/earthkit-data/examples/test6.grib",
]
download(url=urls, target="out.data", auth=auth)
assert auth.calls["head"] == set(urls)
assert auth.calls["get"] == set(urls)
assert os.path.getsize("out.data") == 2492
with open("out.data", "rb") as f:
assert f.read(4) == b"GRIB"
def test_auth_multi_parts():
auth = Auth()
urls = [
"http://get.ecmwf.int/test-data/earthkit-data/examples/test.grib",
"http://get.ecmwf.int/test-data/earthkit-data/examples/test6.grib",
]
download(url=urls, target="out.data", parts=((0, 4),), auth=auth)
assert auth.calls["head"] == set(urls)
assert auth.calls["get"] == set(urls)
assert os.path.getsize("out.data") == 8
with open("out.data", "rb") as f:
assert f.read(4) == b"GRIB"
assert f.read(4) == b"GRIB"
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
# test_order()
|
