Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_state.py
| # Copyright (C) 2017-2021 The Software Heritage developers | # Copyright (C) 2017-2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from io import BytesIO | from xml.etree import ElementTree | ||||
| from django.urls import reverse_lazy as reverse | from django.urls import reverse_lazy as reverse | ||||
| from rest_framework import status | from rest_framework import status | ||||
| from swh.deposit.config import ( | from swh.deposit.config import ( | ||||
| DEPOSIT_STATUS_DEPOSITED, | DEPOSIT_STATUS_DEPOSITED, | ||||
| DEPOSIT_STATUS_REJECTED, | DEPOSIT_STATUS_REJECTED, | ||||
| STATE_IRI, | STATE_IRI, | ||||
| ) | ) | ||||
| from swh.deposit.models import DEPOSIT_STATUS_DETAIL, DEPOSIT_STATUS_LOAD_SUCCESS | from swh.deposit.models import DEPOSIT_STATUS_DETAIL, DEPOSIT_STATUS_LOAD_SUCCESS | ||||
| from swh.deposit.parsers import parse_xml | from swh.deposit.utils import NAMESPACES | ||||
| def test_post_deposit_with_status_check(authenticated_client, deposited_deposit): | def test_post_deposit_with_status_check(authenticated_client, deposited_deposit): | ||||
| """Successful but not loaded deposit should have a status 'deposited' | """Successful but not loaded deposit should have a status 'deposited' | ||||
| """ | """ | ||||
| deposit = deposited_deposit | deposit = deposited_deposit | ||||
| status_url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | status_url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | ||||
| # check status | # check status | ||||
| status_response = authenticated_client.get(status_url) | status_response = authenticated_client.get(status_url) | ||||
| assert status_response.status_code == status.HTTP_200_OK | assert status_response.status_code == status.HTTP_200_OK | ||||
| r = parse_xml(BytesIO(status_response.content)) | r = ElementTree.fromstring(status_response.content) | ||||
| assert int(r["swh:deposit_id"]) == deposit.id | assert int(r.findtext("swh:deposit_id", namespaces=NAMESPACES)) == deposit.id | ||||
| assert r["swh:deposit_status"] == DEPOSIT_STATUS_DEPOSITED | |||||
| assert ( | assert ( | ||||
| r["swh:deposit_status_detail"] | r.findtext("swh:deposit_status", namespaces=NAMESPACES) | ||||
| == DEPOSIT_STATUS_DEPOSITED | |||||
| ) | |||||
| assert ( | |||||
| r.findtext("swh:deposit_status_detail", namespaces=NAMESPACES) | |||||
| == DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_DEPOSITED] | == DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_DEPOSITED] | ||||
| ) | ) | ||||
| assert r["swh:deposit_external_id"] == deposit.external_id | assert ( | ||||
| assert r["swh:deposit_origin_url"] == deposit.origin_url | r.findtext("swh:deposit_external_id", namespaces=NAMESPACES) | ||||
| == deposit.external_id | |||||
| ) | |||||
| assert ( | |||||
| r.findtext("swh:deposit_origin_url", namespaces=NAMESPACES) | |||||
| == deposit.origin_url | |||||
| ) | |||||
| def test_status_unknown_deposit(authenticated_client, deposit_collection): | def test_status_unknown_deposit(authenticated_client, deposit_collection): | ||||
| """Unknown deposit status should return 404 response | """Unknown deposit status should return 404 response | ||||
| """ | """ | ||||
| unknown_deposit_id = 999 | unknown_deposit_id = 999 | ||||
| status_url = reverse(STATE_IRI, args=[deposit_collection.name, unknown_deposit_id]) | status_url = reverse(STATE_IRI, args=[deposit_collection.name, unknown_deposit_id]) | ||||
| Show All 19 Lines | def test_status_deposit_rejected(authenticated_client, rejected_deposit): | ||||
| url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | ||||
| # when | # when | ||||
| status_response = authenticated_client.get(url) | status_response = authenticated_client.get(url) | ||||
| # then | # then | ||||
| assert status_response.status_code == status.HTTP_200_OK | assert status_response.status_code == status.HTTP_200_OK | ||||
| r = parse_xml(BytesIO(status_response.content)) | r = ElementTree.fromstring(status_response.content) | ||||
| assert int(r["swh:deposit_id"]) == deposit.id | assert int(r.findtext("swh:deposit_id", namespaces=NAMESPACES)) == deposit.id | ||||
| assert r["swh:deposit_status"] == DEPOSIT_STATUS_REJECTED | assert ( | ||||
| assert r["swh:deposit_status_detail"] == "Deposit failed the checks" | r.findtext("swh:deposit_status", namespaces=NAMESPACES) | ||||
| == DEPOSIT_STATUS_REJECTED | |||||
| ) | |||||
| assert ( | |||||
| r.findtext("swh:deposit_status_detail", namespaces=NAMESPACES) | |||||
| == "Deposit failed the checks" | |||||
| ) | |||||
| if deposit.swhid: | if deposit.swhid: | ||||
| assert r["swh:deposit_swhid"] == deposit.swhid | assert r.findtext("swh:deposit_swhid", namespaces=NAMESPACES) == deposit.swhid | ||||
| def test_status_with_http_accept_header_should_not_break( | def test_status_with_http_accept_header_should_not_break( | ||||
| authenticated_client, partial_deposit | authenticated_client, partial_deposit | ||||
| ): | ): | ||||
| """Asking deposit status with Accept header should return 200 | """Asking deposit status with Accept header should return 200 | ||||
| """ | """ | ||||
| Show All 17 Lines | def test_status_complete_deposit(authenticated_client, complete_deposit): | ||||
| deposit = complete_deposit | deposit = complete_deposit | ||||
| url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | url = reverse(STATE_IRI, args=[deposit.collection.name, deposit.id]) | ||||
| # when | # when | ||||
| status_response = authenticated_client.get(url) | status_response = authenticated_client.get(url) | ||||
| # then | # then | ||||
| assert status_response.status_code == status.HTTP_200_OK | assert status_response.status_code == status.HTTP_200_OK | ||||
| r = parse_xml(BytesIO(status_response.content)) | r = ElementTree.fromstring(status_response.content) | ||||
| assert int(r["swh:deposit_id"]) == deposit.id | assert int(r.findtext("swh:deposit_id", namespaces=NAMESPACES)) == deposit.id | ||||
| assert r["swh:deposit_status"] == DEPOSIT_STATUS_LOAD_SUCCESS | assert ( | ||||
| r.findtext("swh:deposit_status", namespaces=NAMESPACES) | |||||
| == DEPOSIT_STATUS_LOAD_SUCCESS | |||||
| ) | |||||
| assert ( | assert ( | ||||
| r["swh:deposit_status_detail"] | r.findtext("swh:deposit_status_detail", namespaces=NAMESPACES) | ||||
| == DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_LOAD_SUCCESS] | == DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_LOAD_SUCCESS] | ||||
| ) | ) | ||||
| assert deposit.swhid is not None | assert deposit.swhid is not None | ||||
| assert r["swh:deposit_swh_id"] == deposit.swhid | assert r.findtext("swh:deposit_swh_id", namespaces=NAMESPACES) == deposit.swhid | ||||
| assert deposit.swhid_context is not None | assert deposit.swhid_context is not None | ||||
| assert r["swh:deposit_swh_id_context"] == deposit.swhid_context | assert ( | ||||
| assert r["swh:deposit_origin_url"] == deposit.origin_url | r.findtext("swh:deposit_swh_id_context", namespaces=NAMESPACES) | ||||
| == deposit.swhid_context | |||||
| ) | |||||
| assert ( | |||||
| r.findtext("swh:deposit_origin_url", namespaces=NAMESPACES) | |||||
| == deposit.origin_url | |||||
| ) | |||||