fix: handle non-ascii username collisions

openedx/api.py

@@ -127,6 +127,81 @@ def _extract_username_suggestions(data, suggestions_extracted):
     return [], suggestions_extracted
 
 
+def _generate_unique_username(base_username, max_length=OPENEDX_USERNAME_MAX_LEN):
+    """
+    Generate a unique username by appending random numbers to the base username.
+
+    Args:
+        base_username (str): The base username to use
+        max_length (int): Maximum length for the generated username
+
+    Returns:
+        str or None: A unique username, or None if no unique username could be generated
+    """
+    ranges = (
+        (0, 9),
+        (10, 99),
+        (100, 999),
+        (1000, 9999),
+        (10000, 99999),
+    )
+
+    for intrange in ranges:
+        random_int = random.randint(intrange[0], intrange[1])  # noqa: S311
+        username_to_try = f"{base_username}_{random_int}"
+
+        if len(username_to_try) > max_length:
+            amount_to_truncate = len(username_to_try) - max_length
+            username_to_try = f"{base_username[:amount_to_truncate]}-{random_int}"
+
+        if not OpenEdxUser.objects.filter(edx_username=username_to_try).exists():
+            return username_to_try
+
+    return None
+
+
+def _handle_username_collision(  # noqa: PLR0913
+    resp, data, open_edx_user, user, suggested_usernames, suggestions_extracted
+):
+    """
+    Handle username collision by trying OpenEdX suggestions or falling back to local generation.
+
+    Args:
+        resp: HTTP response from OpenEdX
+        data: Parsed JSON response data
+        open_edx_user: OpenEdxUser instance
+        user: User instance
+        suggested_usernames: List of suggested usernames from previous attempts
+        suggestions_extracted: Boolean indicating if suggestions were already extracted
+
+    Returns:
+        tuple: (new_username, should_continue, should_reset_attempts)
+    """
+    if not _is_duplicate_username_error(resp, data):
+        return None, False, False
+
+    suggestions, suggestions_extracted = _extract_username_suggestions(
+        data, suggestions_extracted
+    )
+    if suggestions:
+        suggested_usernames = suggestions
+
+    if not suggested_usernames:
+        log.info(
+            "OpenEdX returned empty username suggestions, falling back to local generation"
+        )
+        base_username = open_edx_user.desired_edx_username or user.username
+
+        new_username = _generate_unique_username(base_username)
+
+        if new_username:
+            return new_username, True, True
+        else:
+            return None, False, False
+
+    return suggested_usernames.pop(0), True, False
+
+
 def _create_edx_user_request(open_edx_user, user, access_token):  # noqa: C901
     """
     Handle the actual user creation request to Open edX with retry logic for duplicate usernames.
@@ -196,17 +271,22 @@ def _create_edx_user_request(open_edx_user, user, access_token):  # noqa: C901
             except (ValueError, requests.exceptions.JSONDecodeError):
                 data = {}
 
-            if _is_duplicate_username_error(resp, data):
-                suggestions, suggestions_extracted = _extract_username_suggestions(
-                    data, suggestions_extracted
+            new_username, should_continue, should_reset_attempts = (
+                _handle_username_collision(
+                    resp,
+                    data,
+                    open_edx_user,
+                    user,
+                    suggested_usernames,
+                    suggestions_extracted,
                 )
-                if suggestions:
-                    suggested_usernames = suggestions
-
-                if not suggested_usernames:
-                    break
+            )
 
-                current_username = suggested_usernames.pop(0)
+            if should_continue:
+                current_username = new_username
+                if should_reset_attempts:
+                    attempt = 0
+                continue
             elif _is_bad_request(resp):
                 open_edx_user.has_sync_error = True
                 open_edx_user.sync_error_data = data
@@ -316,31 +396,13 @@ def reconcile_edx_username(user, *, desired_username=None):
         if not OpenEdxUser.objects.filter(edx_username=edx_username).exists():
             edx_user.save()
         else:
-            ranges = (
-                (0, 9),
-                (10, 99),
-                (100, 999),
-                (1000, 9999),
-                (10000, 99999),
-            )
-
-            for intrange in ranges:
-                random_int = random.randint(intrange[0], intrange[1])  # noqa: S311
-                username_to_try = f"{edx_username}_{random_int}"
-
-                if len(username_to_try) > OPENEDX_USERNAME_MAX_LEN:
-                    amount_to_truncate = len(username_to_try) - OPENEDX_USERNAME_MAX_LEN
-                    username_to_try = (
-                        f"{edx_username[:amount_to_truncate]}-{random_int}"
-                    )
-
-                if not OpenEdxUser.objects.filter(
-                    edx_username=username_to_try,
-                ).exists():
-                    edx_user.edx_username = username_to_try
-                    edx_user.desired_edx_username = username_to_try
-                    edx_user.save()
-                    break
+            unique_username = _generate_unique_username(edx_username)
+            if unique_username:
+                edx_user.edx_username = unique_username
+                edx_user.desired_edx_username = unique_username
+                edx_user.save()
+            else:
+                log.warning("Could not generate unique username for %s", edx_username)
 
         return True
 

openedx/api_test.py

@@ -24,6 +24,7 @@
     ACCESS_TOKEN_HEADER_NAME,
     OPENEDX_AUTH_DEFAULT_TTL_IN_SECONDS,
     OPENEDX_REGISTRATION_VALIDATION_PATH,
+    _generate_unique_username,
     bulk_retire_edx_users,
     create_edx_auth_token,
     create_edx_user,
@@ -233,9 +234,41 @@ def test_create_edx_user(  # noqa: PLR0913
 
 @responses.activate
 @pytest.mark.usefixtures("application")
-def test_create_edx_user_conflict(settings):
+@pytest.mark.parametrize(
+    (
+        "username_suggestions",
+        "base_username",
+        "expected_username_pattern",
+        "test_description",
+    ),
+    [
+        (
+            ["openedx-generated-username"],
+            "testuser",
+            lambda username: username == "openedx-generated-username",
+            "with OpenEdX suggestions",
+        ),
+        (
+            [],
+            "José",
+            lambda username: username.startswith("José_")
+            and len(username) > len("José"),
+            "with empty suggestions (non-ASCII fallback)",
+        ),
+    ],
+)
+def test_create_edx_user_conflict(
+    settings,
+    username_suggestions,
+    base_username,
+    expected_username_pattern,
+    test_description,
+):
     """Test that create_edx_user handles a 409 response from the edX API"""
-    user = UserFactory.create(openedx_user__has_been_synced=False)
+    user = UserFactory.create(
+        openedx_user__has_been_synced=False,
+        openedx_user__desired_edx_username=base_username,
+    )
 
     resp1 = responses.add(
         responses.GET,
@@ -248,7 +281,7 @@ def test_create_edx_user_conflict(settings):
         f"{settings.OPENEDX_API_BASE_URL}/user_api/v1/account/registration/",
         json={
             "error_code": "duplicate-username",
-            "username_suggestions": ["openedx-generated-username"],
+            "username_suggestions": username_suggestions,
         },
         status=status.HTTP_409_CONFLICT,
     )
@@ -272,7 +305,25 @@ def test_create_edx_user_conflict(settings):
     edx_user = user.openedx_users.first()
 
     assert edx_user.has_been_synced is True
-    assert edx_user.edx_username == "openedx-generated-username"
+    assert expected_username_pattern(edx_user.edx_username), (
+        f"Username {edx_user.edx_username} doesn't match expected pattern for {test_description}"
+    )
+
+
+@pytest.mark.parametrize(
+    ("base_username", "expected_prefix"),
+    [
+        ("testuser", "testuser_"),
+        ("José", "José_"),
+        ("user123", "user123_"),
+    ],
+)
+def test_generate_unique_username(base_username, expected_prefix):
+    """Test that _generate_unique_username generates unique usernames"""
+
+    username = _generate_unique_username(base_username)
+    assert username.startswith(expected_prefix)
+    assert len(username) > len(base_username)
 
 
 @responses.activate