diff --git a/swh/deposit/api/private/deposit_list.py b/swh/deposit/api/private/deposit_list.py index 4bb40e31..30ca768e 100644 --- a/swh/deposit/api/private/deposit_list.py +++ b/swh/deposit/api/private/deposit_list.py @@ -1,68 +1,198 @@ # Copyright (C) 2018-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information - +from typing import Any, Dict +from xml.etree import ElementTree + +from django.conf import settings +from django.core.paginator import Paginator +from django.db.models import CharField, Q, TextField +from django.http import JsonResponse +from rest_framework.decorators import ( + api_view, + authentication_classes, + permission_classes, +) from rest_framework.generics import ListAPIView +from rest_framework.permissions import AllowAny +from rest_framework.request import Request +import sentry_sdk from swh.deposit.api.utils import DefaultPagination, DepositSerializer +from swh.deposit.utils import parse_swh_deposit_origin, parse_swh_metadata_provenance +from swh.model.swhids import QualifiedSWHID from . import APIPrivateView from ...models import Deposit +def _enrich_deposit_with_metadata(deposit: Deposit) -> Deposit: + deposit_requests = deposit.depositrequest_set.filter(type="metadata") + deposit_requests = deposit_requests.order_by("-id") + # enrich deposit with raw metadata when we have some + if deposit_requests and len(deposit_requests) > 0: + raw_meta = deposit_requests[0].raw_metadata + if raw_meta: + deposit.set_raw_metadata(raw_meta) + return deposit + + class APIList(ListAPIView, APIPrivateView): """Deposit request class to list the deposit's status per page. HTTP verbs supported: GET """ serializer_class = DepositSerializer pagination_class = DefaultPagination def paginate_queryset(self, queryset): """Return a single page of results. This enriches the queryset results with metadata if any. """ page_result = self.paginator.paginate_queryset( queryset, self.request, view=self ) deposits = [] for deposit in page_result: - deposit_requests = deposit.depositrequest_set.filter( - type="metadata" - ).order_by("-id") - # enrich deposit with raw metadata when we have some - if deposit_requests and len(deposit_requests) > 0: - raw_meta = deposit_requests[0].raw_metadata - if raw_meta: - deposit.set_raw_metadata(raw_meta) - + _enrich_deposit_with_metadata(deposit) deposits.append(deposit) return deposits def get_queryset(self): """Retrieve queryset of deposits (with some optional filtering).""" params = self.request.query_params exclude_like = params.get("exclude") username = params.get("username") if username: deposits_qs = Deposit.objects.select_related("client").filter( client__username=username ) else: deposits_qs = Deposit.objects.all() if exclude_like: # sql injection: A priori, nothing to worry about, django does it for # queryset # https://docs.djangoproject.com/en/3.0/topics/security/#sql-injection-protection # noqa deposits_qs = deposits_qs.exclude(external_id__startswith=exclude_like) return deposits_qs.order_by("id") + + +def _deposit_search_query(search_value: str) -> Q: + fields = [f for f in Deposit._meta.fields if isinstance(f, (CharField, TextField))] + queries = [Q(**{f.name + "__icontains": search_value}) for f in fields] + search_query = Q() + for query in queries: + search_query = search_query | query + return search_query + + +@api_view() +@authentication_classes([]) +@permission_classes([AllowAny]) +def deposit_list_datatables(request: Request) -> JsonResponse: + """Special API view to list and filter deposits, produced responses are intended + to be consumed by datatables js framework used in deposits admin Web UI.""" + table_data: Dict[str, Any] = {} + table_data["draw"] = int(request.GET.get("draw", 1)) + try: + username = request.GET.get("username") + if username: + deposits = Deposit.objects.select_related("client").filter( + client__username=username + ) + else: + deposits = Deposit.objects.all() + + deposits_count = deposits.count() + search_value = request.GET.get("search[value]") + if search_value: + deposits = deposits.filter(_deposit_search_query(search_value)) + + exclude_pattern = request.GET.get("excludePattern") + if exclude_pattern: + deposits = deposits.exclude(_deposit_search_query(exclude_pattern)) + + column_order = request.GET.get("order[0][column]") + field_order = request.GET.get("columns[%s][name]" % column_order, "id") + order_dir = request.GET.get("order[0][dir]", "desc") + + if order_dir == "desc": + field_order = "-" + field_order + + deposits = deposits.order_by(field_order) + + length = int(request.GET.get("length", 10)) + page = int(request.GET.get("start", 0)) // length + 1 + paginator = Paginator(deposits, length) + + data = [ + DepositSerializer(_enrich_deposit_with_metadata(d)).data + for d in paginator.page(page).object_list + ] + + table_data["recordsTotal"] = deposits_count + table_data["recordsFiltered"] = deposits.count() + data_list = [] + for d in data: + data_dict = { + "id": d["id"], + "type": d["type"], + "external_id": d["external_id"], + "raw_metadata": d["raw_metadata"], + "reception_date": d["reception_date"], + "status": d["status"], + "status_detail": d["status_detail"], + "swhid": d["swhid"], + "swhid_context": d["swhid_context"], + } + provenance = None + raw_metadata = d["raw_metadata"] + # for meta deposit, the uri should be the url provenance + if raw_metadata and d["type"] == "meta": # metadata provenance + provenance = parse_swh_metadata_provenance( + ElementTree.fromstring(raw_metadata) + ) + # For code deposits the uri is the origin + # First, trying to determine it out of the raw metadata associated with the + # deposit + elif raw_metadata and d["type"] == "code": + create_origin_url, add_to_origin_url = parse_swh_deposit_origin( + ElementTree.fromstring(raw_metadata) + ) + provenance = create_origin_url or add_to_origin_url + + # For code deposits, if not provided, use the origin_url + if not provenance and d["type"] == "code": + if d["origin_url"]: + provenance = d["origin_url"] + + # If still not found, fallback using the swhid context + if not provenance and d["swhid_context"]: + swhid = QualifiedSWHID.from_string(d["swhid_context"]) + provenance = swhid.origin + + data_dict["uri"] = provenance # could be None + + data_list.append(data_dict) + + table_data["data"] = data_list + + except Exception as exc: + sentry_sdk.capture_exception(exc) + table_data[ + "error" + ] = "An error occurred while retrieving the list of deposits !" + if settings.DEBUG: + table_data["error"] += "\n" + str(exc) + + return JsonResponse(table_data) diff --git a/swh/deposit/api/private/urls.py b/swh/deposit/api/private/urls.py index e48290d6..f9f8357a 100644 --- a/swh/deposit/api/private/urls.py +++ b/swh/deposit/api/private/urls.py @@ -1,78 +1,84 @@ -# Copyright (C) 2017-2020 The Software Heritage developers +# Copyright (C) 2017-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from django.conf.urls import url from ...config import ( PRIVATE_CHECK_DEPOSIT, PRIVATE_GET_DEPOSIT_METADATA, PRIVATE_GET_RAW_CONTENT, PRIVATE_LIST_DEPOSITS, + PRIVATE_LIST_DEPOSITS_DATATABLES, PRIVATE_PUT_DEPOSIT, ) from .deposit_check import APIChecks -from .deposit_list import APIList +from .deposit_list import APIList, deposit_list_datatables from .deposit_read import APIReadArchives, APIReadMetadata from .deposit_update_status import APIUpdateStatus urlpatterns = [ # Retrieve deposit's raw archives' content # -> GET url( r"^(?P[^/]+)/(?P[^/]+)/raw/$", APIReadArchives.as_view(), name=PRIVATE_GET_RAW_CONTENT, ), # Update deposit's status # -> PUT url( r"^(?P[^/]+)/(?P[^/]+)/update/$", APIUpdateStatus.as_view(), name=PRIVATE_PUT_DEPOSIT, ), # Retrieve metadata information on a specific deposit # -> GET url( r"^(?P[^/]+)/(?P[^/]+)/meta/$", APIReadMetadata.as_view(), name=PRIVATE_GET_DEPOSIT_METADATA, ), # Check archive and metadata information on a specific deposit # -> GET url( r"^(?P[^/]+)/(?P[^/]+)/check/$", APIChecks.as_view(), name=PRIVATE_CHECK_DEPOSIT, ), # Retrieve deposit's raw archives' content # -> GET url( r"^(?P[^/]+)/raw/$", APIReadArchives.as_view(), name=PRIVATE_GET_RAW_CONTENT + "-nc", ), # Update deposit's status # -> PUT url( r"^(?P[^/]+)/update/$", APIUpdateStatus.as_view(), name=PRIVATE_PUT_DEPOSIT + "-nc", ), # Retrieve metadata information on a specific deposit # -> GET url( r"^(?P[^/]+)/meta/$", APIReadMetadata.as_view(), name=PRIVATE_GET_DEPOSIT_METADATA + "-nc", ), # Check archive and metadata information on a specific deposit # -> GET url( r"^(?P[^/]+)/check/$", APIChecks.as_view(), name=PRIVATE_CHECK_DEPOSIT + "-nc", ), url(r"^deposits/$", APIList.as_view(), name=PRIVATE_LIST_DEPOSITS), + url( + r"^deposits/datatables/$", + deposit_list_datatables, + name=PRIVATE_LIST_DEPOSITS_DATATABLES, + ), ] diff --git a/swh/deposit/config.py b/swh/deposit/config.py index e3bba9ff..585b3be4 100644 --- a/swh/deposit/config.py +++ b/swh/deposit/config.py @@ -1,121 +1,122 @@ -# Copyright (C) 2017-2020 The Software Heritage developers +# Copyright (C) 2017-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import os from typing import Any, Dict from swh.core import config from swh.deposit import __version__ from swh.model.model import MetadataAuthority, MetadataAuthorityType, MetadataFetcher from swh.scheduler import get_scheduler from swh.scheduler.interface import SchedulerInterface from swh.storage import get_storage from swh.storage.interface import StorageInterface # IRIs (Internationalized Resource identifier) sword 2.0 specified EDIT_IRI = "edit_iri" SE_IRI = "se_iri" EM_IRI = "em_iri" CONT_FILE_IRI = "cont_file_iri" SD_IRI = "servicedocument" COL_IRI = "upload" STATE_IRI = "state_iri" PRIVATE_GET_RAW_CONTENT = "private-download" PRIVATE_CHECK_DEPOSIT = "check-deposit" PRIVATE_PUT_DEPOSIT = "private-update" PRIVATE_GET_DEPOSIT_METADATA = "private-read" PRIVATE_LIST_DEPOSITS = "private-deposit-list" +PRIVATE_LIST_DEPOSITS_DATATABLES = "private-deposit-list-datatables" ARCHIVE_KEY = "archive" RAW_METADATA_KEY = "raw-metadata" ARCHIVE_TYPE = "archive" METADATA_TYPE = "metadata" AUTHORIZED_PLATFORMS = ["development", "production", "testing"] DEPOSIT_STATUS_REJECTED = "rejected" DEPOSIT_STATUS_PARTIAL = "partial" DEPOSIT_STATUS_DEPOSITED = "deposited" DEPOSIT_STATUS_VERIFIED = "verified" DEPOSIT_STATUS_LOAD_SUCCESS = "done" DEPOSIT_STATUS_LOAD_FAILURE = "failed" # Release author for deposit SWH_PERSON = { "name": "Software Heritage", "fullname": "Software Heritage", "email": "robot@softwareheritage.org", } DEFAULT_CONFIG = { "max_upload_size": 209715200, "checks": True, } def setup_django_for(platform=None, config_file=None): """Setup function for command line tools (swh.deposit.create_user) to initialize the needed db access. Note: Do not import any django related module prior to this function call. Otherwise, this will raise an django.core.exceptions.ImproperlyConfigured error message. Args: platform (str): the platform the scheduling is running config_file (str): Extra configuration file (typically for the production platform) Raises: ValueError in case of wrong platform inputs. """ if platform is not None: if platform not in AUTHORIZED_PLATFORMS: raise ValueError("Platform should be one of %s" % AUTHORIZED_PLATFORMS) if "DJANGO_SETTINGS_MODULE" not in os.environ: os.environ["DJANGO_SETTINGS_MODULE"] = "swh.deposit.settings.%s" % platform if config_file: os.environ.setdefault("SWH_CONFIG_FILENAME", config_file) import django django.setup() class APIConfig: """API Configuration centralized class. This loads explicitly the configuration file out of the SWH_CONFIG_FILENAME environment variable. """ def __init__(self): self.config: Dict[str, Any] = config.load_from_envvar(DEFAULT_CONFIG) self.scheduler: SchedulerInterface = get_scheduler(**self.config["scheduler"]) self.tool = { "name": "swh-deposit", "version": __version__, "configuration": {"sword_version": "2"}, } self.storage: StorageInterface = get_storage(**self.config["storage"]) self.storage_metadata: StorageInterface = get_storage( **self.config["storage_metadata"] ) def swh_deposit_authority(self): return MetadataAuthority( type=MetadataAuthorityType.REGISTRY, url=self.config["swh_authority_url"], ) def swh_deposit_fetcher(self): return MetadataFetcher( name=self.tool["name"], version=self.tool["version"], ) diff --git a/swh/deposit/tests/api/test_deposit_private_list.py b/swh/deposit/tests/api/test_deposit_private_list.py index 7017d8d4..4511a3cc 100644 --- a/swh/deposit/tests/api/test_deposit_private_list.py +++ b/swh/deposit/tests/api/test_deposit_private_list.py @@ -1,172 +1,402 @@ # Copyright (C) 2017-2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information +from django.http import QueryDict +from django.test import override_settings from django.urls import reverse_lazy as reverse +import pytest from rest_framework import status from swh.deposit.api.converters import convert_status_detail -from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS, PRIVATE_LIST_DEPOSITS +from swh.deposit.config import ( + DEPOSIT_STATUS_LOAD_SUCCESS, + PRIVATE_LIST_DEPOSITS, + PRIVATE_LIST_DEPOSITS_DATATABLES, +) from swh.deposit.models import DEPOSIT_CODE, DEPOSIT_METADATA_ONLY, DepositClient from swh.deposit.tests.conftest import internal_create_deposit STATUS_DETAIL = { "url": { "summary": "At least one compatible url field. Failed", "fields": ["testurl"], }, "metadata": [ { "summary": "Mandatory fields missing", "fields": ["9", 10, 1.212], }, ], "archive": [ { "summary": "Invalid archive", "fields": ["3"], }, { "summary": "Unsupported archive", "fields": [2], }, ], } +@pytest.fixture() +def partial_deposit_only_metadata(partial_deposit_only_metadata): + partial_deposit_only_metadata.type = DEPOSIT_METADATA_ONLY + partial_deposit_only_metadata.save() + return partial_deposit_only_metadata + + def test_deposit_list( partial_deposit_with_metadata, partial_deposit_only_metadata, partial_deposit, authenticated_client, ): """Deposit list api should return all deposits in a paginated way""" partial_deposit_with_metadata.status_detail = STATUS_DETAIL partial_deposit_with_metadata.save() deposit1 = partial_deposit_with_metadata deposit2 = partial_deposit_only_metadata - deposit2.type = DEPOSIT_METADATA_ONLY - deposit2.save() deposit3 = partial_deposit main_url = reverse(PRIVATE_LIST_DEPOSITS) url = f"{main_url}?page_size=1" response = authenticated_client.get(url) assert response.status_code == status.HTTP_200_OK data_p1 = response.json() assert data_p1["count"] == 3 # total nb of deposits expected_next_p1 = f"{main_url}?page=2&page_size=1" assert data_p1["next"].endswith(expected_next_p1) is True assert data_p1["previous"] is None assert len(data_p1["results"]) == 1 # page of size 1 deposit_d = data_p1["results"][0] assert deposit_d["id"] == deposit1.id assert deposit_d["status"] == deposit1.status expected_status_detail = convert_status_detail(STATUS_DETAIL) assert deposit_d["status_detail"] == expected_status_detail assert deposit_d["raw_metadata"] is not None assert deposit_d["type"] == DEPOSIT_CODE assert ( deposit_d["raw_metadata"] == deposit1.depositrequest_set.filter(type="metadata")[0].raw_metadata ) # then 2nd page response2 = authenticated_client.get(data_p1["next"]) assert response2.status_code == status.HTTP_200_OK data_p2 = response2.json() assert data_p2["count"] == 3 # total nb of deposits expected_next_p2 = f"{main_url}?page=3&page_size=1" assert data_p2["next"].endswith(expected_next_p2) assert data_p2["previous"].endswith(url) assert len(data_p2["results"]) == 1 # page of size 1 deposit2_d = data_p2["results"][0] assert deposit2_d["id"] == deposit2.id assert deposit2_d["status"] == deposit2.status assert deposit2_d["raw_metadata"] is not None assert deposit2_d["type"] == DEPOSIT_METADATA_ONLY assert ( deposit2_d["raw_metadata"] == deposit2.depositrequest_set.filter(type="metadata")[0].raw_metadata ) # then 3rd (and last) page response3 = authenticated_client.get(data_p2["next"]) assert response3.status_code == status.HTTP_200_OK data_p3 = response3.json() assert data_p3["count"] == 3 # total nb of deposits assert data_p3["next"] is None, "No more page beyond that point" assert data_p3["previous"] == data_p1["next"] assert len(data_p3["results"]) == 1 # page of size 1 deposit3_d = data_p3["results"][0] assert deposit3_d["id"] == deposit3.id assert deposit3_d["status"] == deposit3.status assert deposit3_d["type"] == DEPOSIT_CODE assert not deposit3.depositrequest_set.filter( type="metadata" ), "No metadata type request for that deposit" # hence no raw metadata set for that deposit assert deposit3_d["raw_metadata"] is None, "no raw metadata for that deposit" def test_deposit_list_exclude(partial_deposit, deposited_deposit, authenticated_client): """Exclusion pattern on external_id should be respected""" partial_deposit.status_detail = STATUS_DETAIL partial_deposit.save() main_url = reverse(PRIVATE_LIST_DEPOSITS) # Testing exclusion pattern exclude_pattern = "external-id" assert partial_deposit.external_id.startswith(exclude_pattern) assert deposited_deposit.external_id.startswith(exclude_pattern) url = f"{main_url}?page_size=1&exclude=external-id" response = authenticated_client.get(url) assert response.status_code == status.HTTP_200_OK data = response.json() assert data["count"] == 0 url = "%s?page_size=1&exclude=dummy" % main_url # that won't exclude anything response = authenticated_client.get(url) assert response.status_code == status.HTTP_200_OK data = response.json() assert data["count"] == 2 def test_deposit_list_for_username( authenticated_client, deposit_another_collection, completed_deposit, deposit_user, deposit_another_user, ): # create a new deposit with a user different from deposit_user, # the one that created completed_deposit internal_create_deposit( client=deposit_another_user, collection=deposit_another_collection, external_id="external-id-bar", status=DEPOSIT_STATUS_LOAD_SUCCESS, ) for user in (deposit_user, deposit_another_user): # check deposit filtering by username url = f"{reverse(PRIVATE_LIST_DEPOSITS)}?username={user.username}" json_response = authenticated_client.get(url).json() assert len(json_response["results"]) == 1 deposit_client = DepositClient.objects.all().get( id=json_response["results"][0]["client"] ) assert deposit_client.username == user.username + + +@pytest.fixture() +def deposits( + partial_deposit_with_metadata, + partial_deposit_only_metadata, + partial_deposit, + completed_deposit, + complete_deposit, +): + # to cover code extracting deposit provenance from swhid_context + complete_deposit.origin_url = None + complete_deposit.save() + + return [ + partial_deposit_with_metadata, + partial_deposit_only_metadata, + partial_deposit, + completed_deposit, + complete_deposit, + ] + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +def test_deposit_list_datatables_empty_query( + deposits, + authenticated_client, +): + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + deposits_dt_data = authenticated_client.get(url).json() + assert deposits_dt_data["draw"] == 1 + assert deposits_dt_data["recordsTotal"] == len(deposits) + assert deposits_dt_data["recordsFiltered"] == len(deposits) + assert len(deposits_dt_data["data"]) == len(deposits) + # by default, deposits are sorted by decreasing ids + assert [d["id"] for d in deposits_dt_data["data"]] == list( + reversed(sorted([d.id for d in deposits])) + ) + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +def test_deposit_list_datatables_pagination( + deposits, + authenticated_client, +): + + deposits_data = [] + for i in range(len(deposits)): + query_params = QueryDict(mutable=True) + query_params.update( + { + "draw": i + 1, + "length": 1, + "start": i, + } + ) + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + "?" + query_params.urlencode() + deposits_dt_data = authenticated_client.get(url).json() + deposits_data += deposits_dt_data["data"] + + assert [d["id"] for d in deposits_data] == list( + reversed(sorted([d.id for d in deposits])) + ) + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +@pytest.mark.parametrize("sort_direction", ["asc", "desc"]) +def test_deposit_list_datatables_ordering( + deposits, + authenticated_client, + sort_direction, +): + + deposits_date_sorted = list(sorted(deposits, key=lambda d: d.reception_date)) + + if sort_direction == "desc": + deposits_date_sorted = list(reversed(deposits_date_sorted)) + + query_params = QueryDict(mutable=True) + query_params.update( + { + "draw": 1, + "length": 10, + "start": 0, + "order[0][column]": 4, + "order[0][dir]": sort_direction, + "columns[4][name]": "reception_date", + } + ) + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + "?" + query_params.urlencode() + + deposits_dt_data = authenticated_client.get(url).json() + + reception_dates = [d["reception_date"] for d in deposits_dt_data["data"]] + expected_dates = [ + d.reception_date.isoformat().replace("+00:00", "Z") + for d in deposits_date_sorted + ] + + assert reception_dates == expected_dates + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +def test_deposit_list_datatables_search( + deposits, + authenticated_client, +): + + query_params = QueryDict(mutable=True) + query_params.update( + { + "draw": 1, + "length": 10, + "start": 0, + "search[value]": DEPOSIT_STATUS_LOAD_SUCCESS, + } + ) + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + "?" + query_params.urlencode() + + deposits_dt_data = authenticated_client.get(url).json() + + deposits_load_success = [ + d for d in deposits if d.status == DEPOSIT_STATUS_LOAD_SUCCESS + ] + + deposits_load_success = list( + reversed(sorted(deposits_load_success, key=lambda d: d.id)) + ) + + assert deposits_load_success + + assert [d.id for d in deposits_load_success] == [ + d["id"] for d in deposits_dt_data["data"] + ] + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +def test_deposit_list_datatables_exclude_pattern( + deposits, + authenticated_client, +): + + query_params = QueryDict(mutable=True) + query_params.update( + { + "draw": 1, + "length": 10, + "start": 0, + "excludePattern": DEPOSIT_STATUS_LOAD_SUCCESS, + } + ) + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + "?" + query_params.urlencode() + + deposits_dt_data = authenticated_client.get(url).json() + + deposits_load_not_success = [ + d for d in deposits if d.status != DEPOSIT_STATUS_LOAD_SUCCESS + ] + + deposits_load_not_success = list( + reversed(sorted(deposits_load_not_success, key=lambda d: d.id)) + ) + + assert deposits_load_not_success + + assert [d.id for d in deposits_load_not_success] == [ + d["id"] for d in deposits_dt_data["data"] + ] + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +def test_deposit_list_datatables_username( + completed_deposit, + deposit_user, + deposit_another_user, + deposit_another_collection, + authenticated_client, +): + + # create a new deposit with a user different from deposit_user, + # the one that created completed_deposit + completed_deposit_another_user = internal_create_deposit( + client=deposit_another_user, + collection=deposit_another_collection, + external_id="external-id-bar", + status=DEPOSIT_STATUS_LOAD_SUCCESS, + ) + + for user, deposit in ( + (deposit_user, completed_deposit), + (deposit_another_user, completed_deposit_another_user), + ): + query_params = QueryDict(mutable=True) + query_params["username"] = user.username + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + "?" + query_params.urlencode() + + deposits_dt_data = authenticated_client.get(url).json() + + assert len(deposits_dt_data["data"]) == 1 + assert deposits_dt_data["data"][0]["id"] == deposit.id + + +@pytest.mark.django_db(transaction=True, reset_sequences=True) +@override_settings(DEBUG=True) +def test_deposit_list_datatables_error( + deposits, + authenticated_client, + mocker, +): + parse_swh_metadata_provenance = mocker.patch( + "swh.deposit.api.private.deposit_list.parse_swh_metadata_provenance" + ) + error_message = "Error when parsing metadata" + parse_swh_metadata_provenance.side_effect = Exception(error_message) + url = reverse(PRIVATE_LIST_DEPOSITS_DATATABLES) + deposits_dt_data = authenticated_client.get(url).json() + assert "error" in deposits_dt_data + assert error_message in deposits_dt_data["error"]