Add trailing slash to AIP recovery destination

.pre-commit-config.yaml

@@ -45,3 +45,4 @@ repos:
     additional_dependencies:
     - types-requests
     - types-python-dateutil
+    - pytest

pyproject.toml

@@ -91,6 +91,12 @@ module = [
 ]
 ignore_errors = true
 
+[[tool.mypy.overrides]]
+module = [
+    "tests.integration.test_integration",
+]
+ignore_errors = false
+
 [tool.tox]
 legacy_tox_ini = """
     [tox]

storage_service/common/startup.py

@@ -2,6 +2,10 @@
 import logging
 import os
 import pathlib
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Union
 
 import django.core.exceptions
 from django.db import connection
@@ -13,12 +17,15 @@
 LOGGER = logging.getLogger(__name__)
 
 
-def startup():
+def startup(space_path: Optional[pathlib.Path] = None) -> None:
+    if space_path is None:
+        space_path = pathlib.Path(os.sep)
+
     LOGGER.info("Running startup")
 
     try:
         with PopulateLock():
-            populate_default_locations()
+            populate_default_locations(space_path)
     except PopulateLockError:
         LOGGER.warning("Another worker is initializing the database.")
 
@@ -57,16 +64,16 @@ def release(self):
             cursor.execute("SELECT RELEASE_LOCK(%s)", (self.name,))
 
 
-def populate_default_locations():
+def populate_default_locations(space_path: pathlib.Path) -> None:
     """Create default local filesystem space and its locations."""
 
     BASE_PATH = pathlib.Path("var") / "archivematica"
 
     try:
         space, space_created = locations_models.Space.objects.get_or_create(
             access_protocol=locations_models.Space.LOCAL_FILESYSTEM,
-            path=os.sep,
-            defaults={"staging_path": os.sep / BASE_PATH / "storage_service"},
+            path=str(space_path),
+            defaults={"staging_path": space_path / BASE_PATH / "storage_service"},
         )
         if space_created:
             locations_models.LocalFilesystem.objects.create(space=space)
@@ -75,7 +82,7 @@ def populate_default_locations():
         LOGGER.info("Multiple default Spaces exist, done default setup.")
         return
 
-    default_locations = [
+    default_locations: List[Dict[str, Union[str, pathlib.Path, bool, None]]] = [
         {
             "purpose": locations_models.Location.TRANSFER_SOURCE,
             "relative_path": "home",

storage_service/common/utils.py

@@ -12,6 +12,8 @@
 import uuid
 from collections import deque
 from collections import namedtuple
+from typing import Any
+from typing import Union
 
 from administration import models
 from django import http
@@ -654,7 +656,9 @@ def extract_tar(tarpath):
 # ########### OTHER ############
 
 
-def generate_checksum(file_path, checksum_type="md5"):
+def generate_checksum(
+    file_path: Union[str, pathlib.Path], checksum_type: str = "md5"
+) -> Any:
     """
     Returns checksum object for `file_path` using `checksum_type`.
 

storage_service/locations/api/resources.py

@@ -1500,7 +1500,7 @@ def aip_store_callback_request(self, request, bundle, **kwargs):
 
             package_dir = os.path.join(tmpdir, basedir)
         else:
-            package_dir = package.full_path()
+            package_dir = package.full_path
             tmpdir = None
 
         safe_files = ("bag-info.txt", "manifest-sha512.txt", "bagit.txt")

storage_service/locations/models/package.py

@@ -583,7 +583,7 @@ def recover_aip(self, origin_location, origin_path):
         # Copy recovery files to storage service staging
         source_path = os.path.join(temp_aip.current_location.relative_path, origin_path)
         destination_path = os.path.join(
-            self.current_location.relative_path, os.path.dirname(self.current_path)
+            self.current_location.relative_path, os.path.dirname(self.current_path), ""
         )
 
         origin_space.move_to_storage_service(

storage_service/locations/views.py

@@ -169,12 +169,12 @@ def execution_logic(aip):
         ).location
 
         try:
-            (success, _, message) = aip.recover_aip(
+            (success, failures, message) = aip.recover_aip(
                 recover_location, os.path.basename(aip.current_path)
             )
         except StorageException:
             recover_path = os.path.join(
-                recover_location.full_path(), os.path.basename(aip.full_path())
+                recover_location.full_path, os.path.basename(aip.full_path)
             )
             message = _("error accessing restore files at %(path)s") % {
                 "path": recover_path