Add bulk add functionality for disposable email domains (#23726)

* Add bulk add functionality for disposable email domains

- Implemented the `bulk_add_disposable_email_domains` task to handle the addition of these domains.
- Added a management command to facilitate bulk addition from a CSV file.
- Created unit tests to ensure the functionality works as expected, including handling of duplicate entries.

* Fix comments

* TMP: Respond to comments.

* Robustify the testimajigger.

* A great offering to the oracle of beautiful python.

* Strip whitespace from domain

* Fix broken test

Author: KevinMind

src/olympia/lib/settings_base.py

@@ -825,6 +825,7 @@ def get_db_config(environ_var, atomic_requests=True):
     },
     'olympia.users.tasks.sync_suppressed_emails_task': {'queue': 'cron'},
     'olympia.users.tasks.send_suppressed_email_confirmation': {'queue': 'devhub'},
+    'olympia.users.tasks.bulk_add_disposable_email_domains': {'queue': 'devhub'},
     # Reviewers.
     'olympia.reviewers.tasks.recalculate_post_review_weight': {'queue': 'reviewers'},
     # Admin.

src/olympia/users/management/commands/bulk_add_disposable_domains.py

@@ -0,0 +1,34 @@
+import csv
+from pathlib import Path
+
+from django.core.management.base import BaseCommand, CommandError
+
+from olympia.core import logger
+from olympia.users.tasks import bulk_add_disposable_email_domains
+
+
+logger = logger.getLogger('z.users')
+
+
+class Command(BaseCommand):
+    help = 'Bulk add disposable email domains'
+
+    def add_arguments(self, parser):
+        parser.add_argument('file', type=str)
+
+    def handle(self, *args, **options):
+        file_path = Path(options['file'])
+        if not file_path.exists():
+            raise CommandError(f'File {file_path} does not exist')
+
+        records = []
+        with file_path.open(newline='') as f:
+            reader = csv.reader(f)
+            next(reader, None)  # skip header row
+            for row in reader:
+                if len(row) >= 2:
+                    domain, provider = row[0], row[1]
+                    records.append((domain.strip(), provider.strip()))
+
+        result = bulk_add_disposable_email_domains.apply(args=[records])
+        logger.info(result)

src/olympia/users/tasks.py

@@ -4,6 +4,8 @@
 import urllib.parse
 
 from django.conf import settings
+from django.core.exceptions import ValidationError
+from django.db.utils import InterfaceError, OperationalError
 from django.urls import reverse
 from django.utils.translation import gettext
 
@@ -24,6 +26,7 @@
 
 from .models import (
     BannedUserContent,
+    DisposableEmailDomainRestriction,
     SuppressedEmail,
     SuppressedEmailVerification,
     UserProfile,
@@ -196,3 +199,53 @@ def send_suppressed_email_confirmation(suppressed_email_verification_id):
     )
 
     verification.save()
+
+
+@task(
+    autoretry_for=(OperationalError, InterfaceError), max_retries=5, retry_backoff=True
+)
+def bulk_add_disposable_email_domains(entries: list[tuple[str, str]], batch_size=1000):
+    if not isinstance(batch_size, int) or batch_size <= 0:
+        raise ValueError('batch_size must be a positive integer')
+
+    task_log.info(f'Adding {len(entries)} disposable email domains')
+
+    records = []
+    errors = []
+
+    for entry in entries:
+        [domain, provider] = entry
+        record = DisposableEmailDomainRestriction(
+            domain=domain,
+            reason=f'Disposable email domain of {provider}',
+        )
+
+        try:
+            record.full_clean()
+            records.append(record)
+        except ValidationError as e:
+            errors.append(e)
+
+    if not records:
+        task_log.info('No valid entries provided')
+        return
+
+    processed_domains = []
+
+    for i in range(0, len(records), batch_size):
+        batch = records[i : i + batch_size]
+        created_objects = DisposableEmailDomainRestriction.objects.bulk_create(
+            batch,
+            batch_size,
+            ignore_conflicts=True,
+        )
+        processed_domains.extend(created_objects)
+        task_log.info(
+            f'Successfully processed {len(created_objects)} '
+            f'of {len(batch)} domains in this batch'
+        )
+
+    task_log.info(
+        f'Processed {len(processed_domains)} domains: '
+        f'{[obj.domain for obj in processed_domains]}'
+    )

src/olympia/users/tests/test_commands.py

@@ -2,13 +2,15 @@
 import json
 import os
 import re
+import tempfile
 import uuid
 from ipaddress import IPv4Address
 from unittest.mock import ANY, patch
 
 from django.core.files.base import ContentFile
-from django.core.management import call_command
+from django.core.management import CommandError, call_command
 
+from celery.result import EagerResult
 from waffle.testutils import override_switch
 
 from olympia import amo
@@ -17,7 +19,11 @@
 from olympia.amo.tests import TestCase, addon_factory, user_factory
 from olympia.amo.utils import SafeStorage
 from olympia.users.management.commands.createsuperuser import Command as CreateSuperUser
-from olympia.users.models import UserProfile, UserRestrictionHistory
+from olympia.users.models import (
+    DisposableEmailDomainRestriction,
+    UserProfile,
+    UserRestrictionHistory,
+)
 
 
 @patch('olympia.users.management.commands.createsuperuser.input')
@@ -473,3 +479,143 @@ def test_migrate_twice(self):
         self.test_migrate()
         # Running the migration command again shouldn't do anything.
         call_command('migrate_user_photos')
+
+
+class TestBulkAddDisposableEmailDomains(TestCase):
+    def test_missing_file_raises_command_error(self):
+        """Test that the command raises CommandError if the file does not exist"""
+
+        for file, message in [
+            (None, 'Error: the following arguments are required: file'),
+            ('none', 'File none does not exist'),
+        ]:
+            with self.assertRaises(CommandError) as e:
+                args = ['bulk_add_disposable_domains']
+                if file:
+                    args.append(file)
+                call_command(*args)
+            assert message == e.exception.args[0]
+
+    def test_valid_file_triggers_bulk_add(self):
+        """Test that a valid CSV file triggers
+        the bulk_add_disposable_email_domains task
+        and creates DisposableDomainRestriction objects.
+        """
+
+        csv_content = (
+            'Domain,Provider\n'
+            'mailfast.pro,incognitomail.co\n'
+            'foo.com,mail.tm\n'
+            'mailpro.live,incognitomail.co\n'
+        )
+
+        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.csv') as tmp:
+            tmp.write(csv_content)
+            tmp.flush()
+            tmp_path = tmp.name
+
+            assert DisposableEmailDomainRestriction.objects.count() == 0
+
+            call_command('bulk_add_disposable_domains', tmp_path)
+
+        expected = [
+            ('mailfast.pro', 'incognitomail.co'),
+            ('foo.com', 'mail.tm'),
+            ('mailpro.live', 'incognitomail.co'),
+        ]
+        for domain, provider in expected:
+            assert DisposableEmailDomainRestriction.objects.filter(
+                domain=domain, reason=f'Disposable email domain of {provider}'
+            ).exists()
+
+        assert DisposableEmailDomainRestriction.objects.count() == len(expected)
+
+    def test_file_with_missing_columns(self):
+        """Test that rows with missing columns are ignored or handled gracefully."""
+        csv_content = (
+            'Domain,Provider\n'
+            'mailfast.pro,incognitomail.co\n'
+            ',mail.tm\n'  # Missing domain
+            'mailpro.live,incognitomail.co\n'
+        )
+
+        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.csv') as tmp:
+            tmp.write(csv_content)
+            tmp.flush()
+            tmp_path = tmp.name
+
+            assert DisposableEmailDomainRestriction.objects.count() == 0
+
+            call_command('bulk_add_disposable_domains', tmp_path)
+
+        expected = [
+            ('mailfast.pro', 'incognitomail.co'),
+            ('mailpro.live', 'incognitomail.co'),
+        ]
+        for domain, provider in expected:
+            assert DisposableEmailDomainRestriction.objects.filter(
+                domain=domain, reason=f'Disposable email domain of {provider}'
+            ).exists()
+
+        assert not DisposableEmailDomainRestriction.objects.filter(domain='').exists()
+        assert DisposableEmailDomainRestriction.objects.count() == len(expected)
+
+    def test_file_with_header_only(self):
+        """Test that a file with only a header row does not trigger any additions."""
+        csv_content = 'Domain,Provider\n'
+
+        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.csv') as tmp:
+            tmp.write(csv_content)
+            tmp.flush()
+            tmp_path = tmp.name
+
+            call_command('bulk_add_disposable_domains', tmp_path)
+            assert DisposableEmailDomainRestriction.objects.count() == 0
+
+    @patch('olympia.users.management.commands.bulk_add_disposable_domains.logger')
+    def test_bulk_add_result_is_printed(self, mock_logger):
+        """result of bulk_add_disposable_email_domains task logged."""
+
+        csv_content = (
+            'Domain,Provider\n'
+            'mailfast.pro,incognitomail.co\n'
+            'mailpro.live,incognitomail.co\n'
+        )
+
+        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.csv') as tmp:
+            tmp.write(csv_content)
+            tmp.flush()
+            tmp_path = tmp.name
+
+            call_command('bulk_add_disposable_domains', tmp_path)
+            result = mock_logger.info.call_args[0][0]
+            assert isinstance(result, EagerResult)
+            assert result.state == 'SUCCESS'
+
+    def test_trims_domains_of_whitespace(self):
+        """Test that domains with whitespace are trimmed."""
+        csv_content = (
+            'Domain,Provider\n'
+            ' mailfast.pro  ,incognitomail.co\n'
+            'foo.com,mail.tm\n'
+            'mailpro.live,incognitomail.co\n'
+        )
+
+        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.csv') as tmp:
+            tmp.write(csv_content)
+            tmp.flush()
+            tmp_path = tmp.name
+
+            assert DisposableEmailDomainRestriction.objects.count() == 0
+
+            call_command('bulk_add_disposable_domains', tmp_path)
+
+        expected = [
+            ('mailfast.pro', 'incognitomail.co'),
+            ('foo.com', 'mail.tm'),
+            ('mailpro.live', 'incognitomail.co'),
+        ]
+        for domain, provider in expected:
+            assert DisposableEmailDomainRestriction.objects.filter(
+                domain=domain, reason=f'Disposable email domain of {provider}'
+            ).exists()

src/olympia/users/tests/test_tasks.py

@@ -7,6 +7,7 @@
 
 from django.conf import settings
 from django.core import mail
+from django.db.utils import OperationalError
 from django.urls import reverse
 
 import pytest
@@ -21,10 +22,12 @@
 from olympia.amo.utils import SafeStorage
 from olympia.users.models import (
     BannedUserContent,
+    DisposableEmailDomainRestriction,
     SuppressedEmail,
     SuppressedEmailVerification,
 )
 from olympia.users.tasks import (
+    bulk_add_disposable_email_domains,
     delete_photo,
     resize_photo,
     send_suppressed_email_confirmation,
@@ -414,3 +417,56 @@ def test_retry_existing_verification(
             verification.reload().status
             == SuppressedEmailVerification.STATUS_CHOICES.Pending
         )
+
+
+class TestBulkAddDisposableEmailDomains(TestCase):
+    def setUp(self):
+        self.user_profile = user_factory()
+        self.entries = [
+            (f'test-{i}-{j}.com', f'provider-{i}')
+            for i in range(10)
+            for j in range(100)
+        ]
+        # Ensure that the task runs with 2 batches by default
+        self.batch_size = len(self.entries) // 2
+
+    def test_bulk_add_disposable_email_domains_success(self):
+        assert DisposableEmailDomainRestriction.objects.count() == 0
+
+        result = bulk_add_disposable_email_domains.apply(
+            args=[self.entries, self.batch_size]
+        )
+
+        assert result.status == 'SUCCESS'
+        assert DisposableEmailDomainRestriction.objects.count() == len(self.entries)
+
+    def test_bulk_add_disposable_email_domains_skips_duplicate_entries(self):
+        [domain, provider] = self.entries[0]
+        DisposableEmailDomainRestriction.objects.create(
+            domain=domain, reason=f'Disposable email domain of {provider}'
+        )
+        assert DisposableEmailDomainRestriction.objects.count() == 1
+
+        result = bulk_add_disposable_email_domains.apply(
+            args=[self.entries, self.batch_size]
+        )
+
+        assert result.status == 'SUCCESS'
+        assert DisposableEmailDomainRestriction.objects.count() == len(self.entries)
+
+    def test_zero_batch_size_raises_error(self):
+        with pytest.raises(ValueError):
+            bulk_add_disposable_email_domains.apply(args=[self.entries, 0])
+
+    def test_retries_on_db_timeout(self):
+        def always_raise_operational_error(*args, **kwargs):
+            raise OperationalError()
+
+        with mock.patch(
+            'olympia.users.tasks.DisposableEmailDomainRestriction.objects.bulk_create',
+            side_effect=always_raise_operational_error,
+        ):
+            with pytest.raises(Retry):
+                bulk_add_disposable_email_domains.apply(
+                    args=[self.entries, self.batch_size]
+                ).get()