diff --git a/src/backend/core/migrations/0023_userreconciliationcsvimport_userreconciliation.py b/src/backend/core/migrations/0023_userreconciliationcsvimport_userreconciliation.py new file mode 100644 index 00000000..da69fd17 --- /dev/null +++ b/src/backend/core/migrations/0023_userreconciliationcsvimport_userreconciliation.py @@ -0,0 +1,57 @@ +# Generated by Django 5.2.14 on 2026-05-21 12:04 + +import django.db.models.deletion +import uuid +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0022_remove_item_numchild_remove_item_numchild_folder_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='UserReconciliationCsvImport', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')), + ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')), + ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')), + ('file', models.FileField(upload_to='imports/', verbose_name='CSV file')), + ('status', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('done', 'Done'), ('error', 'Error')], default='pending', max_length=20)), + ('logs', models.TextField(blank=True)), + ], + options={ + 'verbose_name': 'user reconciliation CSV import', + 'verbose_name_plural': 'user reconciliation CSV imports', + 'db_table': 'drive_user_reconciliation_csv_import', + }, + ), + migrations.CreateModel( + name='UserReconciliation', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')), + ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')), + ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')), + ('active_email', models.EmailField(max_length=254, verbose_name='Active email address')), + ('inactive_email', models.EmailField(max_length=254, verbose_name='Email address to deactivate')), + ('active_email_checked', models.BooleanField(default=False)), + ('inactive_email_checked', models.BooleanField(default=False)), + ('active_email_confirmation_id', models.UUIDField(default=uuid.uuid4, editable=False, null=True, unique=True)), + ('inactive_email_confirmation_id', models.UUIDField(default=uuid.uuid4, editable=False, null=True, unique=True)), + ('source_unique_id', models.CharField(blank=True, max_length=100, null=True, verbose_name='Unique ID in the source file')), + ('status', models.CharField(choices=[('pending', 'Pending'), ('ready', 'Ready'), ('done', 'Done'), ('error', 'Error')], default='pending', max_length=20)), + ('logs', models.TextField(blank=True)), + ('active_user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='active_user', to=settings.AUTH_USER_MODEL)), + ('inactive_user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='inactive_user', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'verbose_name': 'user reconciliation', + 'verbose_name_plural': 'user reconciliations', + 'db_table': 'drive_user_reconciliation', + 'ordering': ['-created_at'], + }, + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 6d4b0713..2d9b55d1 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -368,6 +368,383 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin): return [] +class UserReconciliation(BaseModel): + """Model to run batch jobs to replace an active user by another one.""" + + active_email = models.EmailField(_("Active email address")) + inactive_email = models.EmailField(_("Email address to deactivate")) + active_email_checked = models.BooleanField(default=False) + inactive_email_checked = models.BooleanField(default=False) + active_user = models.ForeignKey( + User, + on_delete=models.CASCADE, + null=True, + blank=True, + related_name="active_user", + ) + inactive_user = models.ForeignKey( + User, + on_delete=models.CASCADE, + null=True, + blank=True, + related_name="inactive_user", + ) + active_email_confirmation_id = models.UUIDField( + default=uuid.uuid4, unique=True, editable=False, null=True + ) + inactive_email_confirmation_id = models.UUIDField( + default=uuid.uuid4, unique=True, editable=False, null=True + ) + source_unique_id = models.CharField( + max_length=100, + blank=True, + null=True, + verbose_name=_("Unique ID in the source file"), + ) + + status = models.CharField( + max_length=20, + choices=[ + ("pending", _("Pending")), + ("ready", _("Ready")), + ("done", _("Done")), + ("error", _("Error")), + ], + default="pending", + ) + logs = models.TextField(blank=True) + + class Meta: + db_table = "drive_user_reconciliation" + verbose_name = _("user reconciliation") + verbose_name_plural = _("user reconciliations") + ordering = ["-created_at"] + + def __str__(self): + return f"Reconciliation from {self.inactive_email} to {self.active_email}" + + def save(self, *args, **kwargs): + """ + For pending queries, identify the actual users and send validation emails. + """ + if self.status == "pending": + self.active_user = User.objects.filter(email=self.active_email).first() + self.inactive_user = User.objects.filter(email=self.inactive_email).first() + + if self.active_user and self.inactive_user: + if not self.active_email_checked: + self.send_reconciliation_confirm_email( + self.active_user, "active", self.active_email_confirmation_id + ) + if not self.inactive_email_checked: + self.send_reconciliation_confirm_email( + self.inactive_user, + "inactive", + self.inactive_email_confirmation_id, + ) + self.status = "ready" + else: + self.status = "error" + self.logs = "Error: Both active and inactive users need to exist." + + super().save(*args, **kwargs) + + @transaction.atomic + def process_reconciliation_request(self): + """ + Process the reconciliation request as a transaction. + + - Transfer item accesses from inactive to active user, updating roles as needed. + - Transfer item favorites from inactive to active user. + - Transfer link traces from inactive to active user. + - Reassign items created by the inactive user to the active user. + - Reassign invitations issued by the inactive user to the active user. + - Activate the active user and deactivate the inactive user. + - Update the reconciliation entry itself. + """ + + # Prepare the data to perform the reconciliation on + updated_accesses, removed_accesses = self.prepare_itemaccess_reconciliation() + updated_linktraces, removed_linktraces = self.prepare_linktrace_reconciliation() + updated_favorites, removed_favorites = self.prepare_itemfavorite_reconciliation() + updated_items = self.prepare_item_creator_reconciliation() + updated_invitations = self.prepare_invitation_issuer_reconciliation() + + self.active_user.is_active = True + self.inactive_user.is_active = False + + # Actually perform the bulk operations + ItemAccess.objects.bulk_update(updated_accesses, ["user", "role"]) + if removed_accesses: + ids_to_delete = [entry.id for entry in removed_accesses] + ItemAccess.objects.filter(id__in=ids_to_delete).delete() + # Bulk delete bypasses ItemAccess.delete(), so the nb_accesses cache + # of the affected items must be invalidated explicitly. + items_to_invalidate = {entry.item_id: entry.item for entry in removed_accesses} + for item in items_to_invalidate.values(): + item.invalidate_nb_accesses_cache() + + ItemFavorite.objects.bulk_update(updated_favorites, ["user"]) + if removed_favorites: + ids_to_delete = [entry.id for entry in removed_favorites] + ItemFavorite.objects.filter(id__in=ids_to_delete).delete() + + LinkTrace.objects.bulk_update(updated_linktraces, ["user"]) + if removed_linktraces: + ids_to_delete = [entry.id for entry in removed_linktraces] + LinkTrace.objects.filter(id__in=ids_to_delete).delete() + + Item.objects.bulk_update(updated_items, ["creator"]) + Invitation.objects.bulk_update(updated_invitations, ["issuer"]) + + User.objects.bulk_update([self.active_user, self.inactive_user], ["is_active"]) + + # Wrap up the reconciliation entry + self.logs += ( + f"Requested update for {len(updated_accesses)} ItemAccess items " + f"and deletion for {len(removed_accesses)} ItemAccess items.\n" + ) + self.status = "done" + self.save() + + self.send_reconciliation_done_email() + + def prepare_itemaccess_reconciliation(self): + """ + Prepare the reconciliation by transferring item accesses from the inactive user + to the active user. + """ + updated_accesses = [] + removed_accesses = [] + inactive_accesses = ItemAccess.objects.filter(user=self.inactive_user) + + # Check items where the active user already has access + inactive_accesses_items = inactive_accesses.values_list("item", flat=True) + existing_accesses = ItemAccess.objects.filter(user=self.active_user).filter( + item__in=inactive_accesses_items + ) + existing_roles_per_item = dict(existing_accesses.values_list("item", "role")) + + for entry in inactive_accesses: + if entry.item_id in existing_roles_per_item: + # Update role if needed + existing_role = existing_roles_per_item[entry.item_id] + max_role = RoleChoices.max(entry.role, existing_role) + if existing_role != max_role: + existing_access = existing_accesses.get(item=entry.item) + existing_access.role = max_role + updated_accesses.append(existing_access) + removed_accesses.append(entry) + else: + entry.user = self.active_user + updated_accesses.append(entry) + + return updated_accesses, removed_accesses + + def prepare_itemfavorite_reconciliation(self): + """ + Prepare the reconciliation by transferring item favorites from the inactive user + to the active user. + """ + updated_favorites = [] + removed_favorites = [] + + existing_favorites = ItemFavorite.objects.filter(user=self.active_user) + existing_favorite_item_ids = set(existing_favorites.values_list("item_id", flat=True)) + + inactive_favorites = ItemFavorite.objects.filter(user=self.inactive_user) + + for entry in inactive_favorites: + if entry.item_id in existing_favorite_item_ids: + removed_favorites.append(entry) + else: + entry.user = self.active_user + updated_favorites.append(entry) + + return updated_favorites, removed_favorites + + def prepare_linktrace_reconciliation(self): + """ + Prepare the reconciliation by transferring link traces from the inactive user + to the active user. + """ + updated_linktraces = [] + removed_linktraces = [] + + existing_linktraces = LinkTrace.objects.filter(user=self.active_user) + inactive_linktraces = LinkTrace.objects.filter(user=self.inactive_user) + + for entry in inactive_linktraces: + if existing_linktraces.filter(item=entry.item).exists(): + removed_linktraces.append(entry) + else: + entry.user = self.active_user + updated_linktraces.append(entry) + + return updated_linktraces, removed_linktraces + + def prepare_item_creator_reconciliation(self): + """ + Prepare the reconciliation by reassigning items created by the inactive user + to the active user. + """ + updated_items = [] + + inactive_items = Item.objects.filter(creator=self.inactive_user) + + for entry in inactive_items: + entry.creator = self.active_user + updated_items.append(entry) + + return updated_items + + def prepare_invitation_issuer_reconciliation(self): + """ + Prepare the reconciliation by reassigning invitations issued by the inactive user + to the active user. + """ + updated_invitations = [] + + inactive_invitations = Invitation.objects.filter(issuer=self.inactive_user) + + for entry in inactive_invitations: + entry.issuer = self.active_user + updated_invitations.append(entry) + + return updated_invitations + + def send_reconciliation_confirm_email(self, user, user_type, confirmation_id, language=None): + """Method allowing to send confirmation email for reconciliation requests.""" + language = language or get_language() + domain = settings.EMAIL_URL_APP or Site.objects.get_current().domain + + message = _( + """You have requested a reconciliation of your user accounts on Drive. + To confirm that you are the one who initiated the request + and that this email belongs to you:""" + ) + + with override(language): + subject = _("Confirm by clicking the link to start the reconciliation") + context = { + "title": subject, + "message": message, + "link": f"{domain}/user-reconciliations/{user_type}/{confirmation_id}/", + "link_label": str(_("Click here")), + "button_label": str(_("Confirm")), + } + + user.send_email(subject, context, language) + + def send_reconciliation_done_email(self, language=None): + """Method allowing to send done email for reconciliation requests.""" + language = language or get_language() + domain = settings.EMAIL_URL_APP or Site.objects.get_current().domain + + message = _( + """Your reconciliation request has been processed. + New documents are likely associated with your account:""" + ) + + with override(language): + subject = _("Your accounts have been merged") + context = { + "title": subject, + "message": message, + "link": f"{domain}/", + "link_label": str(_("Click here to see")), + "button_label": str(_("See my documents")), + } + + self.active_user.send_email(subject, context, language) + + +class UserReconciliationCsvImport(BaseModel): + """Model to import reconciliation requests from an external source.""" + + file = models.FileField(upload_to="imports/", verbose_name=_("CSV file")) + status = models.CharField( + max_length=20, + choices=[ + ("pending", _("Pending")), + ("running", _("Running")), + ("done", _("Done")), + ("error", _("Error")), + ], + default="pending", + ) + logs = models.TextField(blank=True) + + class Meta: + db_table = "drive_user_reconciliation_csv_import" + verbose_name = _("user reconciliation CSV import") + verbose_name_plural = _("user reconciliation CSV imports") + + def __str__(self): + return f"User reconciliation CSV import {self.id}" + + def send_email(self, subject, emails, context=None, language=None): + """Generate and send email to the user from a template.""" + + if not settings.EMAIL_HOST: + logger.debug("EMAIL_HOST host is not set, skipping email sending") + return + + context = context or {} + domain = settings.EMAIL_URL_APP or Site.objects.get_current().domain + language = language or get_language() + context.update( + { + "brandname": settings.EMAIL_BRAND_NAME, + "domain": domain, + "logo_img": settings.EMAIL_LOGO_IMG, + } + ) + + with override(language): + msg_html = render_to_string("mail/html/reconciliation.html", context) + msg_plain = render_to_string("mail/text/reconciliation.txt", context) + subject = str(subject) # Force translation + + try: + send_mail( + subject.capitalize(), + msg_plain, + settings.EMAIL_FROM, + emails, + html_message=msg_html, + fail_silently=False, + ) + except smtplib.SMTPException as exception: + logger.error("email to %s was not sent: %s", emails, exception) + + def send_reconciliation_error_email(self, recipient_email, other_email, language=None): + """Method allowing to send email for reconciliation requests with errors.""" + language = language or get_language() + + emails = [recipient_email] + + message = _( + """Your request for reconciliation was unsuccessful. + Reconciliation failed for the following email addresses: + {recipient_email}, {other_email}. + Please check for typos. + You can submit another request with the valid email addresses.""" + ).format(recipient_email=recipient_email, other_email=other_email) + + with override(language): + subject = _("Reconciliation of your Drive accounts not completed") + context = { + "title": subject, + "message": message, + "link": settings.USER_RECONCILIATION_FORM_URL, + "link_label": str(_("Click here")), + "button_label": str(_("Make a new request")), + } + + self.send_email(subject, emails, context, language) + + class AnnotateUserRoleQuerySetMixin: """Mixin to use in a QuerySet to add user_roles annotation.""" diff --git a/src/backend/core/tests/test_models_user_reconciliation.py b/src/backend/core/tests/test_models_user_reconciliation.py new file mode 100644 index 00000000..0cb80a07 --- /dev/null +++ b/src/backend/core/tests/test_models_user_reconciliation.py @@ -0,0 +1,200 @@ +""" +Unit tests for the UserReconciliation model. +""" + +from django.core import mail + +import pytest + +from core import factories, models + +pytestmark = pytest.mark.django_db + + +def create_reconciliation(active_user, inactive_user, **kwargs): + """Create a ready-to-process reconciliation between two users.""" + return models.UserReconciliation.objects.create( + active_email=active_user.email, + inactive_email=inactive_user.email, + active_email_checked=True, + inactive_email_checked=True, + **kwargs, + ) + + +def test_models_user_reconciliation_save_resolves_users_and_sends_emails(): + """Saving a pending reconciliation resolves users and sends confirmation emails.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + + reconciliation = models.UserReconciliation.objects.create( + active_email=active.email, inactive_email=inactive.email + ) + + assert reconciliation.status == "ready" + assert reconciliation.active_user == active + assert reconciliation.inactive_user == inactive + + # pylint: disable-next=no-member + assert len(mail.outbox) == 2 + # pylint: disable-next=no-member + assert {mail.outbox[0].to[0], mail.outbox[1].to[0]} == { + "active@example.com", + "inactive@example.com", + } + + +def test_models_user_reconciliation_save_missing_user_sets_error(): + """Saving fails into error status when one of the users does not exist.""" + active = factories.UserFactory(email="active@example.com") + + reconciliation = models.UserReconciliation.objects.create( + active_email=active.email, inactive_email="ghost@example.com" + ) + + assert reconciliation.status == "error" + assert "Both active and inactive users need to exist" in reconciliation.logs + # pylint: disable-next=no-member + assert len(mail.outbox) == 0 + + +def test_models_user_reconciliation_save_checked_emails_skip_confirmation(): + """No confirmation email is sent when both emails are already checked.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + + reconciliation = create_reconciliation(active, inactive) + + assert reconciliation.status == "ready" + # pylint: disable-next=no-member + assert len(mail.outbox) == 0 + + +def test_models_user_reconciliation_transfers_item_access(): + """Accesses of the inactive user are transferred to the active user.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + item = factories.ItemFactory() + access = factories.UserItemAccessFactory( + item=item, user=inactive, role=models.RoleChoices.EDITOR + ) + + create_reconciliation(active, inactive).process_reconciliation_request() + + access.refresh_from_db() + assert access.user == active + assert access.role == models.RoleChoices.EDITOR + + +def test_models_user_reconciliation_merges_item_access_role_on_conflict(): + """When both users have access, the higher role is kept and the duplicate removed.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + item = factories.ItemFactory() + active_access = factories.UserItemAccessFactory( + item=item, user=active, role=models.RoleChoices.READER + ) + inactive_access = factories.UserItemAccessFactory( + item=item, user=inactive, role=models.RoleChoices.ADMIN + ) + + create_reconciliation(active, inactive).process_reconciliation_request() + + active_access.refresh_from_db() + assert active_access.role == models.RoleChoices.ADMIN + assert not models.ItemAccess.objects.filter(id=inactive_access.id).exists() + + +def test_models_user_reconciliation_transfers_favorites_and_dedup(): + """Favorites are transferred, duplicates on the active side are removed.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + shared_item = factories.ItemFactory() + inactive_only_item = factories.ItemFactory() + models.ItemFavorite.objects.create(user=active, item=shared_item) + duplicate = models.ItemFavorite.objects.create(user=inactive, item=shared_item) + transferred = models.ItemFavorite.objects.create(user=inactive, item=inactive_only_item) + + create_reconciliation(active, inactive).process_reconciliation_request() + + assert not models.ItemFavorite.objects.filter(id=duplicate.id).exists() + transferred.refresh_from_db() + assert transferred.user == active + + +def test_models_user_reconciliation_transfers_link_traces_and_dedup(): + """Link traces are transferred, duplicates on the active side are removed.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + shared_item = factories.ItemFactory() + inactive_only_item = factories.ItemFactory() + models.LinkTrace.objects.create(user=active, item=shared_item) + duplicate = models.LinkTrace.objects.create(user=inactive, item=shared_item) + transferred = models.LinkTrace.objects.create(user=inactive, item=inactive_only_item) + + create_reconciliation(active, inactive).process_reconciliation_request() + + assert not models.LinkTrace.objects.filter(id=duplicate.id).exists() + transferred.refresh_from_db() + assert transferred.user == active + + +def test_models_user_reconciliation_invalidates_nb_accesses_cache(): + """Removing a duplicate access must invalidate the item's nb_accesses cache.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + item = factories.ItemFactory() + factories.UserItemAccessFactory(item=item, user=active, role=models.RoleChoices.READER) + factories.UserItemAccessFactory(item=item, user=inactive, role=models.RoleChoices.ADMIN) + + # Prime the cache with the current count + assert models.Item.objects.get(pk=item.pk).nb_accesses == 2 + + create_reconciliation(active, inactive).process_reconciliation_request() + + # The inactive duplicate is removed: the cache must reflect a single access + assert models.Item.objects.get(pk=item.pk).nb_accesses == 1 + + +def test_models_user_reconciliation_reassigns_item_creator(): + """Items created by the inactive user are reassigned to the active user.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + item = factories.ItemFactory(creator=inactive) + + create_reconciliation(active, inactive).process_reconciliation_request() + + item.refresh_from_db() + assert item.creator == active + + +def test_models_user_reconciliation_reassigns_invitation_issuer(): + """Invitations issued by the inactive user are reassigned to the active user.""" + active = factories.UserFactory(email="active@example.com") + inactive = factories.UserFactory(email="inactive@example.com") + invitation = factories.InvitationFactory(issuer=inactive) + + create_reconciliation(active, inactive).process_reconciliation_request() + + invitation.refresh_from_db() + assert invitation.issuer == active + + +def test_models_user_reconciliation_toggles_is_active_and_sends_done_email(): + """Processing activates the active user, deactivates the inactive one, and notifies.""" + active = factories.UserFactory(email="active@example.com", is_active=False) + inactive = factories.UserFactory(email="inactive@example.com", is_active=True) + + reconciliation = create_reconciliation(active, inactive) + reconciliation.process_reconciliation_request() + + active.refresh_from_db() + inactive.refresh_from_db() + assert active.is_active is True + assert inactive.is_active is False + assert reconciliation.status == "done" + + # pylint: disable-next=no-member + assert len(mail.outbox) == 1 + # pylint: disable-next=no-member + assert mail.outbox[0].to == ["active@example.com"]