Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_collection.py
| Show First 20 Lines • Show All 99 Lines • ▼ Show 20 Lines | ): | ||||
| assert response.status_code == status.HTTP_400_BAD_REQUEST | assert response.status_code == status.HTTP_400_BAD_REQUEST | ||||
| msg = "You can only act on deposit with status '%s'" % ( | msg = "You can only act on deposit with status '%s'" % ( | ||||
| DEPOSIT_STATUS_PARTIAL, | DEPOSIT_STATUS_PARTIAL, | ||||
| ) | ) | ||||
| assert msg in response.content.decode("utf-8") | assert msg in response.content.decode("utf-8") | ||||
| def test_add_deposit_when_partial_makes_new_deposit( | def test_add_deposit_when_partial_makes_new_deposit( | ||||
| authenticated_client, deposit_collection, partial_deposit, atom_dataset | authenticated_client, | ||||
| deposit_collection, | |||||
| partial_deposit, | |||||
| atom_dataset, | |||||
| deposit_user, | |||||
| ): | ): | ||||
| """Posting deposit on collection when previous is partial makes new deposit | """Posting deposit on collection when previous is partial makes new deposit | ||||
| """ | """ | ||||
| deposit = partial_deposit | deposit = partial_deposit | ||||
| assert deposit.status == DEPOSIT_STATUS_PARTIAL | assert deposit.status == DEPOSIT_STATUS_PARTIAL | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| # adding a new deposit with the same external id | # adding a new deposit with the same external id | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| reverse(COL_IRI, args=[deposit_collection.name]), | reverse(COL_IRI, args=[deposit_collection.name]), | ||||
| content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
| data=atom_dataset["entry-data0"], | data=atom_dataset["entry-data0"] % origin_url, | ||||
| HTTP_SLUG=deposit.external_id, | HTTP_SLUG=deposit.external_id, | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED, response.content.decode() | ||||
| response_content = parse_xml(BytesIO(response.content)) | response_content = parse_xml(BytesIO(response.content)) | ||||
| deposit_id = response_content["swh:deposit_id"] | deposit_id = response_content["swh:deposit_id"] | ||||
| assert deposit_id != deposit.id # new deposit | assert deposit_id != deposit.id # new deposit | ||||
| new_deposit = Deposit.objects.get(pk=deposit_id) | new_deposit = Deposit.objects.get(pk=deposit_id) | ||||
| assert new_deposit != deposit | assert new_deposit != deposit | ||||
| assert new_deposit.parent is None | assert new_deposit.parent is None | ||||
| def test_add_deposit_when_failed_makes_new_deposit_with_no_parent( | def test_add_deposit_when_failed_makes_new_deposit_with_no_parent( | ||||
| authenticated_client, deposit_collection, failed_deposit, atom_dataset | authenticated_client, deposit_collection, failed_deposit, atom_dataset, deposit_user | ||||
| ): | ): | ||||
| """Posting deposit on collection when deposit done makes new deposit with | """Posting deposit on collection when deposit done makes new deposit with | ||||
| parent | parent | ||||
| """ | """ | ||||
| deposit = failed_deposit | deposit = failed_deposit | ||||
| assert deposit.status == DEPOSIT_STATUS_LOAD_FAILURE | assert deposit.status == DEPOSIT_STATUS_LOAD_FAILURE | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| # adding a new deposit with the same external id as a completed deposit | # adding a new deposit with the same external id as a completed deposit | ||||
| # creates the parenting chain | # creates the parenting chain | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| reverse(COL_IRI, args=[deposit_collection.name]), | reverse(COL_IRI, args=[deposit_collection.name]), | ||||
| content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
| data=atom_dataset["entry-data0"], | data=atom_dataset["entry-data0"] % origin_url, | ||||
| HTTP_SLUG=deposit.external_id, | HTTP_SLUG=deposit.external_id, | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| response_content = parse_xml(BytesIO(response.content)) | response_content = parse_xml(BytesIO(response.content)) | ||||
| deposit_id = response_content["swh:deposit_id"] | deposit_id = response_content["swh:deposit_id"] | ||||
| assert deposit_id != deposit.id | assert deposit_id != deposit.id | ||||
| new_deposit = Deposit.objects.get(pk=deposit_id) | new_deposit = Deposit.objects.get(pk=deposit_id) | ||||
| assert new_deposit != deposit | assert new_deposit != deposit | ||||
| assert new_deposit.parent is None | assert new_deposit.parent is None | ||||
| def test_add_deposit_when_done_makes_new_deposit_with_parent_old_one( | def test_add_deposit_when_done_makes_new_deposit_with_parent_old_one( | ||||
| authenticated_client, deposit_collection, completed_deposit, atom_dataset | authenticated_client, | ||||
| deposit_collection, | |||||
| completed_deposit, | |||||
| atom_dataset, | |||||
| deposit_user, | |||||
| ): | ): | ||||
| """Posting deposit on collection when deposit done makes new deposit with | """Posting deposit on collection when deposit done makes new deposit with | ||||
| parent | parent | ||||
| """ | """ | ||||
| # given multiple deposit already loaded | # given multiple deposit already loaded | ||||
| deposit = completed_deposit | deposit = completed_deposit | ||||
| assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| # adding a new deposit with the same external id as a completed deposit | # adding a new deposit with the same external id as a completed deposit | ||||
| # creates the parenting chain | # creates the parenting chain | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| reverse(COL_IRI, args=[deposit_collection.name]), | reverse(COL_IRI, args=[deposit_collection.name]), | ||||
| content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
| data=atom_dataset["entry-data0"], | data=atom_dataset["entry-data0"] % origin_url, | ||||
| HTTP_SLUG=deposit.external_id, | HTTP_SLUG=deposit.external_id, | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| response_content = parse_xml(BytesIO(response.content)) | response_content = parse_xml(BytesIO(response.content)) | ||||
| deposit_id = response_content["swh:deposit_id"] | deposit_id = response_content["swh:deposit_id"] | ||||
| assert deposit_id != deposit.id | assert deposit_id != deposit.id | ||||
| new_deposit = Deposit.objects.get(pk=deposit_id) | new_deposit = Deposit.objects.get(pk=deposit_id) | ||||
| assert deposit.collection == new_deposit.collection | assert deposit.collection == new_deposit.collection | ||||
| assert deposit.external_id == new_deposit.external_id | assert deposit.origin_url == origin_url | ||||
| assert new_deposit != deposit | assert new_deposit != deposit | ||||
| assert new_deposit.parent == deposit | assert new_deposit.parent == deposit | ||||
| def test_add_deposit_external_id_conflict_no_parent( | def test_add_deposit_external_id_conflict_no_parent( | ||||
| authenticated_client, | authenticated_client, | ||||
| another_authenticated_client, | another_authenticated_client, | ||||
| deposit_collection, | deposit_collection, | ||||
| deposit_another_collection, | deposit_another_collection, | ||||
| atom_dataset, | atom_dataset, | ||||
| sample_archive, | sample_archive, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Posting a deposit with an external_id conflicting with an external_id | """Posting a deposit with an external_id conflicting with an external_id | ||||
| of a different client does not create a parent relationship | of a different client does not create a parent relationship | ||||
| """ | """ | ||||
| external_id = "foobar" | external_id = "foobar" | ||||
| origin_url = deposit_user.provider_url + external_id | |||||
| # create a deposit for that other user, with the same slug | # create a deposit for that other user, with the same slug | ||||
| other_deposit = create_deposit( | other_deposit = create_deposit( | ||||
| another_authenticated_client, | another_authenticated_client, | ||||
| deposit_another_collection.name, | deposit_another_collection.name, | ||||
| sample_archive, | sample_archive, | ||||
| external_id, | external_id, | ||||
| DEPOSIT_STATUS_LOAD_SUCCESS, | DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
| ) | ) | ||||
| # adding a new deposit with the same external id as a completed deposit | # adding a new deposit with the same external id as a completed deposit | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| reverse(COL_IRI, args=[deposit_collection.name]), | reverse(COL_IRI, args=[deposit_collection.name]), | ||||
| content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
| data=atom_dataset["entry-data0"], | data=atom_dataset["entry-data0"] % origin_url, | ||||
| HTTP_SLUG=external_id, | HTTP_SLUG=external_id, | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| response_content = parse_xml(BytesIO(response.content)) | response_content = parse_xml(BytesIO(response.content)) | ||||
| deposit_id = response_content["swh:deposit_id"] | deposit_id = response_content["swh:deposit_id"] | ||||
| assert other_deposit.id != deposit_id | assert other_deposit.id != deposit_id | ||||
| new_deposit = Deposit.objects.get(pk=deposit_id) | new_deposit = Deposit.objects.get(pk=deposit_id) | ||||
| assert new_deposit.parent is None | assert new_deposit.parent is None | ||||
| def test_add_deposit_external_id_conflict_with_parent( | def test_add_deposit_external_id_conflict_with_parent( | ||||
| authenticated_client, | authenticated_client, | ||||
| another_authenticated_client, | another_authenticated_client, | ||||
| deposit_collection, | deposit_collection, | ||||
| deposit_another_collection, | deposit_another_collection, | ||||
| completed_deposit, | completed_deposit, | ||||
| atom_dataset, | atom_dataset, | ||||
| sample_archive, | sample_archive, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Posting a deposit with an external_id conflicting with an external_id | """Posting a deposit with an external_id conflicting with an external_id | ||||
| of a different client creates a parent relationship with the deposit | of a different client creates a parent relationship with the deposit | ||||
| of the right client instead of the last matching deposit | of the right client instead of the last matching deposit | ||||
| """ | """ | ||||
| # given multiple deposit already loaded | # given multiple deposit already loaded | ||||
| deposit = completed_deposit | deposit = completed_deposit | ||||
| assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| # create a deposit for that other user, with the same slug | # create a deposit for that other user, with the same slug | ||||
| other_deposit = create_deposit( | other_deposit = create_deposit( | ||||
| another_authenticated_client, | another_authenticated_client, | ||||
| deposit_another_collection.name, | deposit_another_collection.name, | ||||
| sample_archive, | sample_archive, | ||||
| deposit.external_id, | deposit.external_id, | ||||
| DEPOSIT_STATUS_LOAD_SUCCESS, | DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
| ) | ) | ||||
| # adding a new deposit with the same external id as a completed deposit | # adding a new deposit with the same external id as a completed deposit | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| reverse(COL_IRI, args=[deposit_collection.name]), | reverse(COL_IRI, args=[deposit_collection.name]), | ||||
| content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
| data=atom_dataset["entry-data0"], | data=atom_dataset["entry-data0"] % origin_url, | ||||
| HTTP_SLUG=deposit.external_id, | HTTP_SLUG=deposit.external_id, | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| response_content = parse_xml(BytesIO(response.content)) | response_content = parse_xml(BytesIO(response.content)) | ||||
| deposit_id = response_content["swh:deposit_id"] | deposit_id = response_content["swh:deposit_id"] | ||||
| assert deposit_id != deposit.id | assert deposit_id != deposit.id | ||||
| assert other_deposit.id != deposit.id | assert other_deposit.id != deposit.id | ||||
| new_deposit = Deposit.objects.get(pk=deposit_id) | new_deposit = Deposit.objects.get(pk=deposit_id) | ||||
| assert deposit.collection == new_deposit.collection | assert deposit.collection == new_deposit.collection | ||||
| assert deposit.external_id == new_deposit.external_id | assert deposit.external_id == new_deposit.external_id | ||||
| assert new_deposit != deposit | assert new_deposit != deposit | ||||
| assert new_deposit.parent == deposit | assert new_deposit.parent == deposit | ||||