diff --git a/src/backend/core/tasks/user_reconciliation.py b/src/backend/core/tasks/user_reconciliation.py new file mode 100644 index 00000000..586163a4 --- /dev/null +++ b/src/backend/core/tasks/user_reconciliation.py @@ -0,0 +1,141 @@ +"""Processing tasks for user reconciliation CSV imports.""" + +import csv +import logging +import traceback +import uuid + +from django.core.exceptions import ValidationError +from django.core.validators import validate_email +from django.db import IntegrityError + +from botocore.exceptions import ClientError + +from core.models import UserReconciliation, UserReconciliationCsvImport + +from drive.celery_app import app + +logger = logging.getLogger(__name__) + + +def _process_row(row, job, counters): + """Process a single row from the CSV file.""" + + source_unique_id = row["id"].strip() + + # Skip entries if they already exist with this source_unique_id + if UserReconciliation.objects.filter(source_unique_id=source_unique_id).exists(): + counters["already_processed_source_ids"] += 1 + return counters + + active_email_checked = row.get("active_email_checked", "0") == "1" + inactive_email_checked = row.get("inactive_email_checked", "0") == "1" + + active_email = row["active_email"] + inactive_emails = row["inactive_email"].split("|") + try: + validate_email(active_email) + except ValidationError: + job.send_reconciliation_error_email( + recipient_email=inactive_emails[0], other_email=active_email + ) + job.logs += f"Invalid active email address on row {source_unique_id}." + counters["rows_with_errors"] += 1 + return counters + + for inactive_email in inactive_emails: + try: + validate_email(inactive_email) + except (ValidationError, ValueError): + job.send_reconciliation_error_email( + recipient_email=active_email, other_email=inactive_email + ) + job.logs += f"Invalid inactive email address on row {source_unique_id}.\n" + counters["rows_with_errors"] += 1 + continue + + if inactive_email == active_email: + job.send_reconciliation_error_email( + recipient_email=active_email, other_email=inactive_email + ) + job.logs += ( + f"Error on row {source_unique_id}: " + f"{active_email} set as both active and inactive email.\n" + ) + counters["rows_with_errors"] += 1 + continue + + UserReconciliation.objects.create( + active_email=active_email, + inactive_email=inactive_email, + active_email_checked=active_email_checked, + inactive_email_checked=inactive_email_checked, + active_email_confirmation_id=uuid.uuid4(), + inactive_email_confirmation_id=uuid.uuid4(), + source_unique_id=source_unique_id, + status="pending", + ) + counters["rec_entries_created"] += 1 + + return counters + + +@app.task +def user_reconciliation_csv_import_job(job_id): + """Process a UserReconciliationCsvImport job. + + Creates UserReconciliation entries from the CSV file. + + Does some sanity checks on the data: + - active_email and inactive_email must be valid email addresses + - active_email and inactive_email cannot be the same + + Rows with errors are logged in the job logs and skipped, but do not cause + the entire job to fail or prevent the next rows from being processed. + """ + try: + job = UserReconciliationCsvImport.objects.get(id=job_id) + except UserReconciliationCsvImport.DoesNotExist: + logger.warning("CSV import job %s no longer exists; skipping.", job_id) + return + + job.status = "running" + job.save() + + counters = { + "rec_entries_created": 0, + "rows_with_errors": 0, + "already_processed_source_ids": 0, + } + + try: + with job.file.open(mode="r") as file: + reader = csv.DictReader(file) + + if not {"active_email", "inactive_email", "id"}.issubset(reader.fieldnames or []): + raise KeyError("CSV is missing mandatory columns: active_email, inactive_email, id") + + for row in reader: + counters = _process_row(row, job, counters) + + job.status = "done" + job.logs += ( + f"Import completed successfully. {reader.line_num} rows processed." + f" {counters['rec_entries_created']} reconciliation entries created." + f" {counters['already_processed_source_ids']} rows were already processed." + f" {counters['rows_with_errors']} rows had errors." + ) + except ( + csv.Error, + KeyError, + ValidationError, + ValueError, + IntegrityError, + OSError, + ClientError, + ) as exception: + # Catch expected I/O/CSV/model errors and record traceback in logs for debugging + job.status = "error" + job.logs += f"{exception!s}\n{traceback.format_exc()}" + finally: + job.save() diff --git a/src/backend/core/tests/data/example_reconciliation_basic.csv b/src/backend/core/tests/data/example_reconciliation_basic.csv new file mode 100644 index 00000000..a457b6da --- /dev/null +++ b/src/backend/core/tests/data/example_reconciliation_basic.csv @@ -0,0 +1,3 @@ +active_email,inactive_email,active_email_checked,inactive_email_checked,id +active1@example.com,inactive1@example.com,1,1,1 +active2@example.com,inactive2@example.com,1,1,2 diff --git a/src/backend/core/tests/data/example_reconciliation_error.csv b/src/backend/core/tests/data/example_reconciliation_error.csv new file mode 100644 index 00000000..773563e3 --- /dev/null +++ b/src/backend/core/tests/data/example_reconciliation_error.csv @@ -0,0 +1,2 @@ +active_email,inactive_email,active_email_checked,inactive_email_checked,id +active1@example.com,,1,1,40 diff --git a/src/backend/core/tests/data/example_reconciliation_grist_form.csv b/src/backend/core/tests/data/example_reconciliation_grist_form.csv new file mode 100644 index 00000000..4796e983 --- /dev/null +++ b/src/backend/core/tests/data/example_reconciliation_grist_form.csv @@ -0,0 +1,2 @@ +active_email,inactive_email,active_email_checked,inactive_email_checked,id +active1@example.com,inactive1@example.com|inactive2@example.com,1,1,10 diff --git a/src/backend/core/tests/data/example_reconciliation_missing_column.csv b/src/backend/core/tests/data/example_reconciliation_missing_column.csv new file mode 100644 index 00000000..a0060a0b --- /dev/null +++ b/src/backend/core/tests/data/example_reconciliation_missing_column.csv @@ -0,0 +1,2 @@ +active_email,inactive_email +active1@example.com,inactive1@example.com diff --git a/src/backend/core/tests/tasks/test_user_reconciliation.py b/src/backend/core/tests/tasks/test_user_reconciliation.py new file mode 100644 index 00000000..29fc6618 --- /dev/null +++ b/src/backend/core/tests/tasks/test_user_reconciliation.py @@ -0,0 +1,120 @@ +"""Tests for the user reconciliation CSV import task.""" + +import logging +import uuid +from pathlib import Path + +from django.core import mail +from django.core.files.base import ContentFile + +import pytest + +from core import factories, models +from core.tasks.user_reconciliation import user_reconciliation_csv_import_job + +pytestmark = pytest.mark.django_db + +DATA_DIR = Path(__file__).parent.parent / "data" + + +def make_import(filename): + """Create a UserReconciliationCsvImport from a fixture CSV file.""" + with open(DATA_DIR / filename, "rb") as file: + csv_file = ContentFile(file.read(), name=filename) + return models.UserReconciliationCsvImport.objects.create(file=csv_file) + + +def test_user_reconciliation_csv_import_missing_job(caplog): + """A missing import job logs a warning and returns without raising.""" + with caplog.at_level(logging.WARNING): + user_reconciliation_csv_import_job(uuid.uuid4()) + + assert "no longer exists" in caplog.text + + +def test_user_reconciliation_csv_import_creates_entries(): + """A well-formed CSV creates one reconciliation entry per row.""" + for email in ["active1", "inactive1", "active2", "inactive2"]: + factories.UserFactory(email=f"{email}@example.com") + + csv_import = make_import("example_reconciliation_basic.csv") + user_reconciliation_csv_import_job(csv_import.id) + csv_import.refresh_from_db() + + assert csv_import.status == "done" + assert models.UserReconciliation.objects.count() == 2 + assert models.UserReconciliation.objects.filter(status="ready").count() == 2 + + +def test_user_reconciliation_csv_import_empty_file(): + """An empty CSV fails into the error status instead of hanging in running.""" + csv_import = models.UserReconciliationCsvImport.objects.create( + file=ContentFile(b"", name="empty.csv") + ) + user_reconciliation_csv_import_job(csv_import.id) + csv_import.refresh_from_db() + + assert csv_import.status == "error" + assert "missing mandatory columns" in csv_import.logs + + +def test_user_reconciliation_csv_import_missing_column(): + """A CSV missing a mandatory column fails into the error status.""" + csv_import = make_import("example_reconciliation_missing_column.csv") + user_reconciliation_csv_import_job(csv_import.id) + csv_import.refresh_from_db() + + assert csv_import.status == "error" + assert "missing mandatory columns" in csv_import.logs + assert models.UserReconciliation.objects.count() == 0 + + +def test_user_reconciliation_csv_import_invalid_email(): + """An invalid email is logged and triggers an error email, without failing the job.""" + factories.UserFactory(email="active1@example.com") + + csv_import = make_import("example_reconciliation_error.csv") + user_reconciliation_csv_import_job(csv_import.id) + csv_import.refresh_from_db() + + assert csv_import.status == "done" + assert "Invalid inactive email address on row 40" in csv_import.logs + assert models.UserReconciliation.objects.count() == 0 + + # pylint: disable-next=no-member + assert len(mail.outbox) == 1 + # pylint: disable-next=no-member + assert mail.outbox[0].to == ["active1@example.com"] + + +def test_user_reconciliation_csv_import_multiple_inactive_emails(): + """A single row may list several inactive emails separated by a pipe.""" + for email in ["active1", "inactive1", "inactive2"]: + factories.UserFactory(email=f"{email}@example.com") + + csv_import = make_import("example_reconciliation_grist_form.csv") + user_reconciliation_csv_import_job(csv_import.id) + csv_import.refresh_from_db() + + assert csv_import.status == "done" + assert models.UserReconciliation.objects.count() == 2 + assert set(models.UserReconciliation.objects.values_list("inactive_email", flat=True)) == { + "inactive1@example.com", + "inactive2@example.com", + } + + +def test_user_reconciliation_csv_import_is_idempotent(): + """Re-importing the same source ids does not create duplicate entries.""" + for email in ["active1", "inactive1", "active2", "inactive2"]: + factories.UserFactory(email=f"{email}@example.com") + + csv_import = make_import("example_reconciliation_basic.csv") + user_reconciliation_csv_import_job(csv_import.id) + + second_import = make_import("example_reconciliation_basic.csv") + user_reconciliation_csv_import_job(second_import.id) + second_import.refresh_from_db() + + assert models.UserReconciliation.objects.count() == 2 + assert "already processed" in second_import.logs