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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
|
import os
import httpx
import pytest
from httpx import Response
from time import sleep
from pyDataverse.api import DataAccessApi, NativeApi, SwordApi
from pyDataverse.auth import ApiTokenAuth
from pyDataverse.exceptions import ApiAuthorizationError
from pyDataverse.exceptions import ApiUrlError
from pyDataverse.models import Dataset
from pyDataverse.utils import read_file
from ..conftest import test_config
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(os.path.dirname(__file__))))
class TestApiConnect(object):
"""Test the NativeApi() class initialization."""
def test_api_connect(self, native_api):
sleep(test_config["wait_time"])
assert isinstance(native_api, NativeApi)
assert not native_api.api_token
assert native_api.api_version == "v1"
assert native_api.base_url == os.getenv("BASE_URL").rstrip("/")
assert native_api.base_url_api_native == "{0}/api/{1}".format(
os.getenv("BASE_URL").rstrip("/"), native_api.api_version
)
def test_api_connect_base_url_wrong(self):
"""Test native_api connection with wrong `base_url`."""
# None
with pytest.raises(ApiUrlError):
NativeApi(None)
class TestApiTokenAndAuthBehavior:
def test_api_token_none_and_auth_none(self):
api = NativeApi("https://demo.dataverse.org")
assert api.api_token is None
assert api.auth is None
def test_api_token_none_and_auth(self):
auth = ApiTokenAuth("mytoken")
api = NativeApi("https://demo.dataverse.org", auth=auth)
assert api.api_token is None
assert api.auth is auth
def test_api_token_and_auth(self):
auth = ApiTokenAuth("mytoken")
# Only one, api_token or auth, should be specified
with pytest.warns(UserWarning):
api = NativeApi(
"https://demo.dataverse.org", api_token="sometoken", auth=auth
)
assert api.api_token is None
assert api.auth is auth
def test_api_token_and_auth_none(self):
api_token = "mytoken"
api = NativeApi("https://demo.dataverse.org", api_token)
assert api.api_token == api_token
assert isinstance(api.auth, ApiTokenAuth)
assert api.auth.api_token == api_token
class TestApiRequests(object):
"""Test the native_api requests."""
dataset_id = None
@classmethod
def setup_class(cls):
"""Create the native_api connection for later use."""
cls.dataverse_id = "test-pyDataverse"
cls.dataset_id = None
def test_get_request(self, native_api):
"""Test successful `.get_request()` request."""
# TODO: test params und auth default
base_url = os.getenv("BASE_URL").rstrip("/")
query_str = base_url + "/api/v1/info/server"
resp = native_api.get_request(query_str)
sleep(test_config["wait_time"])
assert isinstance(resp, Response)
def test_get_dataverse(self, native_api):
"""Test successful `.get_dataverse()` request`."""
resp = native_api.get_dataverse(":root")
sleep(test_config["wait_time"])
assert isinstance(resp, Response)
if not os.environ.get("TRAVIS"):
class TestApiToken(object):
"""Test user rights."""
def test_token_missing(self):
BASE_URL = os.getenv("BASE_URL").rstrip("/")
api = NativeApi(BASE_URL)
resp = api.get_info_version()
assert resp.json()["data"]["version"] == os.getenv("DV_VERSION")
# assert resp.json()["data"]["build"] == "267-a91d370"
with pytest.raises(ApiAuthorizationError):
ds = Dataset()
ds.from_json(
read_file(
os.path.join(
BASE_DIR, "tests/data/dataset_upload_min_default.json"
)
)
)
api.create_dataset(":root", ds.json())
def test_token_empty_string(self):
BASE_URL = os.getenv("BASE_URL").rstrip("/")
api = NativeApi(BASE_URL, "")
resp = api.get_info_version()
assert resp.json()["data"]["version"] == os.getenv("DV_VERSION")
# assert resp.json()["data"]["build"] == "267-a91d370"
with pytest.raises(ApiAuthorizationError):
ds = Dataset()
ds.from_json(
read_file(
os.path.join(
BASE_DIR, "tests/data/dataset_upload_min_default.json"
)
)
)
api.create_dataset(":root", ds.json())
# def test_token_no_rights(self):
# BASE_URL = os.getenv("BASE_URL")
# API_TOKEN = os.getenv("API_TOKEN_NO_RIGHTS")
# api = NativeApi(BASE_URL, API_TOKEN)
# resp = api.get_info_version()
# assert resp.json()["data"]["version"] == os.getenv("DV_VERSION")
# assert resp.json()["data"]["build"] == "267-a91d370"
# with pytest.raises(ApiAuthorizationError):
# ds = Dataset()
# ds.from_json(
# read_file(
# os.path.join(
# BASE_DIR, "tests/data/dataset_upload_min_default.json"
# )
# )
# )
# api.create_dataset(":root", ds.json())
def test_token_right_create_dataset_rights(self):
BASE_URL = os.getenv("BASE_URL").rstrip("/")
api_su = NativeApi(BASE_URL, os.getenv("API_TOKEN_SUPERUSER"))
# api_nru = NativeApi(BASE_URL, os.getenv("API_TOKEN_TEST_NO_RIGHTS"))
resp = api_su.get_info_version()
assert resp.json()["data"]["version"] == os.getenv("DV_VERSION")
# assert resp.json()["data"]["build"] == "267-a91d370"
# resp = api_nru.get_info_version()
# assert resp.json()["data"]["version"] == os.getenv("DV_VERSION")
# assert resp.json()["data"]["build"] == "267-a91d370"
ds = Dataset()
ds.from_json(
read_file(
os.path.join(BASE_DIR, "tests/data/dataset_upload_min_default.json")
)
)
resp = api_su.create_dataset(":root", ds.json())
pid = resp.json()["data"]["persistentId"]
assert resp.json()["status"] == "OK"
# with pytest.raises(ApiAuthorizationError):
# resp = api_nru.get_dataset(pid)
resp = api_su.delete_dataset(pid)
assert resp.json()["status"] == "OK"
def test_token_should_not_be_exposed_on_error(self):
BASE_URL = os.getenv("BASE_URL")
API_TOKEN = os.getenv("API_TOKEN")
api = DataAccessApi(BASE_URL, API_TOKEN)
result = api.get_datafile("does-not-exist").json()
assert API_TOKEN not in result["requestUrl"]
@pytest.mark.parametrize(
"auth", (True, False, "api-token", ApiTokenAuth("some-token"))
)
def test_using_auth_on_individual_requests_is_deprecated(self, auth):
BASE_URL = os.getenv("BASE_URL")
API_TOKEN = os.getenv("API_TOKEN")
api = DataAccessApi(BASE_URL, auth=ApiTokenAuth(API_TOKEN))
with pytest.warns(DeprecationWarning):
api.get_datafile("does-not-exist", auth=auth)
@pytest.mark.parametrize(
"auth", (True, False, "api-token", ApiTokenAuth("some-token"))
)
def test_using_auth_on_individual_requests_is_deprecated_unauthorized(
self, auth
):
BASE_URL = os.getenv("BASE_URL")
no_auth_api = DataAccessApi(BASE_URL)
with pytest.warns(DeprecationWarning):
no_auth_api.get_datafile("does-not-exist", auth=auth)
def test_sword_api_requires_http_basic_auth(self):
BASE_URL = os.getenv("BASE_URL")
API_TOKEN = os.getenv("API_TOKEN")
api = SwordApi(BASE_URL, api_token=API_TOKEN)
assert isinstance(api.auth, httpx.BasicAuth)
def test_sword_api_can_authenticate(self):
BASE_URL = os.getenv("BASE_URL")
API_TOKEN = os.getenv("API_TOKEN")
api = SwordApi(BASE_URL, api_token=API_TOKEN)
response = api.get_service_document()
assert response.status_code == 200
def test_sword_api_cannot_authenticate_without_token(self):
BASE_URL = os.getenv("BASE_URL")
api = SwordApi(BASE_URL)
with pytest.raises(ApiAuthorizationError):
api.get_service_document()
|
