Skip to content

test_versioning#

_get_dataset_from_dataset_versions#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
13
14
15
16
17
def _get_dataset_from_dataset_versions(dataset_versions, ds_id):
    for ds in dataset_versions:
        if ds["id"] == ds_id:
            return ds
    return None

test_dataset_revisions_latest_published#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@pytest.mark.parametrize(
    "query_param",
    ["latest_published=true", "published_revision=1", "all_published_revisions=true"],
)
def test_dataset_revisions_latest_published(
    admin_client, dataset_a, data_catalog, reference_data, query_param
):
    assert dataset_a.response.status_code == 201

    res1 = admin_client.get(
        f"/v3/datasets/{dataset_a.response.data['id']}/revisions?{query_param}",
        content_type="application/json",
    )
    assert res1.status_code == 200
    if not isinstance(res1.data, list):
        res1.data = [res1.data]
    for row in res1.data:
        for field, value in row.items():
            if field not in ["created", "modified"]:
                assert value == dataset_a.response.data[field]

test_dataset_versions#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
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
def test_dataset_versions(admin_client, user_client, dataset_a_json, data_catalog, reference_data):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201
    dataset1 = Dataset.objects.get(id=res1.data["id"])

    res2 = admin_client.post(
        f"/v3/datasets/{res1.data['id']}/new-version", content_type="application/json"
    )
    assert res2.status_code == 201
    dataset2 = Dataset.objects.get(id=res2.data["id"])

    dataset_a_json["title"] = {"en": "new_title"}
    dataset_a_json["state"] = "draft"
    res3 = admin_client.put(
        f"/v3/datasets/{res2.data['id']}", dataset_a_json, content_type="application/json"
    )
    assert res3.status_code == 200
    assert len(res3.data["dataset_versions"]) == 2
    assert len(res3.data["field_of_science"]) == 1

    # Ensure objects are separate copies where required
    assert dataset1.id != dataset2.id
    assert list(dataset1.field_of_science.all()) == list(dataset2.field_of_science.all())
    assert list(dataset1.theme.all()) == list(dataset2.theme.all())
    assert list(dataset1.language.all()) == list(dataset2.language.all())
    assert dataset1.access_rights.id != dataset2.access_rights.id
    assert dataset1.access_rights.license.first().id != dataset2.access_rights.license.first().id
    assert list(dataset1.temporal.all()) != list(dataset2.temporal.all())
    assert list(dataset1.keyword) == list(dataset2.keyword)

    # ensure normal user doesn't see version draft
    res = user_client.get(f"/v3/datasets/{res1.data['id']}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 1
    # but admin does
    res = admin_client.get(f"/v3/datasets/{res1.data['id']}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 2

test_dataset_version_numbers#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def test_dataset_version_numbers(admin_client, dataset_a_json, data_catalog, reference_data):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201
    ds1_id = res1.data["id"]

    res2 = admin_client.post(f"/v3/datasets/{ds1_id}/new-version", content_type="application/json")
    assert res2.status_code == 201
    ds2_id = res2.data["id"]

    # Ensure version numbers are serialized correctly in dataset_versions
    res1 = admin_client.get(f"/v3/datasets/{ds1_id}")
    dataset_versions1 = res1.data["dataset_versions"]
    ds1_version = _get_dataset_from_dataset_versions(dataset_versions1, ds1_id)["version"]
    ds2_version = _get_dataset_from_dataset_versions(dataset_versions1, ds2_id)["version"]
    assert res1.data["version"] == 1
    assert ds1_version == 1
    assert ds2_version == 2

    # Ensure that dataset_versions are identical
    res2 = admin_client.get(f"/v3/datasets/{ds2_id}")
    dataset_versions2 = res2.data["dataset_versions"]
    assert res2.data["version"] == 2
    assert dataset_versions1 == dataset_versions2

test_creating_version_of_a_draft#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
107
108
109
110
111
112
113
def test_creating_version_of_a_draft(admin_client, dataset_c_json, data_catalog, reference_data):
    res1 = admin_client.post("/v3/datasets", dataset_c_json, content_type="application/json")
    assert res1.status_code == 201

    res2 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res2.status_code == 400
    assert res2.data["state"] == "Cannot make a new version of a draft."

test_creating_version_in_wrong_catalog#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
116
117
118
119
120
121
122
123
124
def test_creating_version_in_wrong_catalog(
    admin_client, dataset_a_json, datacatalog_without_versioning, reference_data
):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201

    res2 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res2.status_code == 400
    assert res2.data["data_catalog"] == "Data catalog doesn't support versioning."

test_versions_with_removed_datasets#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
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
def test_versions_with_removed_datasets(
    admin_client, dataset_a_json, data_catalog, reference_data
):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201

    # second version, publish
    res = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res.status_code == 201
    res2 = admin_client.post(f"/v3/datasets/{res.data['id']}/publish")
    assert res2.status_code == 200

    # third version, publish
    res = admin_client.post(f"/v3/datasets/{res2.data['id']}/new-version")
    assert res.status_code == 201
    res3 = admin_client.post(f"/v3/datasets/{res.data['id']}/publish")
    assert res3.status_code == 200

    # delete third version
    res = admin_client.delete(f"/v3/datasets/{res3.data['id']}")
    assert res.status_code == 204

    # try to create a new version from version 3
    res3 = admin_client.post(f"/v3/datasets/{res3.data['id']}/new-version")
    assert res3.status_code == 404

    # create a new version from version 2
    res = admin_client.post(f"/v3/datasets/{res2.data['id']}/new-version")
    assert res.status_code == 201

test_creating_new_version_of_not_latest_dataset#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def test_creating_new_version_of_not_latest_dataset(
    admin_client, dataset_a_json, data_catalog, reference_data
):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201

    res2 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res2.status_code == 201

    res3 = admin_client.post(f"/v3/datasets/{res2.data['id']}/publish")
    assert res3.status_code == 200

    res4 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res4.status_code == 400
    assert (
        res4.data["dataset_versions"]
        == "Newer version of this dataset exists. Only the latest existing version of the dataset can be used to make a new version."
    )

test_creating_new_version_with_existing_version_draft#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def test_creating_new_version_with_existing_version_draft(
    admin_client, dataset_a_json, data_catalog, reference_data
):
    res1 = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201

    res2 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res2.status_code == 201

    res3 = admin_client.post(f"/v3/datasets/{res1.data['id']}/new-version")
    assert res3.status_code == 400
    assert (
        res3.data["dataset_versions"]
        == "There is an existing draft of a new version of this dataset."
    )

test_version_draft_permissions#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
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
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def test_version_draft_permissions(
    admin_client, user_client, user_client_2, dataset_a_json, data_catalog, reference_data
):
    # user1 creates a dataset
    res1 = user_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res1.status_code == 201
    assert res1.data["version"] == 1
    ds1_id = res1.data["id"]

    # user1 creates a second version
    res2 = user_client.post(f"/v3/datasets/{ds1_id}/new-version")
    assert res2.status_code == 201
    assert res2.data["version"] == 2
    ds2_id = res2.data["id"]

    # ensure user1 can see the second version draft in dataset_versions
    res = user_client.get(f"/v3/datasets/{ds1_id}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 2

    # ensure user2 cannot see the second version draft in dataset_versions
    res = user_client_2.get(f"/v3/datasets/{ds1_id}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 1

    # user1 publishes the second version draft
    res = user_client.post(f"/v3/datasets/{ds2_id}/publish")
    assert res.status_code == 200
    assert res.data["version"] == 2

    # ensure user2 can see it now
    res = user_client_2.get(f"/v3/datasets/{ds1_id}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 2

    # user1 creates a draft of edits of second version
    res3 = user_client.post(f"/v3/datasets/{ds2_id}/create-draft")
    assert res3.status_code == 201
    assert res3.data["version"] == 2
    ds3_id = res3.data["id"]

    # ensure user1 can see the draft information
    res = user_client.get(f"/v3/datasets/{ds1_id}?include_nulls=true")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 3
    assert "next_draft" in res.data["dataset_versions"][1].keys()

    # ensure user2 cannot see this information
    res = user_client_2.get(f"/v3/datasets/{ds1_id}?include_nulls=true")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 2
    assert "next_draft" not in res.data["dataset_versions"][1].keys()

    # user1 makes yet another version (3)
    res4 = user_client.post(f"/v3/datasets/{ds2_id}/new-version?include_nulls=true")
    assert res4.status_code == 201
    ds4_id = res4.data["id"]

    # ensure version numbering is correct even with all the drafts (v2 edit draft and v3 version draft)
    assert res4.data["version"] == 3

    # ensure correct amount of versions is shown to both users (2 published, 2 drafts)
    assert len(res4.data["dataset_versions"]) == 4

    res = user_client_2.get(f"/v3/datasets/{ds1_id}")
    assert res.status_code == 200
    assert len(res.data["dataset_versions"]) == 2

    # ensure that version numbers are serialized correctly
    res1 = admin_client.get(f"/v3/datasets/{ds1_id}")
    res2 = admin_client.get(f"/v3/datasets/{ds2_id}")
    res3 = admin_client.get(f"/v3/datasets/{ds3_id}")
    res4 = admin_client.get(f"/v3/datasets/{ds4_id}")

    ds1_version = res1.data["version"]
    ds2_version = res2.data["version"]
    ds3_version = res3.data["version"]
    ds4_version = res4.data["version"]

    assert ds1_version == 1
    assert ds2_version == 2
    assert ds3_version == 2
    assert ds4_version == 3

    # ensure that version numbers in dataset_versions are serialized correctly
    ds1_versions = res1.data["dataset_versions"]
    ds2_versions = res2.data["dataset_versions"]
    ds3_versions = res3.data["dataset_versions"]
    ds4_versions = res4.data["dataset_versions"]

    ds1_version = _get_dataset_from_dataset_versions(ds1_versions, ds1_id)["version"]
    ds2_version = _get_dataset_from_dataset_versions(ds1_versions, ds2_id)["version"]
    ds3_version = _get_dataset_from_dataset_versions(ds1_versions, ds3_id)["version"]
    ds4_version = _get_dataset_from_dataset_versions(ds1_versions, ds4_id)["version"]

    assert ds1_version == 1
    assert ds2_version == 2
    assert ds3_version == 2
    assert ds4_version == 3

    # ensure that dataset_versions are identical
    assert ds1_versions == ds2_versions
    assert ds1_versions == ds3_versions
    assert ds1_versions == ds4_versions

test_list_latest_versions#

Source code in tests/unit/apps/core/api/dataset/test_versioning.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
@pytest.mark.usefixtures("reference_data", "data_catalog")
def test_list_latest_versions(admin_client, user_client, dataset_a_json):
    # Create published dataset + another with 2 published and 1 draft version
    dataset_a_json["title"] = {"en": "dataset"}
    res = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res.status_code == 201

    dataset_a_json["title"] = {"en": "version 1 something"}
    res = admin_client.post("/v3/datasets", dataset_a_json, content_type="application/json")
    assert res.status_code == 201

    res = admin_client.post(
        f"/v3/datasets/{res.data['id']}/new-version", content_type="application/json"
    )
    assert res.status_code == 201

    res = admin_client.patch(
        f"/v3/datasets/{res.data['id']}",
        {"title": {"en": "version 2"}},
        content_type="application/json",
    )
    assert res.status_code == 200

    res = admin_client.post(
        f"/v3/datasets/{res.data['id']}/publish",
        content_type="application/json",
    )
    assert res.status_code == 200

    res = admin_client.post(
        f"/v3/datasets/{res.data['id']}/new-version", content_type="application/json"
    )
    assert res.status_code == 201

    res = admin_client.patch(
        f"/v3/datasets/{res.data['id']}",
        {"title": {"en": "version 3 draft"}},
        content_type="application/json",
    )
    assert res.status_code == 200

    # Admin should see latest versions even if they are drafts
    res = admin_client.get("/v3/datasets?latest_versions=true&ordering=created&pagination=false")
    assert res.status_code == 200
    assert len(res.data) == 2
    assert res.data[0]["title"]["en"] == "dataset"
    assert res.data[1]["title"]["en"] == "version 3 draft"

    # Ordering should work
    res = admin_client.get("/v3/datasets?latest_versions=true&ordering=-created&pagination=false")
    assert res.status_code == 200
    assert res.data[0]["title"]["en"] == "version 3 draft"
    assert res.data[1]["title"]["en"] == "dataset"

    # Filtering should match any latest dataset
    res = admin_client.get("/v3/datasets?latest_versions=true&pagination=false&title=something")
    assert res.status_code == 200
    assert len(res.data) == 0

    res = admin_client.get("/v3/datasets?latest_versions=true&pagination=false&search=draft")
    assert res.status_code == 200
    assert len(res.data) == 1
    assert res.data[0]["title"]["en"] == "version 3 draft"

    # Other user should only see published versions
    res = user_client.get("/v3/datasets?latest_versions=true&ordering=created&pagination=false")
    assert res.status_code == 200
    assert len(res.data) == 2
    assert res.data[0]["title"]["en"] == "dataset"
    assert res.data[1]["title"]["en"] == "version 2"