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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
import pathlib
import pytest
from cads_api_client import ApiClient, catalogue, processing
@pytest.fixture
def cat(api_root_url: str, api_anon_key: str) -> catalogue.Catalogue:
client = ApiClient(url=api_root_url, key=api_anon_key, maximum_tries=0)
return client.catalogue_api
def test_from_collection_to_process(cat: catalogue.Catalogue) -> None:
collection_id = "test-adaptor-dummy"
dataset = cat.collection(collection_id)
res = dataset.retrieve_process()
assert isinstance(res, processing.Process)
def test_collection_submit(cat: catalogue.Catalogue) -> None:
collection_id = "test-adaptor-dummy"
dataset = cat.collection(collection_id)
res = dataset.submit()
assert isinstance(res, processing.Remote)
assert isinstance(res.request_uid, str)
assert isinstance(res.status, str)
def test_collection_retrieve_with_dummy_adaptor(
cat: catalogue.Catalogue, tmp_path: pathlib.Path
) -> None:
collection_id = "test-adaptor-dummy"
dataset = cat.collection(collection_id)
target = str(tmp_path / "dummy.txt")
res = dataset.retrieve(
target=target,
)
assert isinstance(res, str)
assert res.endswith(target)
def test_collection_retrieve_with_url_cds_adaptor(
cat: catalogue.Catalogue, tmp_path: pathlib.Path
) -> None:
collection_id = "test-adaptor-url"
dataset = cat.collection(collection_id)
target = str(tmp_path / "wfde1.zip")
res = dataset.retrieve(
variable="grid_point_altitude",
reference_dataset="cru",
version="2.1",
target=target,
)
assert isinstance(res, str)
assert res.endswith(target)
target = str(tmp_path / "wfde2.zip")
res = dataset.retrieve(
variable="grid_point_altitude",
reference_dataset="cru",
version="2.1",
format="zip",
target=target,
)
assert isinstance(res, str)
assert res.endswith(target)
def test_collection_retrieve_with_direct_mars_cds_adaptor(
cat: catalogue.Catalogue, tmp_path: pathlib.Path
) -> None:
collection_id = "test-adaptor-direct-mars"
dataset = cat.collection(collection_id)
target = str(tmp_path / "era5-complete.grib")
request = {
"levelist": "1",
"dataset": "reanalysis",
"time": "00:00:00",
"param": "155",
"date": "1940-01-01",
"expect": "any",
"levtype": "pl",
"number": "all",
"class": "ea",
}
res = dataset.retrieve(target=target, **request)
assert isinstance(res, str)
assert res.endswith(target)
def test_collection_retrieve_with_mars_cds_adaptor(
cat: catalogue.Catalogue, tmp_path: pathlib.Path
) -> None:
collection_id = "test-adaptor-mars"
dataset = cat.collection(collection_id)
target = str(tmp_path / "era5.grib")
res = dataset.retrieve(
product_type="reanalysis",
variable="2m_temperature",
year="2016",
month="01",
day="02",
time="00:00",
target=target,
)
assert isinstance(res, str)
assert res.endswith(target)
@pytest.mark.skip(reason="discontinued adaptor")
def test_collection_retrieve_with_legacy_cds_adaptor(
cat: catalogue.Catalogue, tmp_path: pathlib.Path
) -> None:
collection_id = "test-adaptor-legacy"
dataset = cat.collection(collection_id)
target = str(tmp_path / "era5.grib")
res = dataset.retrieve(
product_type="reanalysis",
variable="temperature",
year="2016",
month="01",
day="02",
time="00:00",
level="1000",
target=target,
retry_options={"maximum_tries": 0},
)
assert isinstance(res, str)
assert res.endswith(target)
|
