Implemented user merging. Relevant for #59 (#73)

Implemented user merging, closes #59. Authored-by: @leonardlorenz

kubeportal/admin.py

@@ -10,6 +10,7 @@
 from django.template.response import TemplateResponse
 from oidc_provider.models import Client
 from sortedm2m_filter_horizontal_widget.forms import SortedFilteredSelectMultiple
+from kubeportal.models import UserState as states
 import logging
 import uuid
 from . import models, admin_views
@@ -352,10 +353,47 @@ def reject(modeladmin, request, queryset):
     for user in queryset:
         if user.reject(request):
             user.save()
-
-
 reject.short_description = "Reject access request for selected users"
 
+def merge_users(modeladmin, request, queryset):
+    if len(queryset) != 2:
+        messages.warning(
+            request, "Please choose exactly two users to merge.")
+        return reverse('admin:index')
+    primary, secondary = queryset.order_by('date_joined')
+
+    # first check if any of the two accounts are rejected.
+    # if any are, make sure to reject both as well.
+    if primary.state == states.ACCESS_REJECTED or secondary.state == states.ACCESS_REJECTED:
+        if primary.reject(request):
+            primary.state = states.ACCESS_REJECTED
+            messages.warning(
+                request, F"Rejected cluster access for '{primary.username}'")
+
+    # primary should be default. if secondary has more rights, then
+    # secondary's values should be merged into primary.
+    if primary.state != states.ACCESS_APPROVED and secondary.state == states.ACCESS_APPROVED:
+        primary.state = states.ACCESS_APPROVED
+        primary.approval_id = secondary.approval_id
+        primary.answered_by = secondary.answered_by
+    # iterate through the groups of secondary and add them to primary
+    # if primary is not in group
+    joined_groups = []
+    for group in secondary.portal_groups.all():
+        if group not in primary.portal_groups.all():
+            primary.portal_groups.add(group)
+            joined_groups.append(group)
+    joined_groups = [str(g) for g in joined_groups]
+    if joined_groups:
+        messages.info(request, F"User '{primary.username}' joined the group(s) {joined_groups}")
+    if primary.comments == "" or primary.comments == None:
+        if  secondary.comments != "" and secondary.comments is not None:
+            primary.comments = secondary.comments
+    primary.save()
+    secondary.delete()
+    messages.info(request, F"The Users '{primary.username}' and '{secondary.username}' have been merged into '{primary.username}' and '{secondary.username}' has been deleted.")
+merge_users.short_description = "Merges two users and keeps the one that joined first."
+
 
 class PortalUserAdmin(UserAdmin):
     list_display = ('username', 'first_name', 'last_name', 'comments',
@@ -366,7 +404,7 @@ class PortalUserAdmin(UserAdmin):
         (None, {'fields': ('state', 'answered_by', 'service_account')}),
         (None, {'fields': ('portal_groups',)})
     )
-    actions = [reject]
+    actions = [reject, merge_users]
     list_filter = ()
 
     def portal_group_list(self, instance):

kubeportal/tests/test_admin.py

@@ -10,8 +10,10 @@
 from kubeportal import models
 from kubeportal.models import KubernetesNamespace
 from kubeportal.models import KubernetesServiceAccount
+from kubeportal.models import PortalGroup
 from kubeportal.tests import AdminLoggedInTestCase
 from unittest.mock import patch
+from kubeportal.admin import merge_users, UserAdmin
 
 
 class Backend(AdminLoggedInTestCase):
@@ -158,8 +160,8 @@ def test_special_k8s_unapproved(self):
         new_svc.save()
         User = get_user_model()
         # create approved user
-        u = User(username="Hugo", 
-                 email="a@b.de", 
+        u = User(username="Hugo",
+                 email="a@b.de",
                  state=models.UserState.ACCESS_APPROVED,
                  service_account = new_svc)
         u.save()
@@ -188,6 +190,108 @@ def _test_user_rejection(self):
     def test_user_rejection(self):
         self._test_user_rejection()
 
+    '''
+    Create two users with the secondary (the later created) one having cluster access,
+    an assigned comment and two assigned groups
+    Merge both users.
+    The primary user should be assigned the cluster access, user comment and all the
+    portal groups of the secondary user.
+    The secondary user should be deleted.
+    '''
+    def test_user_merge_access_approved(self):
+        User = get_user_model()
+        primary = User(
+                username="HUGO",
+                email="a@b.de")
+        primary.save()
+
+        ns = KubernetesNamespace(name="default")
+        ns.save()
+        new_svc = KubernetesServiceAccount(name="foobar", namespace=ns)
+        new_svc.save()
+        secondary = User(
+                username="hugo",
+                state=models.UserState.ACCESS_APPROVED,
+                email="a@b.de",
+                comments = "secondary user comment",
+                service_account = new_svc)
+        secondary.save()
+
+        group1 = PortalGroup(name="testgroup1")
+        group1.save()
+        group2 = PortalGroup(name="testgroup2")
+        group2.save()
+
+        secondary.portal_groups.add(group1)
+        secondary.portal_groups.add(group2)
+        secondary.save()
+
+        # Build full-fledged request object for logged-in admin
+        request = self._build_full_request_mock('admin:index')
+        # approve secondary for cluster access
+        secondary.approve(request, new_svc)
+
+        # the merge method only accepts a queryset of users since that's what
+        # the admin interface creates
+        queryset_of_users = User.objects.filter(pk__in = [primary.id, secondary.id])
+
+        # merge both users. shouldn't return anything
+        assert(not merge_users(UserAdmin, request, queryset_of_users))
+
+        # the primary user has been altered but the old object is still in memory
+        # we need to query for the updated user again
+        primary = User.objects.get(pk = primary.id)
+
+        # Does primary have all the values of secondary user?
+        self.assertEquals(primary.comments, "secondary user comment")
+        assert(primary.portal_groups.filter(name = group1.name))
+        assert(primary.portal_groups.filter(name = group2.name))
+        assert(primary.has_access_approved)
+
+    '''
+    Create two users with the secondary (the later created) one having rejected cluster access,
+    Merge both users.
+    The primary user should be assigned the rejected cluster access.
+    The secondary user should be deleted.
+    '''
+    def test_user_merge_access_rejected(self):
+        User = get_user_model()
+        primary = User(
+                username="HUGO",
+                email="a@b.de")
+        primary.save()
+
+        ns = KubernetesNamespace(name="default")
+        ns.save()
+        new_svc = KubernetesServiceAccount(name="foobar", namespace=ns)
+        new_svc.save()
+        secondary = User(
+                username="hugo",
+                state=models.UserState.ACCESS_APPROVED,
+                email="a@b.de",
+                comments = "secondary user comment",
+                service_account = new_svc)
+        secondary.save()
+
+        # Build full-fledged request object for logged-in admin
+        request = self._build_full_request_mock('admin:index')
+        # reject cluster access for secondary
+        secondary.reject(request)
+
+        # the merge method only accepts a queryset of users since that's what
+        # the admin interface creates
+        queryset_of_users = User.objects.filter(pk__in = [primary.id, secondary.id])
+
+        # merge both users. shouldn't return anything
+        assert(not merge_users(UserAdmin, request, queryset_of_users))
+
+        # the primary user has been altered but the old object is still in memory
+        # we need to query for the updated user again
+        primary = User.objects.get(pk = primary.id)
+
+        assert(primary.has_access_rejected)
+
+
     @override_settings(EMAIL_BACKEND='django.core.mail.backends.smtp.EmailBackend', EMAIL_HOST_PASSWORD='sdsds')
     def test_user_rejection_mail_broken(self):
         self._test_user_rejection()