Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_collection_list.py
| Show First 20 Lines • Show All 105 Lines • ▼ Show 20 Lines | ): | ||||
| assert data3["swh:count"] == "2" # total result of 2 deposits across all results | assert data3["swh:count"] == "2" # total result of 2 deposits across all results | ||||
| deposits3 = data3["atom:entry"] # list here | deposits3 = data3["atom:entry"] # list here | ||||
| assert isinstance(deposits3, list) | assert isinstance(deposits3, list) | ||||
| assert len(deposits3) == 2 | assert len(deposits3) == 2 | ||||
| header_link3 = parse_header_links(response3["Link"]) | header_link3 = parse_header_links(response3["Link"]) | ||||
| assert header_link3 == [] # no pagination as all results received in one round | assert header_link3 == [] # no pagination as all results received in one round | ||||
| assert deposit in deposits3 | assert deposit in deposits3 | ||||
| assert deposit2 in deposits3 | assert deposit2 in deposits3 | ||||
| def test_deposit_collection_list_origin_filter_ko( | |||||
| partial_deposit, deposited_deposit, authenticated_client | |||||
| ): | |||||
| """Deposit list with irrelevant filter on origin renders no result | |||||
| """ | |||||
| irrelevant_origin = "foobar" | |||||
| client_id = authenticated_client.deposit_client.id | |||||
| # Both deposit were deposited by the authenticated client... | |||||
| assert partial_deposit.client.id == client_id | |||||
| assert deposited_deposit.client.id == client_id | |||||
| # ... but have no origin_url matching the filter below... | |||||
| assert partial_deposit.origin_url != irrelevant_origin | |||||
| assert deposited_deposit.origin_url != irrelevant_origin | |||||
| coll = partial_deposit.collection | |||||
| url = reverse(COL_IRI, args=(coll.name,)) | |||||
| response = authenticated_client.get( | |||||
| f"{url}?page_size=1&origin_url={irrelevant_origin}" | |||||
| ) | |||||
| assert response.status_code == status.HTTP_200_OK | |||||
| data = parse_xml(BytesIO(response.content))["atom:feed"] | |||||
| # so they won't get listed | |||||
| assert data["swh:count"] == "0" | |||||
| def test_deposit_collection_list_origin_filter_nominal( | |||||
| partial_deposit, deposited_deposit, authenticated_client | |||||
| ): | |||||
| """Deposit list with correct origin filters accordingly the deposits | |||||
| """ | |||||
| filtered_origin = deposited_deposit.origin_url | |||||
| client_id = authenticated_client.deposit_client.id | |||||
| # patch partial deposit so the filtering works as expected | |||||
| partial_deposit.origin_url = filtered_origin | |||||
| partial_deposit.save() | |||||
| # Both deposit were deposited by the authenticated client | |||||
| assert partial_deposit.client.id == client_id | |||||
| assert deposited_deposit.client.id == client_id | |||||
| # both have the right origin so they will get returned | |||||
| assert partial_deposit.origin_url == filtered_origin | |||||
| assert deposited_deposit.origin_url == filtered_origin | |||||
| deposit_id = str(partial_deposit.id) | |||||
| deposit_id2 = str(deposited_deposit.id) | |||||
| assert partial_deposit.id < deposited_deposit.id | |||||
| coll = partial_deposit.collection | |||||
| # requesting the listing of the deposit for the user's collection | |||||
| url = reverse(COL_IRI, args=(coll.name,)) | |||||
| response = authenticated_client.get( | |||||
| f"{url}?origin_url={filtered_origin}&page_size=1" | |||||
| ) | |||||
| assert response.status_code == status.HTTP_200_OK | |||||
| data = parse_xml(BytesIO(response.content))["atom:feed"] | |||||
| assert data["swh:count"] == "2" # total result of 2 deposits | |||||
| header_link = parse_header_links(response["Link"]) | |||||
| assert len(header_link) == 1 # only 1 next link | |||||
| assert header_link[0]["rel"] == "next" | |||||
| # only 1 deposit in the response | |||||
| deposit = data["atom:entry"] # dict as only 1 value (a-la js) | |||||
| assert isinstance(deposit, dict) | |||||
| assert deposit["swh:id"] == deposit_id | |||||
| assert deposit["swh:status"] == DEPOSIT_STATUS_PARTIAL | |||||
| # then 2nd page | |||||
| next_page_link = header_link[0]["url"] | |||||
| response2 = authenticated_client.get(next_page_link) | |||||
| assert response2.status_code == status.HTTP_200_OK | |||||
| data2 = parse_xml(BytesIO(response2.content))["atom:feed"] | |||||
| assert data2["swh:count"] == "2" # still total of 2 deposits across all results | |||||
| header_link2 = parse_header_links(response2["Link"]) | |||||
| assert len(header_link2) == 1 # only 1 previous link | |||||
| assert header_link2[0]["rel"] == "previous" | |||||
| # only 1 deposit in the response | |||||
| deposit2 = data2["atom:entry"] # dict as only 1 value (a-la js) | |||||
| assert isinstance(deposit2, dict) | |||||
| assert deposit2["swh:id"] == deposit_id2 | |||||
| assert deposit2["swh:status"] == DEPOSIT_STATUS_DEPOSITED | |||||
| # Retrieve every deposit in one query (no page_size parameter) | |||||
| response3 = authenticated_client.get(url) | |||||
| assert response3.status_code == status.HTTP_200_OK | |||||
| data3 = parse_xml(BytesIO(response3.content))["atom:feed"] | |||||
| assert data3["swh:count"] == "2" # total result of 2 deposits across all results | |||||
| deposits3 = data3["atom:entry"] # list here | |||||
| assert isinstance(deposits3, list) | |||||
| assert len(deposits3) == 2 | |||||
| header_link3 = parse_header_links(response3["Link"]) | |||||
| assert header_link3 == [] # no pagination as all results received in one round | |||||
| assert deposit in deposits3 | |||||
| assert deposit2 in deposits3 | |||||