diff --git a/openedx/api.py b/openedx/api.py index fe33e418c5..9e87db0983 100644 --- a/openedx/api.py +++ b/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 diff --git a/openedx/api_test.py b/openedx/api_test.py index b18bd5e0e3..5701a7e66f 100644 --- a/openedx/api_test.py +++ b/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