From 4b9eddfe44a93d575b40915d9ecc41fd14c4eb80 Mon Sep 17 00:00:00 2001 From: "Douglas Cerna (Soy Douglas)" Date: Tue, 7 May 2024 17:50:58 +0200 Subject: [PATCH] Fix SIP arrangement from ArchivesSpace pane --- src/dashboard/src/components/access/urls.py | 1 + src/dashboard/src/components/access/views.py | 4 +- tests/dashboard/test_access.py | 207 +++++++++++++++++++ 3 files changed, 210 insertions(+), 2 deletions(-) diff --git a/src/dashboard/src/components/access/urls.py b/src/dashboard/src/components/access/urls.py index 3cb41408ff..5e9937fa16 100644 --- a/src/dashboard/src/components/access/urls.py +++ b/src/dashboard/src/components/access/urls.py @@ -19,6 +19,7 @@ re_path( r"archivesspace/(?P[A-Za-z0-9-_]+)/copy_from_arrange/$", views.access_arrange_start_sip, + name="access_arrange_start_sip", ), re_path( r"archivesspace/(?P[A-Za-z0-9-_]+)/create_directory_within_arrange/$", diff --git a/src/dashboard/src/components/access/views.py b/src/dashboard/src/components/access/views.py index c47c873c42..e624f3ea09 100644 --- a/src/dashboard/src/components/access/views.py +++ b/src/dashboard/src/components/access/views.py @@ -97,7 +97,7 @@ def _get_sip(func): @wraps(func) def wrapper(request, mapping): arrange = SIPArrange.objects.get( - arrange_path=os.path.join(mapping.arrange_path, "") + arrange_path=os.path.join(mapping.arrange_path, "").encode() ) if arrange.sip is None: arrange.sip = SIP.objects.create(uuid=(uuid.uuid4()), currentpath=None) @@ -421,7 +421,7 @@ def access_arrange_start_sip(client, request, mapping): """ try: arrange = SIPArrange.objects.get( - arrange_path=os.path.join(mapping.arrange_path, "") + arrange_path=os.path.join(mapping.arrange_path, "").encode() ) except SIPArrange.DoesNotExist: response = { diff --git a/tests/dashboard/test_access.py b/tests/dashboard/test_access.py index abc5497e4f..718b6fb37a 100644 --- a/tests/dashboard/test_access.py +++ b/tests/dashboard/test_access.py @@ -1,7 +1,9 @@ import json import pathlib +from unittest import mock import archivematicaFunctions +import pytest from components import helpers from django.test import TestCase from django.test.client import Client @@ -56,3 +58,208 @@ def test_arrange_contents(self): in response_dict["entries"] ) assert len(response_dict["entries"]) == 1 + + +@pytest.mark.django_db +@pytest.fixture +def dashboard_uuid(): + helpers.set_setting("dashboard_uuid", "test-uuid") + + +def _encode_record_id(record_id): + return record_id.replace("/", "") + + +@pytest.mark.django_db +def test_access_arrange_start_sip_fails_if_arrange_mapping_does_not_exist( + dashboard_uuid, admin_client +): + record_id = "/repositories/2/archival_objects/1" + + response = admin_client.get( + reverse( + "access:access_arrange_start_sip", + kwargs={"record_id": _encode_record_id(record_id)}, + ) + ) + assert response.status_code == 404 + + result = json.loads(response.content.decode()) + assert result == { + "message": f"No SIP Arrange mapping exists for record {_encode_record_id(record_id)}", + "success": False, + } + + +@pytest.mark.django_db +@mock.patch("components.access.views.get_as_system_client") +def test_access_arrange_start_sip_fails_if_arrange_does_not_exist( + get_as_system_client, dashboard_uuid, admin_client +): + record_id = "/repositories/2/archival_objects/1" + models.SIPArrangeAccessMapping.objects.create( + arrange_path="/foobar", + system=models.SIPArrangeAccessMapping.ARCHIVESSPACE, + identifier=_encode_record_id(record_id), + ) + + response = admin_client.get( + reverse( + "access:access_arrange_start_sip", + kwargs={"record_id": _encode_record_id(record_id)}, + ) + ) + assert response.status_code == 404 + + result = json.loads(response.content.decode()) + assert result == { + "message": f"No SIP Arrange object exists for record {_encode_record_id(record_id)}", + "success": False, + } + + +@pytest.mark.django_db +@mock.patch( + "components.access.views.get_as_system_client", + return_value=mock.Mock( + **{ + "get_record.side_effect": [ + # Archival object. + {"resource": {"ref": "/repositories/2/resources/10"}}, + # Resource. + {"linked_agents": []}, + ] + } + ), +) +def test_access_arrange_start_sip_fails_if_resource_creators_cannot_be_fetched( + get_as_system_client, dashboard_uuid, admin_client +): + record_id = "/repositories/2/archival_objects/1" + mapping = models.SIPArrangeAccessMapping.objects.create( + arrange_path="/foobar", + system=models.SIPArrangeAccessMapping.ARCHIVESSPACE, + identifier=_encode_record_id(record_id), + ) + models.SIPArrange.objects.create( + arrange_path=f"{pathlib.Path(mapping.arrange_path)}/".encode() + ) + + response = admin_client.get( + reverse( + "access:access_arrange_start_sip", + kwargs={"record_id": _encode_record_id(record_id)}, + ) + ) + assert response.status_code == 502 + + result = json.loads(response.content.decode()) + assert result == { + "message": "Unable to fetch ArchivesSpace creator", + "success": False, + } + + +@pytest.mark.django_db +@mock.patch("components.filesystem_ajax.views.copy_from_arrange_to_completed_common") +@mock.patch("components.access.views.get_as_system_client") +def test_access_arrange_start_sip( + get_as_system_client, + copy_from_arrange_to_completed_common, + dashboard_uuid, + admin_client, + caplog, +): + # Mock expected responses from ArchivesSpace. + archival_object = { + "resource": {"ref": "/repositories/2/resources/10"}, + "notes": [{"type": "odd", "subnotes": [{"content": "A note"}]}], + "display_string": "Object, 2024", + "parent": {"ref": "/repositories/2/resources/1"}, + } + resource = {"linked_agents": [{"ref": "/agents/people/3", "role": "creator"}]} + creator = {"display_name": {"sort_name": "Foo, Bar"}} + parent = {"title": "Parent resource"} + digital_object = {"id": "do"} + get_as_system_client.return_value = mock.Mock( + **{ + "get_record.side_effect": [archival_object, resource, creator, parent], + "add_digital_object.side_effect": [ + digital_object, + ], + } + ) + + # Mock interaction with copy_from_arrange_to_completed_common. + sip = models.SIP.objects.create() + expected_status_code = 201 + expected_response_content = {"message": "SIP created.", "sip_uuid": str(sip.uuid)} + copy_from_arrange_to_completed_common.return_value = ( + expected_status_code, + expected_response_content, + ) + + # Set database fixtures. + record_id = "/repositories/2/archival_objects/1" + mapping = models.SIPArrangeAccessMapping.objects.create( + arrange_path="/foobar", + system=models.SIPArrangeAccessMapping.ARCHIVESSPACE, + identifier=_encode_record_id(record_id), + ) + models.SIPArrange.objects.create( + arrange_path=f"{pathlib.Path(mapping.arrange_path)}/".encode() + ) + models.ArchivesSpaceDigitalObject.objects.create( + resourceid=_encode_record_id(record_id), started=False + ) + + response = admin_client.post( + reverse( + "access:access_arrange_start_sip", + kwargs={"record_id": _encode_record_id(record_id)}, + ), + data=json.dumps({}), + content_type="application/json", + ) + assert response.status_code == expected_status_code + + result = json.loads(response.content.decode()) + assert result == expected_response_content + + assert models.DublinCore.objects.count() == 1 + assert ( + models.DublinCore.objects.filter( + metadataappliestotype_id=models.MetadataAppliesToType.SIP_TYPE, + metadataappliestoidentifier=sip.uuid, + title=archival_object["display_string"], + creator=creator["display_name"]["sort_name"], + description=archival_object["notes"][0]["subnotes"][0]["content"], + rights=" ".join( + [ + "This content may be under copyright.", + "Researchers are responsible for determining the", + "appropriate use or reuse of materials.", + ] + ), + relation=parent["title"], + ).count() + == 1 + ) + + assert models.ArchivesSpaceDigitalObject.objects.count() == 1 + assert ( + models.ArchivesSpaceDigitalObject.objects.filter( + resourceid=_encode_record_id(record_id), + started=True, + remoteid=digital_object["id"], + sip_id=expected_response_content["sip_uuid"], + ).count() + == 1 + ) + + assert [r.message for r in caplog.records] == [ + f"archival object {archival_object}", + f"resource {resource}", + f"creator {creator}", + f"New SIP UUID {sip.uuid}", + ]