Update tests to use deployment xform

kobo/apps/project_ownership/tests/api/v2/test_api.py

@@ -502,8 +502,8 @@ def test_data_accessible_to_new_user(self):
             assert attachment['filename'].startswith('anotheruser/')
 
         # Get the mongo_uuid for the transferred asset (XForm)
-        xform = XForm.objects.get(kpi_asset_uid=self.asset.uid)
-        mongo_uuid = xform.mongo_uuid
+        self.asset.deployment.xform.refresh_from_db()
+        mongo_uuid = self.asset.deployment.xform.mongo_uuid
 
         assert (
             settings.MONGO_DB.instances.count_documents(
@@ -556,9 +556,9 @@ def test_mongo_uuid_after_transfer(self):
             response = self.client.post(self.invite_url, data=payload, format='json')
         assert response.status_code == status.HTTP_201_CREATED
 
-        # Retrieve the mongo_uuid for the transferred asset (XForm)
-        xform = XForm.objects.get(kpi_asset_uid=self.asset.uid)
-        mongo_uuid = xform.mongo_uuid
+        # Get the mongo_uuid for the transferred asset (XForm)
+        self.asset.deployment.xform.refresh_from_db()
+        mongo_uuid = self.asset.deployment.xform.mongo_uuid
 
         # Verify MongoDB now uses mongo_uuid as the identifier
         assert (

kpi/deployment_backends/openrosa_backend.py

@@ -63,6 +63,7 @@
     SubmissionNotFoundException,
     XPathNotFoundException,
 )
+from kpi.fields import KpiUidField
 from kpi.interfaces.sync_backend_media import SyncBackendMediaInterface
 from kpi.models.asset_file import AssetFile
 from kpi.models.object_permission import ObjectPermission
@@ -967,6 +968,14 @@ def set_enketo_open_rosa_server(self, require_auth: bool, enketo_id: str = None)
             server_url,
         )
 
+    def set_mongo_uuid(self):
+        """
+        Set the `mongo_uuid` for the associated XForm if it's not already set.
+        """
+        if not self.xform.mongo_uuid:
+            self.xform.mongo_uuid = KpiUidField.generate_unique_id()
+            self.xform.save(update_fields=['mongo_uuid'])
+
     def set_validation_status(
         self,
         submission_id: int,
@@ -1243,20 +1252,6 @@ def xform(self):
         self._xform = xform
         return self._xform
 
-    def generate_unique_key(self, length=20):
-
-        characters = string.ascii_letters + string.digits
-        # Generate a secure key with the specified length
-        return ''.join(secrets.choice(characters) for _ in range(length))
-
-    def set_mongo_uuid(self):
-        """
-        Set the `mongo_uuid` for the associated XForm if it's not already set.
-        """
-        if not self.xform.mongo_uuid:
-            self.xform.mongo_uuid = self.generate_unique_key()
-            self.xform.save(update_fields=['mongo_uuid'])
-
     @property
     def xform_id(self):
         return self.xform.pk