initial fix

src/registrar/tests/test_admin.py

@@ -2734,7 +2734,7 @@ def test_transfer_user_transfers_user_portfolio_roles(self):
 
     @less_console_noise_decorator
     def test_transfer_user_transfers_user_portfolio_roles_no_error_when_duplicates(self):
-        """Assert that duplicate portfolio user roles do not throw errorsd"""
+        """Assert that duplicate portfolio user roles do not throw errors"""
         portfolio1 = Portfolio.objects.create(organization_name="Hotel California", creator=self.user2)
         UserPortfolioPermission.objects.create(
             user=self.user1, portfolio=portfolio1, roles=[UserPortfolioRoleChoices.ORGANIZATION_ADMIN]
@@ -2759,7 +2759,7 @@ def test_transfer_user_transfers_user_portfolio_roles_no_error_when_duplicates(s
 
             messages.error.assert_not_called()
 
-    @less_console_noise_decorator
+    # @less_console_noise_decorator
     def test_transfer_user_transfers_domain_request_creator_and_investigator(self):
         """Assert that domain request fields get transferred"""
         domain_request = completed_domain_request(user=self.user2, name="wasteland.gov", investigator=self.user2)
@@ -2776,6 +2776,7 @@ def test_transfer_user_transfers_domain_request_creator_and_investigator(self):
         self.assertEquals(domain_request.creator, self.user1)
         self.assertEquals(domain_request.investigator, self.user1)
 
+
     @less_console_noise_decorator
     def test_transfer_user_transfers_domain_information_creator(self):
         """Assert that domain fields get transferred"""
@@ -2866,7 +2867,7 @@ def test_transfer_user_deletes_old_user(self):
         with self.assertRaises(User.DoesNotExist):
             self.user2.refresh_from_db()
 
-    @less_console_noise_decorator
+    # @less_console_noise_decorator
     def test_transfer_user_throws_transfer_and_delete_success_messages(self):
         """Test that success messages for data transfer and user deletion are displayed."""
         # Ensure the setup for VerifiedByStaff
@@ -2882,6 +2883,10 @@ def test_transfer_user_throws_transfer_and_delete_success_messages(self):
             submit_form["selected_user"] = self.user2.pk
             after_submit = submit_form.submit().follow()
 
+            print("mock_success_message.call_args_list:")
+            for call in mock_success_message.call_args_list:
+                print(call)
+
             self.assertContains(after_submit, "<h1>Change user</h1>")
 
             mock_success_message.assert_any_call(
@@ -2898,7 +2903,7 @@ def test_transfer_user_throws_transfer_and_delete_success_messages(self):
     def test_transfer_user_throws_error_message(self):
         """Test that an error message is thrown if the transfer fails."""
         with patch(
-            "registrar.views.TransferUserView.transfer_user_fields_and_log", side_effect=Exception("Simulated Error")
+            "registrar.views.TransferUserView.transfer_related_fields_and_log", side_effect=Exception("Simulated Error")
         ):
             with patch("django.contrib.messages.error") as mock_error:
                 # Access the transfer user page

src/registrar/views/transfer_user.py

@@ -1,7 +1,10 @@
 import logging
+from django.db import transaction
+from django.db.models import Manager,ForeignKey, OneToOneField, ManyToManyField, ManyToOneRel
 
 from django.shortcuts import render, get_object_or_404, redirect
 from django.views import View
+from registrar import models
 from registrar.models.domain import Domain
 from registrar.models.domain_information import DomainInformation
 from registrar.models.domain_request import DomainRequest
@@ -21,22 +24,8 @@
 class TransferUserView(View):
     """Transfer user methods that set up the transfer_user template and handle the forms on it."""
 
-    JOINS = [
-        (DomainRequest, "creator"),
-        (DomainInformation, "creator"),
-        (Portfolio, "creator"),
-        (DomainRequest, "investigator"),
-        (UserDomainRole, "user"),
-        (VerifiedByStaff, "requestor"),
-        (UserPortfolioPermission, "user"),
-    ]
-
-    # Future-proofing in case joined fields get added on the user model side
-    # This was tested in the first portfolio model iteration and works
-    USER_FIELDS: List[Any] = []
-
     def get(self, request, user_id):
-        """current_user referes to the 'source' user where the button that redirects to this view was clicked.
+        """current_user refers to the 'source' user where the button that redirects to this view was clicked.
         other_users exclude current_user and populate a dropdown, selected_user is the selection in the dropdown.
 
         This also querries the relevant domains and domain requests, and the admin context needed for the sidenav."""
@@ -71,86 +60,234 @@ def get(self, request, user_id):
 
     def post(self, request, user_id):
         """This handles the transfer from selected_user to current_user then deletes selected_user.
-
-        NOTE: We have a ticket to refactor this into a more solid lookup for related fields in #2645"""
-
+        """
         current_user = get_object_or_404(User, pk=user_id)
         selected_user_id = request.POST.get("selected_user")
         selected_user = get_object_or_404(User, pk=selected_user_id)
 
         try:
-            change_logs = []
+            # Make this atomic so that we don't get any partial transfers
+            with transaction.atomic():
+                change_logs = []
 
-            # Transfer specific fields
-            self.transfer_user_fields_and_log(selected_user, current_user, change_logs)
+                self._delete_duplicate_user_domain_roles_and_log(selected_user, current_user, change_logs)
 
-            # Perform the updates and log the changes
-            for model_class, field_name in self.JOINS:
-                self.update_joins_and_log(model_class, field_name, selected_user, current_user, change_logs)
+                self._delete_duplicate_user_portfolio_permissions_and_log(selected_user, current_user, change_logs)
+                # Dynamically handle related fields 
+                self.transfer_related_fields_and_log(selected_user, current_user, change_logs)
 
-            # Success message if any related objects were updated
-            if change_logs:
-                success_message = f"Data transferred successfully for the following objects: {change_logs}"
-                messages.success(request, success_message)
+                # Success message if any related objects were updated
+                logger.debug(f"change_logs: {change_logs}")
+                if change_logs:
+                    logger.debug(f"change_logs: {change_logs}")
+                    success_message = f"Data transferred successfully for the following objects: {change_logs}"
+                    messages.success(request, success_message)
 
-            selected_user.delete()
-            messages.success(request, f"Deleted {selected_user} {selected_user.username}")
+                logger.debug("Deleting old user")
+                selected_user.delete()
+                messages.success(request, f"Deleted {selected_user} {selected_user.username}")
 
         except Exception as e:
             messages.error(request, f"An error occurred during the transfer: {e}")
+            logger.error(f"An error occurred during the transfer: {e}", exc_info=True)
 
         return redirect("admin:registrar_user_change", object_id=user_id)
-
-    @classmethod
-    def update_joins_and_log(cls, model_class, field_name, selected_user, current_user, change_logs):
+
+    def _delete_duplicate_user_portfolio_permissions_and_log(self, selected_user, current_user, change_logs):
         """
-        Helper function to update the user join fields for a given model and log the changes.
+        Check and remove duplicate UserPortfolioPermission objects from the selected_user based on portfolios associated with the current_user.
         """
+        try:
+            # Fetch portfolios associated with the current_user
+            current_user_portfolios = UserPortfolioPermission.objects.filter(user=current_user).values_list('portfolio_id', flat=True)
+
+            # Identify duplicates in selected_user for these portfolios
+            duplicates = (
+                UserPortfolioPermission.objects
+                .filter(user=selected_user, portfolio_id__in=current_user_portfolios)
+            )
+
+            duplicate_count = duplicates.count()
+
+            if duplicate_count > 0:
+                # Log the specific duplicates before deletion for better traceability
+                duplicate_permissions = list(duplicates)
+                logger.debug(f"Duplicate permissions to be removed: {duplicate_permissions}")
+
+                duplicates.delete()
+                logger.info(f"Removed {duplicate_count} duplicate UserPortfolioPermission(s) from user_id {selected_user.id} for portfolios already associated with user_id {current_user.id}")
+                change_logs.append(f"Removed {duplicate_count} duplicate UserPortfolioPermission(s) from user_id {selected_user.id} for portfolios already associated with user_id {current_user.id}")
+
+        except Exception as e:
+            logger.error(f"Failed to check and remove duplicate UserPortfolioPermissions: {e}", exc_info=True)
+            raise
 
-        filter_kwargs = {field_name: selected_user}
-        updated_objects = model_class.objects.filter(**filter_kwargs)
+    def _delete_duplicate_user_domain_roles_and_log(self, selected_user, current_user, change_logs):
+        """
+        Check and remove duplicate UserDomainRole objects from the selected_user based on domains associated with the current_user.
+        Retain one instance per domain to maintain data integrity.
+        """
 
-        for obj in updated_objects:
-            # Check for duplicate UserDomainRole before updating
-            if model_class == UserDomainRole:
-                if model_class.objects.filter(user=current_user, domain=obj.domain).exists():
-                    continue  # Skip the update to avoid a duplicate
+        try:
+            # Fetch domains associated with the current_user
+            current_user_domains = UserDomainRole.objects.filter(user=current_user).values_list('domain_id', flat=True)
+
+            # Identify duplicates in selected_user for these domains
+            duplicates = (
+                UserDomainRole.objects
+                .filter(user=selected_user, domain_id__in=current_user_domains)
+            )
+
+            duplicate_count = duplicates.count()
+
+            if duplicate_count > 0:
+                duplicates.delete()
+                logger.info(
+                    f"Removed {duplicate_count} duplicate UserDomainRole(s) from user_id {selected_user.id} "
+                    f"for domains already associated with user_id {current_user.id}"
+                )
+                change_logs.append(
+                    f"Removed {duplicate_count} duplicate UserDomainRole(s) from user_id {selected_user.id} "
+                    f"for domains already associated with user_id {current_user.id}"
+                )
+
+        except Exception as e:
+            logger.error(f"Failed to check and remove duplicate UserDomainRoles: {e}", exc_info=True)
+            raise
 
-            if model_class == UserPortfolioPermission:
-                if model_class.objects.filter(user=current_user, portfolio=obj.portfolio).exists():
-                    continue  # Skip the update to avoid a duplicate
+    def transfer_related_fields_and_log(self, selected_user, current_user, change_logs):
+        """
+        Dynamically find all related fields to the User model and transfer them from selected_user to current_user.
+        Handles ForeignKey, OneToOneField, ManyToManyField, and ManyToOneRel relationships.
+        """
+        user_model = User
+
+        # Handle forward relationships
+        for related_field in user_model._meta.get_fields():
+            if related_field.is_relation and related_field.related_model:
+                if isinstance(related_field, ForeignKey):
+                    self._handle_foreign_key(related_field, selected_user, current_user, change_logs)
+                elif isinstance(related_field, OneToOneField):
+                    self._handle_one_to_one(related_field, selected_user, current_user, change_logs)
+                elif isinstance(related_field, ManyToManyField):
+                    self._handle_many_to_many(related_field, selected_user, current_user, change_logs)
+                elif isinstance(related_field, ManyToOneRel):
+                    self._handle_many_to_one_rel(related_field, selected_user, current_user, change_logs)
+
+        # # Handle reverse relationships
+        for related_object in user_model._meta.related_objects:
+            if isinstance(related_object, ManyToOneRel):
+                self._handle_many_to_one_rel(related_object, selected_user, current_user, change_logs)
+            elif isinstance(related_object.field, OneToOneField):
+                self._handle_one_to_one_reverse(related_object, selected_user, current_user, change_logs)
+            elif isinstance(related_object.field, ForeignKey):
+                self._handle_foreign_key_reverse(related_object, selected_user, current_user, change_logs)
+            elif isinstance(related_object.field, ManyToManyField):
+                self._handle_many_to_many_reverse(related_object, selected_user, current_user, change_logs)
+
+    def _handle_foreign_key(self, related_field: ForeignKey, selected_user, current_user, change_logs):
+        related_name = related_field.get_accessor_name()
+        # for foreign key relationships, getattr returns a manager
+        related_manager = getattr(selected_user, related_name, None)
+
+        if related_manager:
+            # get all the related objects
+            related_queryset = related_manager.all()
+            for obj in related_queryset:
+                # set the foreign key to the current user
+                setattr(obj, related_field.field.name, current_user)
+                obj.save()
+                log_entry = f'Transferred {related_field.field.name} from {selected_user} to {current_user}'
+                logger.info(log_entry)
+                change_logs.append(log_entry)
+
+    def _handle_one_to_one(self, related_field: OneToOneField, selected_user, current_user, change_logs):
+        related_name = related_field.get_accessor_name()
+        # for one to one relationships, getattr returns the related object
+        related_object = getattr(selected_user, related_name, None)
+
+        if related_object:
+            # set the one to one field to the current user
+            setattr(related_object, related_field.field.name, current_user)
+            related_object.save()
+            log_entry = f'Transferred {related_field.field.name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
+
+    def _handle_many_to_many(self, related_field: ManyToManyField, selected_user, current_user, change_logs):
+        # for many to many relationships, getattr returns a manager
+        related_manager = getattr(selected_user, related_field.name, None)
+        if related_manager:
+            # get all the related objects
+            related_queryset = related_manager.all()
+            # add the related objects to the current user
+            getattr(current_user, related_field.name).add(*related_queryset)
+            log_entry = f'Transferred {related_field.name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
+
+    def _handle_many_to_one_rel(self, related_object: ManyToOneRel, selected_user: User, current_user: User, change_logs: List[str]):
+        """
+        Handles ManyToOneRel relationships, where multiple objects relate to the User via a ForeignKey.
+        """
+        related_model = related_object.related_model
+        related_name = related_object.field.name
 
-            # Update the field on the object and save it
-            setattr(obj, field_name, current_user)
+        # for many to one relationships, we need a queryset
+        related_queryset = related_model.objects.filter(**{related_name: selected_user})
+        for obj in related_queryset:
+            setattr(obj, related_name, current_user)
             obj.save()
+            log_entry = f'Transferred {related_name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
 
-            # Log the change
-            cls.log_change(obj, field_name, selected_user, current_user, change_logs)
-
-    @classmethod
-    def transfer_user_fields_and_log(cls, selected_user, current_user, change_logs):
+    def _handle_one_to_one_reverse(self, related_object: OneToOneField, selected_user: User, current_user: User, change_logs: List[str]):
         """
-        Transfers portfolio fields from the selected_user to the current_user.
-        Logs the changes for each transferred field.
+        Handles reverse OneToOneField relationships.
         """
-        for field in cls.USER_FIELDS:
-            field_value = getattr(selected_user, field, None)
-
-            if field_value:
-                setattr(current_user, field, field_value)
-                cls.log_change(current_user, field, field_value, field_value, change_logs)
+        related_model = related_object.related_model
+        field_name = related_object.field.name
 
-        current_user.save()
-
-    @classmethod
-    def log_change(cls, obj, field_name, field_value, new_value, change_logs):
-        """Logs the change for a specific field on an object"""
-        log_entry = f'Changed {field_name} from "{field_value}" to "{new_value}" on {obj}'
+        try:
+            related_instance = related_model.objects.filter(**{field_name: selected_user}).first()
+            setattr(related_instance, field_name, current_user)
+            related_instance.save()
+            log_entry = f'Transferred {field_name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
+        except related_model.DoesNotExist:
+            logger.warning(f"No related instance found for reverse OneToOneField {field_name} for {selected_user}")
+
+    def _handle_foreign_key_reverse(self, related_object: ForeignKey, selected_user: User, current_user: User, change_logs: List[str]):
+        """
+        Handles reverse ForeignKey relationships.
+        """
+        related_model = related_object.related_model
+        field_name = related_object.field.name
 
-        logger.info(log_entry)
+        related_queryset = related_model.objects.filter(**{field_name: selected_user})
+        for obj in related_queryset:
+            setattr(obj, field_name, current_user)
+            obj.save()
+            log_entry = f'Transferred {field_name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
 
-        # Collect the related object for the success message
-        change_logs.append(log_entry)
+    def _handle_many_to_many_reverse(self, related_object: ManyToManyField, selected_user: User, current_user: User, change_logs: List[str]):
+        """
+        Handles reverse ManyToManyField relationships.
+        """
+        related_model = related_object.related_model
+        field_name = related_object.field.name
+
+        related_manager = related_model.objects.filter(**{field_name: selected_user})
+        if related_manager:
+            related_qs = related_manager.all()
+            getattr(current_user, field_name).add(*related_qs)
+            log_entry = f'Transferred {field_name} from {selected_user} to {current_user}'
+            logger.info(log_entry)
+            change_logs.append(log_entry)
 
     @classmethod
     def get_domains(cls, user):