/
test_assets_api.py
282 lines (248 loc) · 10.9 KB
/
test_assets_api.py
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
import json
from flask import url_for
import pytest
from flexmeasures.data.models.generic_assets import GenericAsset
from flexmeasures.data.services.users import find_user_by_email
from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
from flexmeasures.api.v3_0.tests.utils import get_asset_post_data
@pytest.mark.parametrize("use_auth", [False, True])
def test_get_assets_badauth(client, setup_api_test_data, use_auth):
"""
Attempt to get assets with wrong or missing auth.
"""
# the case without auth: authentication will fail
headers = {"content-type": "application/json"}
query = {}
if use_auth:
# in this case, we successfully authenticate, but fail authorization
headers["Authorization"] = get_auth_token(
client, "test_dummy_user_3@seita.nl", "testtest"
)
test_prosumer = find_user_by_email("test_prosumer_user@seita.nl")
query = {"account_id": test_prosumer.account.id}
get_assets_response = client.get(
url_for("AssetAPI:index"), query_string=query, headers=headers
)
print("Server responded with:\n%s" % get_assets_response.json)
if use_auth:
assert get_assets_response.status_code == 403
else:
assert get_assets_response.status_code == 401
def test_get_asset_nonaccount_access(client, setup_api_test_data):
"""Without being on the same account, test correct responses when accessing one asset."""
with UserContext("test_prosumer_user@seita.nl") as prosumer1:
prosumer1_assets = prosumer1.account.generic_assets
with UserContext("test_supplier_user_4@seita.nl") as supplieruser4:
supplieruser4_assets = supplieruser4.account.generic_assets
headers = {
"content-type": "application/json",
"Authorization": get_auth_token(
client, "test_supplier_user_4@seita.nl", "testtest"
),
}
# okay to look at assets in own account
asset_response = client.get(
url_for("AssetAPI:fetch_one", id=supplieruser4_assets[0].id),
headers=headers,
follow_redirects=True,
)
assert asset_response.status_code == 200
# not okay to see assets owned by other accounts
asset_response = client.get(
url_for("AssetAPI:fetch_one", id=prosumer1_assets[0].id),
headers=headers,
follow_redirects=True,
)
assert asset_response.status_code == 403
# proper 404 for non-existing asset
asset_response = client.get(
url_for("AssetAPI:fetch_one", id=8171766575),
headers=headers,
follow_redirects=True,
)
assert asset_response.status_code == 404
assert "not found" in asset_response.json["message"]
@pytest.mark.parametrize("account_name, num_assets", [("Prosumer", 1), ("Supplier", 2)])
def test_get_assets(
client, setup_api_test_data, setup_accounts, account_name, num_assets
):
"""
Get assets per account.
Our user here is admin, so is allowed to see all assets.
"""
auth_token = get_auth_token(client, "test_admin_user@seita.nl", "testtest")
query = {"account_id": setup_accounts[account_name].id}
get_assets_response = client.get(
url_for("AssetAPI:index"),
query_string=query,
headers={"content-type": "application/json", "Authorization": auth_token},
)
print("Server responded with:\n%s" % get_assets_response.json)
assert get_assets_response.status_code == 200
assert len(get_assets_response.json) == num_assets
if account_name == "Supplier": # one deep dive
turbine = {}
for asset in get_assets_response.json:
if asset["name"] == "Test wind turbine":
turbine = asset
assert turbine
assert turbine["account_id"] == setup_accounts["Supplier"].id
def test_get_public_assets(client, setup_api_test_data, setup_accounts):
auth_token = get_auth_token(client, "test_admin_user@seita.nl", "testtest")
get_assets_response = client.get(
url_for("AssetAPI:public"),
headers={"content-type": "application/json", "Authorization": auth_token},
)
print("Server responded with:\n%s" % get_assets_response.json)
assert len(get_assets_response.json) == 1
assert get_assets_response.json[0]["name"] == "troposphere"
def test_alter_an_asset(client, setup_api_test_data, setup_accounts):
# without being an account-admin, no asset can be created ...
with UserContext("test_prosumer_user@seita.nl") as prosumer1:
auth_token = prosumer1.get_auth_token() # not an account admin
with AccountContext("Test Prosumer Account") as prosumer:
prosumer_asset = prosumer.generic_assets[0]
asset_creation_response = client.post(
url_for("AssetAPI:post"),
headers={"content-type": "application/json", "Authorization": auth_token},
json={},
)
print(f"Creation Response: {asset_creation_response.json}")
assert asset_creation_response.status_code == 403
# ... or deleted ...
asset_delete_response = client.delete(
url_for("AssetAPI:delete", id=prosumer_asset.id),
headers={"content-type": "application/json", "Authorization": auth_token},
json={},
)
print(f"Deletion Response: {asset_delete_response.json}")
assert asset_delete_response.status_code == 403
# ... but editing is allowed.
asset_edit_response = client.patch(
url_for("AssetAPI:patch", id=prosumer_asset.id),
headers={"content-type": "application/json", "Authorization": auth_token},
json={
"latitude": prosumer_asset.latitude
}, # we're not changing values to keep other tests clean here
)
print(f"Editing Response: {asset_edit_response.json}")
assert asset_edit_response.status_code == 200
@pytest.mark.parametrize(
"bad_json_str",
[
None,
"{",
'{"hallo": world}',
],
)
def test_alter_an_asset_with_bad_json_attributes(
client, setup_api_test_data, setup_accounts, bad_json_str
):
"""Check whether updating an asset's attributes with a badly structured JSON fails."""
with UserContext("test_prosumer_user@seita.nl") as prosumer1:
auth_token = prosumer1.get_auth_token()
with AccountContext("Test Prosumer Account") as prosumer:
prosumer_asset = prosumer.generic_assets[0]
asset_edit_response = client.patch(
url_for("AssetAPI:patch", id=prosumer_asset.id),
headers={"content-type": "application/json", "Authorization": auth_token},
json={"attributes": bad_json_str},
)
print(f"Editing Response: {asset_edit_response.json}")
assert asset_edit_response.status_code == 422
def test_alter_an_asset_with_json_attributes(
client, setup_api_test_data, setup_accounts
):
"""Check whether updating an asset's attributes with a properly structured JSON succeeds."""
with UserContext("test_prosumer_user@seita.nl") as prosumer1:
auth_token = prosumer1.get_auth_token()
with AccountContext("Test Prosumer Account") as prosumer:
prosumer_asset = prosumer.generic_assets[0]
asset_edit_response = client.patch(
url_for("AssetAPI:patch", id=prosumer_asset.id),
headers={"content-type": "application/json", "Authorization": auth_token},
json={
"attributes": json.dumps(prosumer_asset.attributes)
}, # we're not changing values to keep other tests clean here
)
print(f"Editing Response: {asset_edit_response.json}")
assert asset_edit_response.status_code == 200
def test_post_an_asset_with_existing_name(client, setup_api_test_data):
"""Catch DB error (Unique key violated) correctly"""
with UserContext("test_admin_user@seita.nl") as admin_user:
auth_token = admin_user.get_auth_token()
with AccountContext("Test Prosumer Account") as prosumer:
prosumer_id = prosumer.id
existing_asset = prosumer.generic_assets[0]
post_data = get_asset_post_data()
post_data["name"] = existing_asset.name
post_data["account_id"] = prosumer_id
asset_creation_response = client.post(
url_for("AssetAPI:post"),
json=post_data,
headers={"content-type": "application/json", "Authorization": auth_token},
)
print(f"Creation Response: {asset_creation_response.json}")
assert asset_creation_response.status_code == 422
assert (
"already exists" in asset_creation_response.json["message"]["json"]["name"][0]
)
def test_post_an_asset_with_nonexisting_field(client, setup_api_test_data):
"""Posting a field that is unexpected leads to a 422"""
with UserContext("test_admin_user@seita.nl") as prosumer:
auth_token = prosumer.get_auth_token()
post_data = get_asset_post_data()
post_data["nnname"] = "This field does not exist"
asset_creation = client.post(
url_for("AssetAPI:post"),
json=post_data,
headers={"content-type": "application/json", "Authorization": auth_token},
)
assert asset_creation.status_code == 422
assert asset_creation.json["message"]["json"]["nnname"][0] == "Unknown field."
def test_posting_multiple_assets(client, setup_api_test_data):
"""We can only send one at a time"""
with UserContext("test_admin_user@seita.nl") as prosumer:
auth_token = prosumer.get_auth_token()
post_data1 = get_asset_post_data()
post_data2 = get_asset_post_data()
post_data2["name"] = "Test battery 3"
asset_creation = client.post(
url_for("AssetAPI:post"),
json=[post_data1, post_data2],
headers={"content-type": "application/json", "Authorization": auth_token},
)
print(f"Response: {asset_creation.json}")
assert asset_creation.status_code == 422
assert asset_creation.json["message"]["json"]["_schema"][0] == "Invalid input type."
def test_post_an_asset_with_invalid_data(client, setup_api_test_data):
"""
Add an asset with some fields having invalid data and one field missing.
The right error messages should be in the response and the number of assets has not increased.
"""
with UserContext("test_admin_user@seita.nl") as prosumer:
num_assets_before = len(prosumer.assets)
auth_token = get_auth_token(client, "test_admin_user@seita.nl", "testtest")
post_data = get_asset_post_data()
post_data["name"] = "Something new"
post_data["longitude"] = 300.9
del post_data["generic_asset_type_id"]
post_asset_response = client.post(
url_for("AssetAPI:post"),
json=post_data,
headers={"content-type": "application/json", "Authorization": auth_token},
)
print("Server responded with:\n%s" % post_asset_response.json)
assert post_asset_response.status_code == 422
assert (
"exceeds the maximum longitude"
in post_asset_response.json["message"]["json"]["longitude"][0]
)
assert (
"required field"
in post_asset_response.json["message"]["json"]["generic_asset_type_id"][0]
)
assert (
GenericAsset.query.filter_by(account_id=prosumer.id).count()
== num_assets_before
)