Add management command to recover deleted attachments

onadata/apps/logger/management/commands/recover_deleted_attachments.py

@@ -0,0 +1,55 @@
+"""
+Module containing the recover_deleted_attachments management command.
+
+Used to recover attachments that were accidentally deleted within the system
+but are still required/present within the submission XML
+
+Sample usage: python manage.py recover_deleted_attachments --form 1
+"""
+from django.core.management.base import BaseCommand
+
+from onadata.apps.logger.models import Instance
+
+
+def recover_deleted_attachments(form_id: str, stdout=None):
+    """
+    Recovers attachments that were accidentally soft-deleted
+
+    :param: (str) form_id: Unique identifier for an XForm object
+    :param: (sys.stdout) stdout: Python standard output. Default: None
+    """
+    instances = Instance.objects.filter(
+        xform__id=form_id, deleted_at__isnull=True)
+    for instance in instances:
+        expected_attachments = instance.get_expected_media()
+        if not instance.attachments.filter(
+                deleted_at__isnull=True).count() == len(expected_attachments):
+            attachments_to_recover = instance.attachments.filter(
+                deleted_at__isnull=False,
+                name__in=instance.get_expected_media())
+            for attachment in attachments_to_recover:
+                attachment.deleted_at = None
+                attachment.deleted_by = None
+                attachment.save()
+
+                if stdout:
+                    stdout.write(
+                        f'Recovered {attachment.name} ID: {attachment.id}')
+            # Regenerate instance JSON
+            instance.json = instance.get_full_dict(load_existing=False)
+            instance.save()
+
+
+class Command(BaseCommand):
+    """
+    Management command used to recover wrongfully deleted
+    attachments.
+    """
+    help = 'Restore wrongly deleted attachments'
+
+    def add_arguments(self, parser):
+        parser.add_argument('-f', '--form', dest='form_id', type=int)
+
+    def handle(self, *args, **options):
+        form_id = options.get('form_id')
+        recover_deleted_attachments(form_id, self.stdout)

onadata/apps/logger/tests/management/commands/test_recover_deleted_attachments.py

@@ -0,0 +1,68 @@
+from io import BytesIO
+from datetime import datetime
+
+from django.conf import settings
+
+from onadata.apps.main.tests.test_base import TestBase
+from onadata.apps.logger.import_tools import django_file
+from onadata.apps.logger.management.commands.recover_deleted_attachments \
+    import recover_deleted_attachments
+from onadata.libs.utils.logger_tools import create_instance
+
+
+class TestRecoverDeletedAttachments(TestBase):
+    def test_recovers_wrongly_deleted_attachments(self):
+        """
+        Test that the command recovers the correct
+        attachment
+        """
+        md = """
+        | survey |       |        |       |
+        |        | type  | name   | label |
+        |        | file  | file   | File  |
+        |        | image | image  | Image |
+        """
+        self._create_user_and_login()
+        self.xform = self._publish_markdown(md, self.user)
+
+        xml_string = f"""
+        <data id="{self.xform.id_string}">
+            <meta>
+                <instanceID>uuid:UJ5jz4EszdgH8uhy8nss1AsKaqBPO5VN7</instanceID>
+            </meta>
+            <file>Health_2011_03_13.xml_2011-03-15_20-30-28.xml</file>
+            <image>1300221157303.jpg</image>
+        </data>
+        """
+        media_root = (f'{settings.PROJECT_ROOT}/apps/logger/tests/Health'
+                      '_2011_03_13.xml_2011-03-15_20-30-28/')
+        image_media = django_file(
+            path=f'{media_root}1300221157303.jpg', field_name='image',
+            content_type='image/jpeg')
+        file_media = django_file(
+            path=f'{media_root}Health_2011_03_13.xml_2011-03-15_20-30-28.xml',
+            field_name='file', content_type='text/xml')
+        instance = create_instance(
+            self.user.username,
+            BytesIO(xml_string.strip().encode('utf-8')),
+            media_files=[file_media, image_media])
+        self.assertEqual(
+            instance.attachments.filter(deleted_at__isnull=True).count(), 2)
+        attachment = instance.attachments.first()
+
+        # Soft delete attachment
+        attachment.deleted_at = datetime.now()
+        attachment.deleted_by = self.user
+        attachment.save()
+
+        self.assertEqual(
+            instance.attachments.filter(deleted_at__isnull=True).count(), 1)
+
+        # Attempt recovery of attachment
+        recover_deleted_attachments(form_id=instance.xform.id)
+
+        self.assertEqual(
+            instance.attachments.filter(deleted_at__isnull=True).count(), 2)
+        attachment.refresh_from_db()
+        self.assertIsNone(attachment.deleted_at)
+        self.assertIsNone(attachment.deleted_by)