Merge pull request #5204 from kobotoolbox/task-1077-update-organizati…

…on-detail-in-serializer-to-return-is_mmo

Task 1077 update organization detail in serializer to return is mmo

hub/models/extra_user_detail.py

@@ -35,6 +35,8 @@ def save(
         if not update_fields or (update_fields and 'data' in update_fields):
             self.standardize_json_field('data', 'organization', str)
             self.standardize_json_field('data', 'name', str)
+            if not created:
+                self._sync_org_name()
 
         super().save(
             force_insert=force_insert,
@@ -59,3 +61,22 @@ def save(
                 self.user.id,
                 self.validated_password,
             )
+
+    def _sync_org_name(self):
+        """
+        Synchronizes the `name` field of the Organization model with the
+        "organization" attribute found in the `data` field of ExtraUserDetail,
+        but only if the user is the owner.
+
+        This ensures that any updates in the metadata are accurately reflected
+        in the organization's name.
+        """
+        user_organization = self.user.organization
+        if user_organization.is_owner(self.user):
+            try:
+                organization_name = self.data['organization'].strip()
+            except (KeyError, AttributeError):
+                organization_name = None
+
+            user_organization.name = organization_name
+            user_organization.save(update_fields=['name'])

kobo/apps/accounts/forms.py

@@ -279,18 +279,24 @@ class SignupForm(KoboSignupMixin, BaseSignupForm):
         'newsletter_subscription',
     ]
 
-
     def clean(self):
         """
         Override parent form to pass extra user's attributes to validation.
         """
         super(SignupForm, self).clean()
 
         User = get_user_model()  # noqa
+
         dummy_user = User()
+        # Using `dummy_user`, a temporary User object, to assign attributes that are
+        # validated during the password similarity check.
         user_username(dummy_user, self.cleaned_data.get('username'))
         user_email(dummy_user, self.cleaned_data.get('email'))
-        setattr(dummy_user, 'organization', self.cleaned_data.get('organization', ''))
+        setattr(
+            dummy_user,
+            'organization_name',
+            self.cleaned_data.get('organization', ''),
+        )
         setattr(dummy_user, 'full_name', self.cleaned_data.get('name', ''))
 
         password = self.cleaned_data.get('password1')

kobo/apps/kobo_auth/models.py

@@ -1,10 +1,12 @@
 from django.conf import settings
 from django.contrib.auth.models import AbstractUser
+from django_request_cache import cache_for_request
 
 from kobo.apps.openrosa.libs.constants import (
     OPENROSA_APP_LABELS,
 )
 from kobo.apps.openrosa.libs.permissions import get_model_permission_codenames
+from kobo.apps.organizations.models import create_organization, Organization
 from kpi.utils.database import update_autofield_sequence, use_db
 
 
@@ -38,6 +40,33 @@ def has_perm(self, perm, obj=None):
         # Otherwise, check in KPI DB
         return super().has_perm(perm, obj)
 
+    @property
+    @cache_for_request
+    def is_org_owner(self):
+        """
+        Shortcut to check if the user is the owner of the organization, allowing
+        direct access via the User object instead of calling `organization.is_owner()`.
+        """
+        return self.organization.is_owner(self)
+
+    @property
+    @cache_for_request
+    def organization(self):
+        # Database allows multiple organizations per user, but we restrict it to one.
+        if organization := Organization.objects.filter(
+            organization_users__user=self
+        ).first():
+            return organization
+
+        try:
+            organization_name = self.extra_details.data['organization'].strip()
+        except (KeyError, AttributeError):
+            organization_name = None
+
+        return create_organization(
+            self, organization_name or f'{self.username}’s organization'
+        )
+
     def sync_to_openrosa_db(self):
         User = self.__class__  # noqa
         User.objects.using(settings.OPENROSA_DB_ALIAS).bulk_create(

kobo/apps/kobo_auth/signals.py

@@ -21,6 +21,17 @@ def create_auth_token(sender, instance=None, created=False, **kwargs):
         Token.objects.get_or_create(user_id=instance.pk)
 
 
+@receiver(post_save, sender=User)
+def create_organization(sender, instance, created, raw, **kwargs):
+    """
+    Create organization for user
+    """
+    user = instance
+    if created:
+        # calling the property will create the organization if it does not exist.
+        user.organization
+
+
 @receiver(post_save, sender=User)
 def default_permissions_post_save(sender, instance, created, raw, **kwargs):
     """

kobo/apps/organizations/constants.py

@@ -0,0 +1,4 @@
+ADMIN_ORG_ROLE = 'admin'
+EXTERNAL_ORG_ROLE = 'external'
+MEMBER_ORG_ROLE = 'member'
+OWNER_ORG_ROLE = 'owner'

kobo/apps/organizations/migrations/0006_update_organization_name.py

@@ -0,0 +1,50 @@
+# Generated by Django 4.2.15 on 2024-10-25 16:08
+
+from django.db import migrations
+from django.core.paginator import Paginator
+
+
+def update_organization_names(apps, schema_editor):
+    Organization = apps.get_model('organizations', 'Organization')
+    OrganizationUser = apps.get_model('organizations', 'OrganizationUser')
+
+    page_size = 2000
+    paginator = Paginator(
+        OrganizationUser.objects.filter(organizationowner__isnull=False)
+        .select_related('user', 'organization', 'user__extra_details')
+        .order_by('pk'),
+        page_size,
+    )
+    for page in paginator.page_range:
+        organization_users = paginator.page(page).object_list
+        organizations = []
+        for organization_user in organization_users:
+            user = organization_user.user
+            organization = organization_user.organization
+            if (
+                organization.name
+                and organization.name.strip() != ''
+                and organization.name.startswith(user.username)
+            ):
+                try:
+                    organization_name = user.extra_details.data['organization'].strip()
+                except (KeyError, AttributeError):
+                    continue
+
+                organization.name = organization_name
+                organizations.append(organization)
+
+        Organization.objects.bulk_update(organizations, ['name'])
+
+
+def noop(apps, schema_editor):
+    pass
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('organizations', '0005_add_mmo_override_field_to_organization'),
+    ]
+
+    operations = [migrations.RunPython(update_organization_names, noop)]

kobo/apps/organizations/models.py

@@ -1,3 +1,4 @@
+from typing import Literal
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
@@ -18,6 +19,16 @@
 
 from kobo.apps.stripe.constants import ACTIVE_STRIPE_STATUSES
 from kpi.fields import KpiUidField
+from .constants import (
+    ADMIN_ORG_ROLE,
+    EXTERNAL_ORG_ROLE,
+    MEMBER_ORG_ROLE,
+    OWNER_ORG_ROLE,
+)
+
+OrganizationRole = Literal[
+    ADMIN_ORG_ROLE, EXTERNAL_ORG_ROLE, MEMBER_ORG_ROLE, OWNER_ORG_ROLE
+]
 
 
 class Organization(AbstractOrganization):
@@ -80,9 +91,55 @@ def email(self):
         except AttributeError:
             return
 
+    @cache_for_request
+    def get_user_role(self, user: 'User') -> OrganizationRole:
+
+        if not self.users.filter(pk=user.pk).exists():
+            return EXTERNAL_ORG_ROLE
+
+        if self.is_owner(user):
+            return OWNER_ORG_ROLE
+
+        if self.is_admin(user):
+            return ADMIN_ORG_ROLE
+
+        return MEMBER_ORG_ROLE
+
+    @cache_for_request
+    def is_admin(self, user: 'User') -> bool:
+        """
+        Only extends super() to add decorator @cache_for_request and avoid
+        multiple calls to DB in the same request
+        """
+
+        return super().is_admin(user)
+
+    @property
+    def is_mmo(self):
+        """
+        Determines if the multi-members feature is active for the organization
+
+        This returns True if:
+        - A superuser has enabled the override (`mmo_override`), or
+        - The organization has an active subscription.
+
+        If the override is enabled, it takes precedence over the subscription status
+        """
+        return self.mmo_override or bool(self.active_subscription_billing_details())
+
+    @cache_for_request
+    def is_owner(self, user):
+        """
+        Only extends super() to add decorator @cache_for_request and avoid
+        multiple calls to DB in the same request
+        """
+
+        return super().is_owner(user)
+
     @property
     @cache_for_request
     def owner_user_object(self) -> 'User':
+
         try:
             return self.owner.organization_user.user
         except ObjectDoesNotExist:

kobo/apps/organizations/serializers.py

@@ -6,6 +6,8 @@
     OrganizationOwner,
     OrganizationUser,
 )
+from kpi.utils.object_permission import get_database_user
+from .constants import EXTERNAL_ORG_ROLE
 
 
 class OrganizationUserSerializer(serializers.ModelSerializer):
@@ -24,17 +26,44 @@ class Meta:
 
 
 class OrganizationSerializer(serializers.ModelSerializer):
-    is_owner = serializers.SerializerMethodField('_is_owner')
 
-    def _is_owner(self, instance):
-        user = self.context['request'].user
-        return instance.owner.organization_user.user.id == user.id
+    is_mmo = serializers.BooleanField(read_only=True)
+    is_owner = serializers.SerializerMethodField()
+    request_user_role = serializers.SerializerMethodField()
 
     class Meta:
         model = Organization
-        fields = ['id', 'name', 'is_active', 'created', 'modified', 'slug', 'is_owner']
+        fields = [
+            'id',
+            'name',
+            'is_active',
+            'created',
+            'modified',
+            'slug',
+            'is_owner',
+            'is_mmo',
+            'request_user_role',
+        ]
         read_only_fields = ['id', 'slug']
 
     def create(self, validated_data):
         user = self.context['request'].user
         return create_organization(user, validated_data['name'])
+
+    def get_is_owner(self, organization):
+
+        # This method is deprecated.
+        # Use `get_request_user_role` to retrieve the value instead.
+        if request := self.context.get('request'):
+            user = get_database_user(request.user)
+            return organization.is_owner(user)
+
+        return False
+
+    def get_request_user_role(self, organization):
+
+        if request := self.context.get('request'):
+            user = get_database_user(request.user)
+            return organization.get_user_role(user)
+
+        return EXTERNAL_ORG_ROLE

kobo/apps/organizations/tests/test_organization.py

@@ -0,0 +1,74 @@
+from django.test import TestCase
+
+from kobo.apps.kobo_auth.shortcuts import User
+from kobo.apps.organizations.models import Organization
+from kobo.apps.organizations.constants import (
+    ADMIN_ORG_ROLE,
+    EXTERNAL_ORG_ROLE,
+    MEMBER_ORG_ROLE,
+    OWNER_ORG_ROLE,
+)
+
+
+class OrganizationTestCase(TestCase):
+
+    fixtures = ['test_data']
+
+    def setUp(self):
+        self.someuser = User.objects.get(username='someuser')
+
+        # someuser is the only member their organization, and the owner as well.
+        self.organization = self.someuser.organization
+
+    def test_owner_user_object_property(self):
+        anotheruser = User.objects.get(username='anotheruser')
+        self.organization.add_user(anotheruser)
+        assert self.organization.owner_user_object == self.someuser
+
+    def test_get_user_role(self):
+        anotheruser = User.objects.get(username='anotheruser')
+        alice = User.objects.create(username='alice', email='[email protected]')
+        external = User.objects.create(
+            username='external', email='[email protected]'
+        )
+        self.organization.add_user(anotheruser, is_admin=True)
+        self.organization.add_user(alice)
+        assert self.organization.get_user_role(self.someuser) == OWNER_ORG_ROLE
+        assert self.organization.get_user_role(anotheruser) == ADMIN_ORG_ROLE
+        assert self.organization.get_user_role(alice) == MEMBER_ORG_ROLE
+        assert self.organization.get_user_role(external) == EXTERNAL_ORG_ROLE
+
+    def test_create_organization_on_user_creation(self):
+        assert not Organization.objects.filter(name__startswith='alice').exists()
+        organization_count = Organization.objects.all().count()
+        User.objects.create_user(
+            username='alice', password='alice', email='[email protected]'
+        )
+        assert Organization.objects.filter(name__startswith='alice').exists()
+        assert Organization.objects.all().count() == organization_count + 1
+
+    def test_sync_org_name_on_save(self):
+        """
+        Tests the synchronization of the organization name with the metadata in
+        ExtraUserDetail upon saving.
+
+        This synchronization should only occur if the user is the owner of the
+        organization.
+        """
+        # someuser is the owner
+        assert self.organization.name == 'someuser’s organization'
+        someuser_extra_details = self.someuser.extra_details
+        someuser_extra_details.data['organization'] = 'SomeUser Technologies'
+        someuser_extra_details.save()
+        self.organization.refresh_from_db()
+        assert self.organization.name == 'SomeUser Technologies'
+
+        # another is an admin
+        anotheruser = User.objects.get(username='anotheruser')
+
+        self.organization.add_user(anotheruser, is_admin=True)
+        anotheruser_extra_details = anotheruser.extra_details
+        anotheruser_extra_details.data['organization'] = 'AnotherUser Enterprises'
+        anotheruser_extra_details.save()
+        self.organization.refresh_from_db()
+        assert self.organization.name == 'SomeUser Technologies'