Add management command to replace form ID root node

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

@@ -0,0 +1,88 @@
+"""
+Management command used to replace the root node of an Instance when
+the root node is the XForm ID
+Example usage:
+    python manage.py replace_form_id_root_node -c -i 1,2,3
+"""
+import re
+from hashlib import sha256
+
+from django.core.management.base import BaseCommand, CommandError
+from django.utils import timezone
+from django.utils.translation import gettext as _
+
+from onadata.apps.logger.models import Instance
+from onadata.apps.logger.models.instance import InstanceHistory
+
+
+def replace_form_id_with_correct_root_node(
+        inst_id: int, root: str = None, commit: bool = False) -> str:
+    inst: Instance = Instance.objects.get(id=inst_id, deleted_at__isnull=True)
+    initial_xml = inst.xml
+    form_id = re.escape(inst.xform.id_string)
+    if not root:
+        root = inst.xform.survey.name
+
+    opening_tag_regex = f"<{form_id}"
+    closing_tag_regex = f"</{form_id}>"
+    edited_xml = re.sub(opening_tag_regex, f'<{root}', initial_xml)
+    edited_xml = re.sub(closing_tag_regex, f'</{root}>', edited_xml)
+
+    if commit:
+        last_edited = timezone.now()
+        history = InstanceHistory.objects.create(
+            xml=initial_xml,
+            checksum=inst.checksum,
+            xform_instance=inst,
+        )
+        inst.last_edited = last_edited
+        inst.checksum = sha256(edited_xml.encode('utf-8')).hexdigest()
+        inst.xml = edited_xml
+        inst.save()
+        return f"Modified Instance ID {inst.id} - History object {history.id}"
+    else:
+        return edited_xml
+
+
+class Command(BaseCommand):
+    help = _("Replaces form ID String with 'data' for an instances root node")
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--instance-ids',
+            '-i',
+            dest='instance_ids',
+            help='Comma-separated list of instance ids.'
+        )
+        parser.add_argument(
+            '--commit-changes',
+            '-c',
+            action='store_true',
+            dest='commit',
+            default=False,
+            help='Save XML changes'
+        )
+        parser.add_argument(
+            '--root-node',
+            '-r',
+            dest='root',
+            default=None,
+            help='Default root node name to replace the form ID with'
+        )
+
+    def handle(self, *args, **options):
+        instance_ids = options.get('instance_ids').split(',')
+        commit = options.get('commit')
+        root = options.get('root')
+
+        if not instance_ids:
+            raise CommandError('No instance id provided.')
+
+        for inst_id in instance_ids:
+            try:
+                msg = replace_form_id_with_correct_root_node(
+                    inst_id, root=root, commit=commit)
+            except Instance.DoesNotExist:
+                msg = f"Instance with ID {inst_id} does not exist"
+
+            self.stdout.write(msg)

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

@@ -0,0 +1,67 @@
+"""
+Module containing the tests for the replace_form_id_root_node
+management command
+"""
+from io import BytesIO
+
+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.replace_form_id_root_node \
+    import replace_form_id_with_correct_root_node
+from onadata.libs.utils.logger_tools import create_instance
+
+
+class TestReplaceFormIDRootNodeCommand(TestBase):
+    """TestReplaceFormIDRootNodeCommand Class"""
+
+    def test_replaces_form_id_root_node(self):
+        """
+        Test that the command correctly replaces the form ID
+        """
+        md = """
+        | survey |       |        |       |
+        |        | type  | name   | label |
+        |        | file  | file   | File  |
+        |        | image | image  | Image |
+        """
+        self._create_user_and_login()
+        xform = self._publish_markdown(md, self.user)
+        id_string = xform.id_string
+
+        xml_string = f"""
+        <{id_string} id="{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>
+        </{id_string}>
+        """
+        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])
+
+        # Attempt replacement of root node name
+        replace_form_id_with_correct_root_node(
+            inst_id=instance.id, root='data', commit=True)
+        instance.refresh_from_db()
+
+        expected_xml = f"""<data id="{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>"""
+        self.assertEqual(instance.xml, expected_xml)