Differential D4611 Diff 16404 swh/deposit/tests/api/test_collection_post_atom.py

Changeset View

Standalone View

swh/deposit/tests/api/test_collection_post_atom.py

Show All 28 Lines	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_error_with_decimal,		data=atom_error_with_decimal,
HTTP_SLUG="external-id",		HTTP_SLUG="external-id",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
)		)

# then		# then
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"]

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
dr = DepositRequest.objects.get(deposit=deposit)		dr = DepositRequest.objects.get(deposit=deposit)

assert dr.metadata is not None		assert dr.metadata is not None
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines	response = authenticated_client.post(
content_type="application/atom+xml;type=entry",		content_type="application/atom+xml;type=entry",
data=atom_dataset["entry-data-parsing-error-prone"],		data=atom_dataset["entry-data-parsing-error-prone"],
HTTP_SLUG="external-id",		HTTP_SLUG="external-id",
)		)
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST
assert b"Malformed xml metadata" in response.content		assert b"Malformed xml metadata" in response.content


def test_post_deposit_atom_no_slug_header(		def test_post_deposit_atom_no_slug_header(
		ardumontUnsubmitted Done Inline Actions test name is wrong from the start, we pass a slug header here. Can you please change it? ardumont: test name is wrong from the start, we pass a slug header here. Can you please change it?
		ardumontUnsubmitted Done Inline Actions ping ^ ardumont: ping ^
		vlorentzAuthorUnsubmitted Done Inline Actions oops, I applied the change to the wrong commit vlorentz: oops, I applied the change to the wrong commit
authenticated_client, deposit_collection, deposit_user, atom_dataset, mocker		authenticated_client, deposit_collection, deposit_user, atom_dataset, mocker
):		):
"""Posting an atom entry without a slug header should generate one		"""Posting an atom entry with a slug header but no origin url generates
		an origin url from the slug

"""		"""
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

id_ = str(uuid.uuid4())		slug = str(uuid.uuid4())
mocker.patch("uuid.uuid4", return_value=id_)

# when		# when
response = authenticated_client.post(		response = authenticated_client.post(
url,		url,
content_type="application/atom+xml;type=entry",		content_type="application/atom+xml;type=entry",
data=atom_dataset["entry-data0"],		data=atom_dataset["entry-data-no-origin-url"],
		HTTP_IN_PROGRESS="false",
		HTTP_SLUG=slug,
		)

		assert response.status_code == status.HTTP_201_CREATED
		response_content = parse_xml(BytesIO(response.content))
		deposit_id = response_content["swh:deposit_id"]

		deposit = Deposit.objects.get(pk=deposit_id)
		assert deposit.collection == deposit_collection
		assert deposit.origin_url == deposit_user.provider_url + slug
		assert deposit.status == DEPOSIT_STATUS_DEPOSITED


		ardumontUnsubmitted Done Inline Actions `nor_slug_` is clearer ardumont: `nor_slug_` is clearer
		def test_post_deposit_atom_no_origin_url_nor_slug_header(
		authenticated_client, deposit_collection, deposit_user, atom_dataset, mocker
		):
		"""Posting an atom entry without an origin url or a slug header should generate one

		"""
		url = reverse(COL_IRI, args=[deposit_collection.name])

		slug = str(uuid.uuid4())
		mocker.patch("uuid.uuid4", return_value=slug)

		# when
		response = authenticated_client.post(
		url,
		content_type="application/atom+xml;type=entry",
		data=atom_dataset["entry-data-no-origin-url"],
# + headers		# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
)		)

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"]

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.origin_url == deposit_user.provider_url + id_		assert deposit.origin_url == deposit_user.provider_url + slug
assert deposit.status == DEPOSIT_STATUS_DEPOSITED		assert deposit.status == DEPOSIT_STATUS_DEPOSITED


def test_post_deposit_atom_with_external_identifier(		def test_post_deposit_atom_with_external_identifier(
authenticated_client, deposit_collection, atom_dataset		authenticated_client, deposit_collection, atom_dataset
):		):
"""Posting an atom entry without a slug header should return a 400		"""Posting an atom entry without a slug header should return a 400

"""		"""
		external_id = "foobar"
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

# when		# when
response = authenticated_client.post(		response = authenticated_client.post(
url,		url,
content_type="application/atom+xml;type=entry",		content_type="application/atom+xml;type=entry",
data=atom_dataset["error-with-external-identifier"],		data=atom_dataset["error-with-external-identifier"] % external_id,
# + headers		# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_SLUG="something",		HTTP_SLUG="something",
)		)

assert b"The 'external_identifier' tag is deprecated" in response.content		assert b"The 'external_identifier' tag is deprecated" in response.content
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST


		def test_post_deposit_atom_with_create_origin_and_external_identifier(
		authenticated_client, deposit_collection, atom_dataset, deposit_user
		):
		"""<atom:external_identifier> was deprecated before <swh:create_origin>
		was introduced, clients should get an error when trying to use both

		"""
		external_id = "foobar"
		origin_url = deposit_user.provider_url + external_id
		url = reverse(COL_IRI, args=[deposit_collection.name])

		document = atom_dataset["error-with-external-identifier-and-create-origin"].format(
		external_id=external_id, url=origin_url,
		)

		# when
		response = authenticated_client.post(
		url,
		content_type="application/atom+xml;type=entry",
		data=document,
		# + headers
		HTTP_IN_PROGRESS="false",
		)

		assert b"<external_identifier> is deprecated" in response.content
		assert response.status_code == status.HTTP_400_BAD_REQUEST


		def test_post_deposit_atom_with_create_origin_and_reference(
		authenticated_client, deposit_collection, atom_dataset, deposit_user
		):
		"""<swh:reference> and <swh:create_origin> are mutually exclusive

		"""
		external_id = "foobar"
		origin_url = deposit_user.provider_url + external_id
		url = reverse(COL_IRI, args=[deposit_collection.name])

		document = atom_dataset["error-with-reference-and-create-origin"].format(
		external_id=external_id, url=origin_url,
		)

		# when
		response = authenticated_client.post(
		url,
		content_type="application/atom+xml;type=entry",
		data=document,
		# + headers
		HTTP_IN_PROGRESS="false",
		)

		assert b"only one may be used on a given deposit" in response.content
		assert response.status_code == status.HTTP_400_BAD_REQUEST


def test_post_deposit_atom_unknown_collection(authenticated_client, atom_dataset):		def test_post_deposit_atom_unknown_collection(authenticated_client, atom_dataset):
"""Posting an atom entry to an unknown collection should return a 404		"""Posting an atom entry to an unknown collection should return a 404

"""		"""
unknown_collection = "unknown-one"		unknown_collection = "unknown-one"
with pytest.raises(DepositCollection.DoesNotExist):		with pytest.raises(DepositCollection.DoesNotExist):
DepositCollection.objects.get(name=unknown_collection)		DepositCollection.objects.get(name=unknown_collection)

response = authenticated_client.post(		response = authenticated_client.post(
reverse(COL_IRI, args=[unknown_collection]), # <- unknown collection		reverse(COL_IRI, args=[unknown_collection]), # <- unknown collection
content_type="application/atom+xml;type=entry",		content_type="application/atom+xml;type=entry",
data=atom_dataset["entry-data0"],		data=atom_dataset["entry-data0"],
HTTP_SLUG="something",		HTTP_SLUG="something",
)		)
assert response.status_code == status.HTTP_404_NOT_FOUND		assert response.status_code == status.HTTP_404_NOT_FOUND
assert b"Unknown collection" in response.content		assert b"Unknown collection" in response.content


def test_post_deposit_atom_entry_initial(		def test_post_deposit_atom_entry_initial(
authenticated_client, deposit_collection, atom_dataset		authenticated_client, deposit_collection, atom_dataset, deposit_user
):		):
"""Posting an initial atom entry should return 201 with deposit receipt		"""Posting an initial atom entry should return 201 with deposit receipt

"""		"""
# given		# given
external_id = "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a"		origin_url = deposit_user.provider_url + "1225c695-cfb8-4ebb-aaaa-80da344efa6a"

with pytest.raises(Deposit.DoesNotExist):		with pytest.raises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)		Deposit.objects.get(origin_url=origin_url)

atom_entry_data = atom_dataset["entry-data0"]		atom_entry_data = atom_dataset["entry-data0"] % origin_url

# when		# when
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_entry_data,		data=atom_entry_data,
HTTP_SLUG=external_id,
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
)		)

# then		# then
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"]

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.external_id == external_id		assert deposit.origin_url == origin_url
assert deposit.status == DEPOSIT_STATUS_DEPOSITED		assert deposit.status == DEPOSIT_STATUS_DEPOSITED

# one associated request to a deposit		# one associated request to a deposit
deposit_request = DepositRequest.objects.get(deposit=deposit)		deposit_request = DepositRequest.objects.get(deposit=deposit)
assert deposit_request.metadata is not None		assert deposit_request.metadata is not None
assert deposit_request.raw_metadata == atom_entry_data		assert deposit_request.raw_metadata == atom_entry_data
assert bool(deposit_request.archive) is False		assert bool(deposit_request.archive) is False


def test_post_deposit_atom_entry_with_codemeta(		def test_post_deposit_atom_entry_with_codemeta(
authenticated_client, deposit_collection, atom_dataset		authenticated_client, deposit_collection, atom_dataset, deposit_user
):		):
"""Posting an initial atom entry should return 201 with deposit receipt		"""Posting an initial atom entry should return 201 with deposit receipt

"""		"""
# given		# given
external_id = "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a"		origin_url = deposit_user.provider_url + "1225c695-cfb8-4ebb-aaaa-80da344efa6a"

with pytest.raises(Deposit.DoesNotExist):		with pytest.raises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)		Deposit.objects.get(origin_url=origin_url)

atom_entry_data = atom_dataset["codemeta-sample"]		atom_entry_data = atom_dataset["codemeta-sample"] % origin_url
# when		# when
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_entry_data,		data=atom_entry_data,
HTTP_SLUG=external_id,
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
)		)

# then		# then
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"]

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.external_id == external_id		assert deposit.origin_url == origin_url
assert deposit.status == DEPOSIT_STATUS_DEPOSITED		assert deposit.status == DEPOSIT_STATUS_DEPOSITED

# one associated request to a deposit		# one associated request to a deposit
deposit_request = DepositRequest.objects.get(deposit=deposit)		deposit_request = DepositRequest.objects.get(deposit=deposit)
assert deposit_request.metadata is not None		assert deposit_request.metadata is not None
assert deposit_request.raw_metadata == atom_entry_data		assert deposit_request.raw_metadata == atom_entry_data
assert bool(deposit_request.archive) is False		assert bool(deposit_request.archive) is False


def test_post_deposit_atom_entry_multiple_steps(		def test_post_deposit_atom_entry_multiple_steps(
authenticated_client, deposit_collection, atom_dataset		authenticated_client, deposit_collection, atom_dataset, deposit_user
):		):
"""After initial deposit, updating a deposit should return a 201		"""After initial deposit, updating a deposit should return a 201

"""		"""
# given		# given
external_id = "urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a"		origin_url = deposit_user.provider_url + "2225c695-cfb8-4ebb-aaaa-80da344efa6a"

with pytest.raises(Deposit.DoesNotExist):		with pytest.raises(Deposit.DoesNotExist):
deposit = Deposit.objects.get(external_id=external_id)		deposit = Deposit.objects.get(origin_url=origin_url)

# when		# when
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-data1"],		data=atom_dataset["entry-data1"],
HTTP_IN_PROGRESS="True",		HTTP_IN_PROGRESS="True",
HTTP_SLUG=external_id,
)		)

# then		# then
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 = int(response_content["swh:deposit_id"])		deposit_id = int(response_content["swh:deposit_id"])

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.external_id == external_id		assert deposit.origin_url is None # not provided yet
assert deposit.status == "partial"		assert deposit.status == "partial"

# one associated request to a deposit		# one associated request to a deposit
deposit_requests = DepositRequest.objects.filter(deposit=deposit)		deposit_requests = DepositRequest.objects.filter(deposit=deposit)
assert len(deposit_requests) == 1		assert len(deposit_requests) == 1

atom_entry_data = atom_dataset["entry-data-minimal"]		atom_entry_data = atom_dataset["entry-only-create-origin"] % (origin_url)

for link in response_content["atom:link"]:		for link in response_content["atom:link"]:
if link["@rel"] == "http://purl.org/net/sword/terms/add":		if link["@rel"] == "http://purl.org/net/sword/terms/add":
se_iri = link["@href"]		se_iri = link["@href"]
break		break
else:		else:
assert False, f"missing SE-IRI from {response_content['link']}"		assert False, f"missing SE-IRI from {response_content['link']}"

# when updating the first deposit post		# when updating the first deposit post
response = authenticated_client.post(		response = authenticated_client.post(
se_iri,		se_iri,
content_type="application/atom+xml;type=entry",		content_type="application/atom+xml;type=entry",
data=atom_entry_data,		data=atom_entry_data,
HTTP_IN_PROGRESS="False",		HTTP_IN_PROGRESS="False",
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED, response.content		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 = int(response_content["swh:deposit_id"])		deposit_id = int(response_content["swh:deposit_id"])

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.external_id == external_id		assert deposit.origin_url == origin_url
assert deposit.status == DEPOSIT_STATUS_DEPOSITED		assert deposit.status == DEPOSIT_STATUS_DEPOSITED

assert len(Deposit.objects.all()) == 1		assert len(Deposit.objects.all()) == 1

# now 2 associated requests to a same deposit		# now 2 associated requests to a same deposit
deposit_requests = DepositRequest.objects.filter(deposit=deposit).order_by("id")		deposit_requests = DepositRequest.objects.filter(deposit=deposit).order_by("id")
assert len(deposit_requests) == 2		assert len(deposit_requests) == 2

Show All 11 Lines