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
|
from __future__ import annotations
import contextlib
import datetime
import os
import pathlib
from typing import Any
import pytest
import requests
from urllib3.exceptions import InsecureRequestWarning
from cads_api_client import ApiClient
does_not_raise = contextlib.nullcontext
@pytest.fixture
def api_anon_client(api_root_url: str, api_anon_key: str) -> ApiClient:
return ApiClient(url=api_root_url, key=api_anon_key, maximum_tries=0)
def test_accept_licence() -> None:
client = ApiClient(maximum_tries=0)
licence = client.licences["licences"][0]
licence_id = licence["id"]
licence_revision = licence["revision"]
expected = {"id": licence_id, "revision": licence_revision}
actual = client.accept_licence(licence_id, licence_revision)
assert expected == actual
assert any(
licence["id"] == licence_id and licence["revision"] == licence_revision
for licence in client.accepted_licences["licences"]
)
def test_delete_request(api_anon_client: ApiClient) -> None:
remote = api_anon_client.submit(
"test-adaptor-dummy", _timestamp=datetime.datetime.now().isoformat()
)
reply = remote.delete()
assert reply["status"] == "dismissed"
with pytest.raises(requests.exceptions.HTTPError):
remote.status
def test_check_authentication(api_root_url: str, api_anon_client: ApiClient) -> None:
assert api_anon_client.check_authentication() == {
"id": -1,
"role": "anonymous",
"sub": "anonymous",
}
bad_client = ApiClient(key="foo", url=api_root_url)
with pytest.raises(requests.exceptions.HTTPError, match="401 Client Error"):
bad_client.check_authentication()
def test_download_result(api_anon_client: ApiClient, tmp_path: pathlib.Path) -> None:
remote = api_anon_client.submit("test-adaptor-dummy")
target = str(tmp_path / "test.grib")
result = api_anon_client.download_result(remote.request_uid, target)
assert result == target
assert os.path.exists(result)
def test_get_remote(api_anon_client: ApiClient, tmp_path: pathlib.Path) -> None:
request_uid = api_anon_client.submit("test-adaptor-dummy").request_uid
result = api_anon_client.get_remote(request_uid)
assert result.request_uid == request_uid
def test_api_client_verify(
api_root_url: str,
api_anon_key: str,
tmp_path: pathlib.Path,
) -> None:
insecure_client = ApiClient(
url=api_root_url, key=api_anon_key, verify=False, maximum_tries=0
)
with pytest.warns(InsecureRequestWarning):
insecure_client.retrieve(
"test-adaptor-dummy", target=str(tmp_path / "test.grib")
)
def test_api_client_timeout(
api_root_url: str,
api_anon_key: str,
tmp_path: pathlib.Path,
) -> None:
client = ApiClient(url=api_root_url, key=api_anon_key, timeout=0)
with pytest.raises(ValueError, match="timeout"):
client.retrieve("test-adaptor-dummy", target=str(tmp_path / "test.grib"))
@pytest.mark.parametrize("progress", [True, False])
def test_api_client_progress(
api_root_url: str,
api_anon_key: str,
tmp_path: pathlib.Path,
progress: bool,
capsys: pytest.CaptureFixture[str],
) -> None:
with capsys.disabled():
client = ApiClient(
url=api_root_url, key=api_anon_key, progress=progress, maximum_tries=0
)
submitted = client.submit("test-adaptor-dummy")
submitted.download(target=str(tmp_path / "test.grib"))
captured = capsys.readouterr()
assert captured.err if progress else not captured.err
@pytest.mark.parametrize(
"cleanup,raises",
[
(True, pytest.raises(requests.exceptions.HTTPError, match="404 Client Error")),
(False, does_not_raise()),
],
)
def test_api_client_cleanup(
api_root_url: str,
api_anon_key: str,
cleanup: bool,
raises: contextlib.nullcontext[Any],
) -> None:
client = ApiClient(
url=api_root_url, key=api_anon_key, cleanup=cleanup, maximum_tries=0
)
remote = client.submit("test-adaptor-dummy")
request_uid = remote.request_uid
del remote
client = ApiClient(url=api_root_url, key=api_anon_key, maximum_tries=0)
with raises:
client.get_request(request_uid)
|
