✨(backend) reconciliation requests update link traces and doc favorites

Adds management of link traces and documents favorites in the reconciliation
requests.

src/backend/core/models.py

@@ -383,8 +383,9 @@ class UserReconciliation(BaseModel):
         """
         Process the reconciliation request as a transaction.
 
-        Its action is threefold:
         - Transfer document accesses from inactive to active user, updating roles as needed.
+        - Transfer document favorites from inactive to active user.
+        - Transfer link traces from inactive to active user.
         - Activate the active user and deactivate the inactive user.
         - Update the reconciliation entry itself.
         """
@@ -393,6 +394,11 @@ class UserReconciliation(BaseModel):
         updated_accesses, removed_accesses = (
             self.prepare_documentaccess_reconciliation()
         )
+        updated_linktraces, removed_linktraces = self.prepare_linktrace_reconciliation()
+        update_favorites, removed_favorites = (
+            self.prepare_document_favorite_reconciliation()
+        )
+
         self.active_user.is_active = True
         self.inactive_user.is_active = False
 
@@ -400,9 +406,19 @@ class UserReconciliation(BaseModel):
         DocumentAccess.objects.bulk_update(updated_accesses, ["user", "role"])
 
         if removed_accesses:
-            ids_to_delete = [rd.id for rd in removed_accesses]
+            ids_to_delete = [entry.id for entry in removed_accesses]
             DocumentAccess.objects.filter(id__in=ids_to_delete).delete()
 
+        DocumentFavorite.objects.bulk_update(update_favorites, ["user"])
+        if removed_favorites:
+            ids_to_delete = [entry.id for entry in removed_favorites]
+            DocumentFavorite.objects.filter(id__in=ids_to_delete).delete()
+
+        LinkTrace.objects.bulk_update(updated_linktraces, ["user"])
+        if removed_linktraces:
+            ids_to_delete = [entry.id for entry in removed_linktraces]
+            LinkTrace.objects.filter(id__in=ids_to_delete).delete()
+
         User.objects.bulk_update([self.active_user, self.inactive_user], ["is_active"])
 
         # Wrap up the reconciliation entry
@@ -423,9 +439,11 @@ class UserReconciliation(BaseModel):
         inactive_accesses = DocumentAccess.objects.filter(user=self.inactive_user)
 
         # Check documents where the active user already has access
-        documents_with_both_users = inactive_accesses.values_list("document", flat=True)
+        inactive_accesses_documents = inactive_accesses.values_list(
+            "document", flat=True
+        )
         existing_accesses = DocumentAccess.objects.filter(user=self.active_user).filter(
-            document__in=documents_with_both_users
+            document__in=inactive_accesses_documents
         )
         existing_roles_per_doc = dict(existing_accesses.values_list("document", "role"))
 
@@ -445,6 +463,50 @@ class UserReconciliation(BaseModel):
 
         return updated_accesses, removed_accesses
 
+    def prepare_document_favorite_reconciliation(self):
+        """
+        Prepare the reconciliation by transferring document favorites from the inactive user
+        to the active user.
+        """
+        updated_favorites = []
+        removed_favorites = []
+
+        existing_favorites = DocumentFavorite.objects.filter(user=self.active_user)
+        existing_favorite_doc_ids = set(
+            existing_favorites.values_list("document_id", flat=True)
+        )
+
+        inactive_favorites = DocumentFavorite.objects.filter(user=self.inactive_user)
+
+        for entry in inactive_favorites:
+            if entry.document_id in existing_favorite_doc_ids:
+                removed_favorites.append(entry)
+            else:
+                entry.user = self.active_user
+                updated_favorites.append(entry)
+
+        return updated_favorites, removed_favorites
+
+    def prepare_linktrace_reconciliation(self):
+        """
+        Prepare the reconciliation by transferring link traces from the inactive user
+        to the active user.
+        """
+        updated_linktraces = []
+        removed_linktraces = []
+
+        existing_linktraces = LinkTrace.objects.filter(user=self.active_user)
+        inactive_linktraces = LinkTrace.objects.filter(user=self.inactive_user)
+
+        for entry in inactive_linktraces:
+            if existing_linktraces.filter(document=entry.document).exists():
+                removed_linktraces.append(entry)
+            else:
+                entry.user = self.active_user
+                updated_linktraces.append(entry)
+
+        return updated_linktraces, removed_linktraces
+
     def send_reconciliation_confirm_email(
         self, user, user_type, confirmation_id, language=None
     ):

src/backend/core/tests/test_models_user_reconciliation.py

@@ -389,10 +389,10 @@ def test_user_reconciliation_only_starts_if_checks_are_made(
     assert (user_1.is_active, user_2.is_active) == users_active_before
 
 
-def test_process_documentaccess_reconciliation(
+def test_process_reconciliation_updates_accesses(
     user_reconciliation_users_and_docs,
 ):
-    """Use the fixture to verify accesses are consolidated on the active user."""
+    """Test that accesses are consolidated on the active user."""
     user_1, user_2, userdocs_u1, userdocs_u2 = user_reconciliation_users_and_docs
 
     u1_2 = userdocs_u1[2]
@@ -456,3 +456,147 @@ def test_process_documentaccess_reconciliation(
     email_content = " ".join(email.body.split())
 
     assert "Your accounts have been merged" in email_content
+
+
+def test_process_reconciliation_updates_linktraces(
+    user_reconciliation_users_and_docs,
+):
+    """Test that linktraces are consolidated on the active user."""
+    user_1, user_2, userdocs_u1, userdocs_u2 = user_reconciliation_users_and_docs
+
+    u1_2 = userdocs_u1[2]
+    u1_5 = userdocs_u1[5]
+    u2doc1 = userdocs_u2[1].document
+    u2doc5 = userdocs_u2[5].document
+
+    doc_both = u1_2.document
+    models.LinkTrace.objects.create(document=doc_both, user=user_1)
+    models.LinkTrace.objects.create(document=doc_both, user=user_2)
+
+    doc_inactive_only = userdocs_u2[4].document
+    models.LinkTrace.objects.create(
+        document=doc_inactive_only, user=user_2, is_masked=True
+    )
+
+    doc_active_only = userdocs_u1[4].document
+    models.LinkTrace.objects.create(document=doc_active_only, user=user_1)
+
+    rec = models.UserReconciliation.objects.create(
+        active_email=user_1.email,
+        inactive_email=user_2.email,
+        active_user=user_1,
+        inactive_user=user_2,
+        active_email_checked=True,
+        inactive_email_checked=True,
+        status="ready",
+    )
+
+    qs = models.UserReconciliation.objects.filter(id=rec.id)
+    process_reconciliation(None, None, qs)
+
+    rec.refresh_from_db()
+    user_1.refresh_from_db()
+    user_2.refresh_from_db()
+    u1_2.refresh_from_db(
+        from_queryset=models.DocumentAccess.objects.select_for_update()
+    )
+    u1_5.refresh_from_db(
+        from_queryset=models.DocumentAccess.objects.select_for_update()
+    )
+
+    # Inactive user should have no linktraces
+    assert models.LinkTrace.objects.filter(user=user_2).count() == 0
+
+    # doc_both should have a single LinkTrace owned by the active user
+    assert (
+        models.LinkTrace.objects.filter(user=user_1, document=doc_both).exists() is True
+    )
+    assert models.LinkTrace.objects.filter(user=user_1, document=doc_both).count() == 1
+    assert (
+        models.LinkTrace.objects.filter(user=user_2, document=doc_both).exists()
+        is False
+    )
+
+    # doc_inactive_only should now be linked to active user and preserve is_masked
+    lt = models.LinkTrace.objects.filter(
+        user=user_1, document=doc_inactive_only
+    ).first()
+    assert lt is not None
+    assert lt.is_masked is True
+
+    # doc_active_only should still belong to active user
+    assert models.LinkTrace.objects.filter(
+        user=user_1, document=doc_active_only
+    ).exists()
+
+
+def test_process_reconciliation_updates_favorites(
+    user_reconciliation_users_and_docs,
+):
+    """Test that favorites are consolidated on the active user."""
+    user_1, user_2, userdocs_u1, userdocs_u2 = user_reconciliation_users_and_docs
+
+    u1_2 = userdocs_u1[2]
+    u1_5 = userdocs_u1[5]
+    u2doc1 = userdocs_u2[1].document
+    u2doc5 = userdocs_u2[5].document
+
+    doc_both = u1_2.document
+    models.DocumentFavorite.objects.create(document=doc_both, user=user_1)
+    models.DocumentFavorite.objects.create(document=doc_both, user=user_2)
+
+    doc_inactive_only = userdocs_u2[4].document
+    models.DocumentFavorite.objects.create(document=doc_inactive_only, user=user_2)
+
+    doc_active_only = userdocs_u1[4].document
+    models.DocumentFavorite.objects.create(document=doc_active_only, user=user_1)
+
+    rec = models.UserReconciliation.objects.create(
+        active_email=user_1.email,
+        inactive_email=user_2.email,
+        active_user=user_1,
+        inactive_user=user_2,
+        active_email_checked=True,
+        inactive_email_checked=True,
+        status="ready",
+    )
+
+    qs = models.UserReconciliation.objects.filter(id=rec.id)
+    process_reconciliation(None, None, qs)
+
+    rec.refresh_from_db()
+    user_1.refresh_from_db()
+    user_2.refresh_from_db()
+    u1_2.refresh_from_db(
+        from_queryset=models.DocumentAccess.objects.select_for_update()
+    )
+    u1_5.refresh_from_db(
+        from_queryset=models.DocumentAccess.objects.select_for_update()
+    )
+
+    # Inactive user should have no document favorites
+    assert models.DocumentFavorite.objects.filter(user=user_2).count() == 0
+
+    # doc_both should have a single DocumentFavorite owned by the active user
+    assert (
+        models.DocumentFavorite.objects.filter(user=user_1, document=doc_both).exists()
+        is True
+    )
+    assert (
+        models.DocumentFavorite.objects.filter(user=user_1, document=doc_both).count()
+        == 1
+    )
+    assert (
+        models.DocumentFavorite.objects.filter(user=user_2, document=doc_both).exists()
+        is False
+    )
+
+    # doc_inactive_only should now be linked to active user
+    df = models.DocumentFavorite.objects.filter(
+        user=user_1, document=doc_inactive_only
+    ).first()
+
+    # doc_active_only should still belong to active user
+    assert models.DocumentFavorite.objects.filter(
+        user=user_1, document=doc_active_only
+    ).exists()