[fix] Handle username collision during registration

docs/user/rest-api.rst

@@ -487,6 +487,26 @@ simple, but can be :ref:`enabled through configuration
 <openwisp_radius_needs_identity_verification>`; if identity verification
 is disabled for a particular org, an empty string will be acceptable.
 
+.. note::
+
+    When the submitted ``username`` matches the local-part of the
+    submitted ``email`` address and that username is already used by
+    another email address in the same organization, the registration API
+    automatically selects the next available username by appending a
+    numeric suffix.
+
+    For example, registrations using the same email local-part can result
+    in:
+
+    - ``johndae@gmail.com`` → ``johndae``
+    - ``johndae@yahoo.com`` → ``johndae1``
+    - ``johndae@outlook.com`` → ``johndae2``
+
+    This does not change the behavior of cross-organization registration,
+    where an existing account is handled as described in :ref:`Registering
+    to Multiple Organizations
+    <radius_registering_to_multiple_organizations>`.
+
 .. _radius_registering_to_multiple_organizations:
 
 Registering to Multiple Organizations

openwisp_radius/api/serializers.py

@@ -37,6 +37,7 @@
 from .. import settings as app_settings
 from ..counters.exceptions import SkipCheck
 from ..utils import (
+    find_available_username,
     get_group_checks,
     get_organization_radius_settings,
     get_user_group,
@@ -593,6 +594,28 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.fields["method"].choices = app_settings.USER_SETTABLE_REGISTRATION_METHODS
 
+    def validate_username(self, username):
+        email = self.initial_data.get("email")
+        local_part = (
+            email.rsplit("@", 1)[0] if isinstance(email, str) and "@" in email else ""
+        )
+
+        organization = self.context["view"].organization
+        existing_username = User.objects.filter(username=username)
+
+        if (
+            username.casefold() == local_part.casefold()
+            and existing_username.exists()
+            and not existing_username.filter(email__iexact=email).exists()
+            and OrganizationUser.objects.filter(
+                organization=organization,
+                user__username=username,
+            ).exists()
+        ):
+            username = find_available_username(username, [])
+
+        return super().validate_username(username)
+
     def validate_phone_number(self, phone_number):
         org = self.context["view"].organization
         if get_organization_radius_settings(org, "sms_verification"):

openwisp_radius/tests/test_api/test_api.py

@@ -207,6 +207,58 @@ def test_register_201(self):
             False,
         )
 
+    def test_register_same_username_different_email(self):
+        User.objects.create(
+            username="johndae1",
+            email="occupied@example.com",
+        )
+        self._register_user(
+            extra_params={
+                "username": "johndae",
+                "email": "johndae@gmail.com",
+            },
+            expect_users=2,
+        )
+
+        self._register_user(
+            extra_params={
+                "username": "johndae",
+                "email": "johndae@yahoo.com",
+            },
+            expect_users=3,
+        )
+
+        self._register_user(
+            extra_params={
+                "username": "johndae",
+                "email": "johndae@outlook.com",
+            },
+            expect_users=4,
+        )
+
+        users = User.objects.filter(
+            email__in=[
+                "johndae@gmail.com",
+                "johndae@yahoo.com",
+                "johndae@outlook.com",
+            ]
+        )
+
+        self.assertEqual(users.count(), 3)
+
+        self.assertEqual(
+            users.get(email="johndae@gmail.com").username,
+            "johndae",
+        )
+        self.assertEqual(
+            users.get(email="johndae@yahoo.com").username,
+            "johndae2",
+        )
+        self.assertEqual(
+            users.get(email="johndae@outlook.com").username,
+            "johndae3",
+        )
+
     def test_register_400_password(self):
         response = self._register_user(
             extra_params={"password1": "password1", "password2": "password2"},
@@ -365,6 +417,87 @@ def test_register_duplicate_different_org(self):
         self.default_org.radius_settings.sms_verification = False
         self.default_org.radius_settings.save()
 
+    def test_register_same_username_different_email_different_org(self):
+        org1 = self.default_org
+        org2 = self._get_org(org_name="org2")
+
+        existing_user = self._create_user(
+            username="johndae",
+            email="johndae@gmail.com",
+        )
+        OrganizationUser.objects.create(
+            user=existing_user,
+            organization=org1,
+        )
+
+        url = reverse("radius:rest_register", args=[org2.slug])
+        response = self.client.post(
+            url,
+            data={
+                "username": "johndae",
+                "email": "johndae@yahoo.com",
+                "password1": "password",
+                "password2": "password",
+            },
+        )
+
+        self.assertEqual(response.status_code, 409)
+        self.assertEqual(
+            response.data,
+            {
+                "details": "A user like the one being registered already exists.",
+                "organizations": [
+                    {"slug": org1.slug, "name": org1.name},
+                ],
+            },
+        )
+        self.assertEqual(
+            User.objects.filter(username="johndae").count(),
+            1,
+        )
+
+    def test_register_same_username_different_email_case_insensitive(self):
+        self._register_user(
+            extra_params={
+                "username": "johndae",
+                "email": "johndae@gmail.com",
+            }
+        )
+
+        response = self._register_user(
+            extra_params={
+                "username": "johndae",
+                "email": "JohnDae@yahoo.com",
+            },
+            expect_users=2,
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(
+            User.objects.get(email__iexact="JohnDae@yahoo.com").username,
+            "johndae1",
+        )
+
+    def test_register_invalid_email_type(self):
+        self._superuser_login()
+
+        url = reverse("radius:rest_register", args=[self.default_org.slug])
+        response = self.client.post(
+            url,
+            data=json.dumps(
+                {
+                    "username": "invalid-email-user",
+                    "email": None,
+                    "password1": "password",
+                    "password2": "password",
+                }
+            ),
+            content_type="application/json",
+        )
+
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("email", response.data)
+
     def test_radius_user_serializer(self):
         self._register_user()
         try: