diff --git a/shelfmark/core/cwa_user_sync.py b/shelfmark/core/cwa_user_sync.py index f633ce5f..3f594e27 100644 --- a/shelfmark/core/cwa_user_sync.py +++ b/shelfmark/core/cwa_user_sync.py @@ -39,6 +39,7 @@ def upsert_cwa_user( email=normalized_email, role=role, allow_email_link=True, + sync_username=True, collision_strategy=collision_strategy, alias_suffix=_CWA_ALIAS_SUFFIX, context=context, diff --git a/shelfmark/core/external_user_linking.py b/shelfmark/core/external_user_linking.py index 75cb3f88..8ddabe5e 100644 --- a/shelfmark/core/external_user_linking.py +++ b/shelfmark/core/external_user_linking.py @@ -108,6 +108,7 @@ def _build_updates( auth_source: str, role: str, sync_role: bool, + username: str | object, email: str | None | object, display_name: str | None | object, subject_field: str | None, @@ -116,6 +117,8 @@ def _build_updates( updates: dict[str, Any] = {"auth_source": auth_source} if sync_role: updates["role"] = _normalize_role(role) + if username is not UNSET: + updates["username"] = _normalize_username(username) if email is not UNSET: updates["email"] = _normalize_email(email) if display_name is not UNSET: @@ -125,10 +128,17 @@ def _build_updates( return updates -def _next_suffix_username(user_db: UserDB, base_username: str) -> str: +def _next_suffix_username( + user_db: UserDB, + base_username: str, + *, + exclude_user_id: int | None = None, +) -> str: candidate = base_username suffix = 1 - while user_db.get_user(username=candidate): + while existing := user_db.get_user(username=candidate): + if exclude_user_id is not None and int(existing.get("id") or 0) == exclude_user_id: + return candidate candidate = f"{base_username}_{suffix}" suffix += 1 return candidate @@ -185,6 +195,38 @@ def _resolve_create_username( return _next_suffix_username(user_db, alias_base), None, "username_collision_alias" +def _resolve_update_username( + user_db: UserDB, + *, + current_user: dict[str, Any], + requested_username: str, + strategy: CollisionStrategy, + alias_suffix: str, +) -> str: + current_user_id = int(current_user["id"]) + existing = user_db.get_user(username=requested_username) + if existing is None or int(existing.get("id") or 0) == current_user_id: + return requested_username + + if strategy == "suffix": + return _next_suffix_username( + user_db, + requested_username, + exclude_user_id=current_user_id, + ) + if strategy == "alias": + return _next_suffix_username( + user_db, + f"{requested_username}{alias_suffix}", + exclude_user_id=current_user_id, + ) + + # `takeover` can select an existing row during creation, but once an + # identity is already matched it must never replace a different username + # owner. Preserve the matched row's current collision-free name instead. + return str(current_user["username"]) + + def upsert_external_user( user_db: UserDB, *, @@ -197,6 +239,7 @@ def upsert_external_user( subject: str | None = None, allow_email_link: bool = False, sync_role: bool = True, + sync_username: bool = False, allow_create: bool = True, collision_strategy: CollisionStrategy = "takeover", alias_suffix: str | None = None, @@ -229,10 +272,26 @@ def upsert_external_user( subject=subject, allow_email_link=allow_email_link, ) + resolved_alias_suffix = alias_suffix or f"__{auth_source}" + update_username: str | object = UNSET + if ( + matched is not None + and sync_username + and normalize_auth_source(matched.get("auth_source"), matched.get("oidc_subject")) + == auth_source + ): + update_username = _resolve_update_username( + user_db, + current_user=matched, + requested_username=normalized_username, + strategy=collision_strategy, + alias_suffix=resolved_alias_suffix, + ) updates = _build_updates( auth_source=auth_source, role=normalized_role, sync_role=sync_role, + username=update_username, email=normalized_email if email is not UNSET else UNSET, display_name=normalized_display_name if display_name is not UNSET else UNSET, subject_field=subject_field, @@ -261,7 +320,6 @@ def upsert_external_user( ) return None, "not_found" - resolved_alias_suffix = alias_suffix or f"__{auth_source}" create_username, takeover_target, create_reason = _resolve_create_username( user_db, auth_source=auth_source, diff --git a/shelfmark/core/user_db.py b/shelfmark/core/user_db.py index 954e84cb..d431b6b6 100644 --- a/shelfmark/core/user_db.py +++ b/shelfmark/core/user_db.py @@ -344,6 +344,7 @@ class UserDB: _ALLOWED_UPDATE_COLUMNS: ClassVar[frozenset[str]] = frozenset( { + "username", "email", "display_name", "password_hash", @@ -353,6 +354,7 @@ class UserDB: } ) _USER_UPDATE_STATEMENTS: ClassVar[dict[str, str]] = { + "username": "UPDATE users SET username = ? WHERE id = ?", "email": "UPDATE users SET email = ? WHERE id = ?", "display_name": "UPDATE users SET display_name = ? WHERE id = ?", "password_hash": "UPDATE users SET password_hash = ? WHERE id = ?", diff --git a/tests/core/test_cwa_user_sync.py b/tests/core/test_cwa_user_sync.py index 1e7c24c9..e867ffb8 100644 --- a/tests/core/test_cwa_user_sync.py +++ b/tests/core/test_cwa_user_sync.py @@ -95,6 +95,64 @@ def test_upsert_updates_existing_cwa_user_by_username_before_email(user_db): assert user["role"] == "admin" +def test_upsert_renames_existing_cwa_user_matched_by_email(user_db): + cwa_user = user_db.create_user( + username="old_reader", + email="reader@example.com", + role="user", + auth_source="cwa", + ) + + user, action = upsert_cwa_user( + user_db, + cwa_username="renamed_reader", + cwa_email="reader@example.com", + role="user", + ) + + assert action == "updated" + assert user["id"] == cwa_user["id"] + assert user["username"] == "renamed_reader" + assert user_db.get_user(username="old_reader") is None + + +def test_upsert_uses_stable_alias_when_renamed_cwa_username_is_taken(user_db): + cwa_user = user_db.create_user( + username="old_reader", + email="reader@example.com", + role="user", + auth_source="cwa", + ) + local_user = user_db.create_user( + username="renamed_reader", + email="local@example.com", + role="user", + auth_source="builtin", + ) + + first, first_action = upsert_cwa_user( + user_db, + cwa_username="renamed_reader", + cwa_email="reader@example.com", + role="admin", + ) + second, second_action = upsert_cwa_user( + user_db, + cwa_username="renamed_reader", + cwa_email="reader@example.com", + role="admin", + ) + + assert first_action == second_action == "updated" + assert first["id"] == second["id"] == cwa_user["id"] + assert first["username"] == second["username"] == "renamed_reader__cwa" + assert first["role"] == second["role"] == "admin" + local_after = user_db.get_user(user_id=local_user["id"]) + assert local_after is not None + assert local_after["username"] == "renamed_reader" + assert local_after["email"] == "local@example.com" + + def test_sync_prunes_cwa_users_missing_from_source(user_db): active_cwa = user_db.create_user( username="active_cwa", diff --git a/tests/core/test_user_db.py b/tests/core/test_user_db.py index 564aab52..1090efbe 100644 --- a/tests/core/test_user_db.py +++ b/tests/core/test_user_db.py @@ -568,11 +568,13 @@ class TestUserCRUD: user = user_db.create_user(username="john", role="user") user_db.update_user( user["id"], + username="jane", role="admin", email="new@example.com", auth_source="proxy", ) updated = user_db.get_user(user_id=user["id"]) + assert updated["username"] == "jane" assert updated["role"] == "admin" assert updated["email"] == "new@example.com" assert updated["auth_source"] == "proxy"